一个HBase数据迁移到Mongodb需求，架构师说“你怎么有那么多意外”

// 查询 hbasepublic List listVoteRecord(VoteRecordable begin, VoteRecordable end) {        byte[] startRow = begin.getRowKey();        byte[] endRow = end.getRowKey();        Scan scan = new Scan(startRow, endRow);        return this.hbaseTemplate.find(begin.getTableName(), scan, this.getRowMapper(begin.getClass()));    }

// mongodb 入库public void batchInsertTotalActivityVoteRecordDoc(List totalActivityVoteRecordDocs) {        BulkOperations operations = mongoTemplate.bulkOps(BulkOperations.BulkMode.ORDERED, TotalActivityVoteRecordDoc.class);        List> upsertList = new ArrayList<>(totalActivityVoteRecordDocs.size());        totalActivityVoteRecordDocs.forEach(data -> {            Query query = new Query(Criteria.where("activityId").is(data.getActivityId()));            Update update = new Update();            update.set("voteCount", data.getVoteCount());            Pair upsertPair = Pair.of(query, update);            upsertList.add(upsertPair);        });
        operations.upsert(upsertList);        operations.execute();    }

@ResponseBody    public void insertHourlyVoteRecordDoc2(@RequestParam(value = "voteItemId") String voteItemId,                                           @RequestParam(value = "beginDate") String beginDate,                                           @RequestParam(value = "endDate") String endDate){        VoteRecordable begin = new HourlyVoteRecord();        begin.setTraceId(voteItemId);        begin.setDate(beginDate);
        VoteRecordable end = new HourlyVoteRecord();        end.setTraceId(voteItemId);        end.setDate(endDate);
        List voteRecordables = voteItemStatService.listVoteRecords(begin, end);        System.out.println(voteRecordables.size() + "voteRecordables大小是多少 insertHourlyVoteRecordDoc");        if (CollectionUtils.isEmpty(voteRecordables)) {            return;        }        List hourlyVoteRecordDocs = new ArrayList<>(voteRecordables.size());        for (VoteRecordable voteRecordable1 : voteRecordables) {            HourlyVoteRecord hourlyVoteRecord = (HourlyVoteRecord) voteRecordable1;            HourlyVoteRecordDoc hourlyVoteRecordDoc = new HourlyVoteRecordDoc();            hourlyVoteRecordDoc.setVoteItemId(new ObjectId(hourlyVoteRecord.getVoteItemId()));            hourlyVoteRecordDoc.setVoteCount(hourlyVoteRecord.getVoteCount());            hourlyVoteRecordDoc.setHour(hourlyVoteRecord.getHour());
            hourlyVoteRecordDocs.add(hourlyVoteRecordDoc);        }        voteRecordService.batchInsertHourlyVoteRecordDoc(hourlyVoteRecordDocs);
    }

"""迁移DailyVoteRecordDocHourlyVoteRecordHourlyActivtiyVoteRecord"""
import threadingimport timefrom datetime import datetime, timedelta
from core_service import vote_servicefrom scripts.biz.vote.migrate_user_vote_records import MigrateLimitException

class MyThread(threading.Thread):    def __init__(self, thread_id, name, archive_activities):        threading.Thread.__init__(self)        self.threadID = thread_id        self.name = name        self.archive_activities = archive_activities
    def run(self):        print("Starting " + self.name)        main(self.archive_activities)        print("exiting " + self.name)

def main(archive_activities):    """    迁移投票记录    :param archive_activities:    :return:    """
    final_date_end_timestamp = 1595433600
    for activity in archive_activities:        date_start = time.strftime('%Y-%m-%d', time.localtime(int(activity['dateStart'] / 1000)))        hour_start = time.strftime('%Y-%m-%d-%H', time.localtime(int(activity['dateStart'] / 1000)))
        date_end_timestamp = int(activity['dateEnd'] / 1000 + 24 * 60 * 60)
        if final_date_end_timestamp < date_end_timestamp:            date_end_timestamp = final_date_end_timestamp
        date_end = time.strftime('%Y-%m-%d', time.localtime(date_end_timestamp))        hour_end = time.strftime('%Y-%m-%d-%H', time.localtime(date_end_timestamp))        ## 迁移 hourly_activity_vote_records        vote_service.batch_insert_hourly_activity_vote_records(activity.get('_id'), hour_start, hour_end)        print(activity)
        vote_items = vote_service.list_vote_items(str(activity['_id']))        try:            for vote_item in vote_items:                # print("user_vote_record: activityId voteItemId score", activity.get('_id'), vote_item.get('_id'))                # print(vote_item)                if vote_item['score'] != 0:                    ## 迁移 daily_vote_records                    vote_service.batch_insert_daily_vote_records(vote_item.get('_id'), date_start, date_end)                    ## 迁移 hourly_vote_records                    print("开始结束时间时间", hour_start, hour_end)                    print("开始结束日期", date_start, date_end)                    vote_service.batch_insert_hourly_vote_records(vote_item.get('_id'), hour_start, hour_end)
        except MigrateLimitException:            print("migrate limit error")


if __name__ == "__main__":    // 调用Java代码接口，查询到所有的活动    archive_activities_local = vote_service.archive_activities_v3()    # 开启多线程    i = 1    available_activities = []    threads = []
    for index, local_activity in enumerate(archive_activities_local):        # 每循环700次开一个线程        if index == i * 700:            thread = MyThread(i, "Thread-" + str(i), available_activities)            thread.start()            threads.append(thread)            i += 1            available_activities = []        available_activities.append(local_activity)
    if available_activities:        thread = MyThread(i, "Thread-" + str(i), available_activities)        thread.start()        threads.append(thread)
    for t in threads:        t.join()

请求如果失败了重试三次@retry(3)def batch_insert_hourly_activity_vote_records(activityId, beginDate, endDate):    """    批量插入活动小时投票记录    :param activityId:    :param beginDate:    :param endDate:    :return:    """    url = "http://xxx/vote/api/internal/dbMigration/insertHourlyActivityVoteRecordDoc?" \          "activityId={activityId}&beginDate={beginDate}&endDate={endDate}".format(activityId=activityId, beginDate=beginDate, endDate=endDate)
    r = requests.session().get(url)