记一次数据批量插入实践

news2024/11/26 4:30:08

背景：

2023月7月份入职新公司，初来乍到还没参入到具体的项目中，技术负责人安排写一个批量处理数据的服务，于是便有了以下文章。

数据流程大概是这样，从clickhouse表中获取数据，并从elasticserach中根据业务序号增加若干属性，批量保存到mysql中，最后再更新clickhouse中数据状态为已生成。

本文以10000条数据示例（本机环境 docker(mysql+clickhouse)、es测试服务器环境）。

思考：

很简单的处理流程，刚参加工作的时候这样的服务写过很多次（jdbc Class.forName Connection PreparedStatment Resulset一些列操作），顺手拈来的事情。

技术框架：springboot+mybatis plus+spring data elasticserach + dynamic datasource
实现思路：

orm框架选择mybatis处理mysql和clickhouse，使用mybatis plus封装好的saveOrUpdateBatch方法省得自己写sql，
使用spring data elasticserach封装好的方法进行es查询；
要同时操作clickhouse和mysql并且需要支持事务，选择dynamic datasource；

和相关同事沟通每天的数据量大概在5-6K(这里解释下数据量为什么自己不去看，内网环境需要vpn，刚入职暂未分配账号)，于是开始了编码工作。

过程实践：

定义方法：

1、生成保存到mysql的数据；

1.1、clickhouse批量查询数据；

1.2、查询数据结果根据业务编号从es中查询数据信息；

1.3、2个结果合并属性并生成mysql入库对象集合。

部分代码：

public List<QcBaseDataInfo> generate(String day, int limit) {
        /**
         * 1、从clickhouse取昨日数据
         * 2、从es中取数据(根据序列号)
         * 3、封装智能质检数据
         */
        Assert.notNull(day, "业务数据日期不允许为空");
        List<ChWicDataInfo> chWicDataInfos = queryFormCh(day, limit);
        List<QcBaseDataInfo> qcBaseDataInfoList = getPropertiesFromEs(chWicDataInfos);
        return qcBaseDataInfoList;
    }

@Override
public List<ChWicDataInfo> queryFormCh(String day, int limit) {
        Assert.notNull(day, "业务数据日期不允许为空");
        List<ChWicDataInfo> queryResult = chWicDataInfoService.list(new QueryWrapper<ChWicDataInfo>().lambda()
                .eq(ChWicDataInfo::getAcctDay, day)
                .eq(ChWicDataInfo::getIsDown, STATUS_ASR_DEAL_FINISHED)
                .eq(ChWicDataInfo::getHadToDataLake, false)
                .last(" limit " + limit));
        return queryResult;
    }

@Override
public List<QcBaseDataInfo> getPropertiesFromEs(List<ChWicDataInfo> chWicDataInfos) {
        List<String> serialNumbers = new ArrayList<>();
        chWicDataInfos.forEach(q -> {
            serialNumbers.add(q.getId());
        });
        log.info("start query serialNumbers：{} from es", serialNumbers);
        List<EsWicDataInfo> esWicDataInfos = esWicDataRepository.findBySerialNumberIn(serialNumbers);
        log.info("query from es end...");
        List<QcBaseDataInfo> qcBaseDataInfoList = construct(chWicDataInfos, esWicDataInfos);
        return qcBaseDataInfoList;
    }

2、数据批量保存/更新到mysql中；

部分代码：

@Override
 public boolean saveOrUpdateBatch(List<QcBaseDataInfo> data) {
        boolean result = qcBaseDataInfoService.saveOrUpdateBatch(data, BATCH_SIZE);
        return result;
 }

3、clickhouse中更新数据状态为已更新；

部分代码：