SpringBoot 使用 MyBatis 和 BatchExecutor 实现百万级数据异步批量插入 Hive

使用 SpringBoot + MyBatis + BatchExecutor 实现百万级数据异步批量插入 Hive

本教程将演示如何使用 SpringBoot、MyBatis 和 BatchExecutor 实现百万级数据异步批量插入 Hive。

1. 数据源配置

首先，在 application.yaml 中增加数据源配置，这里以 MySQL 为例：

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/testdb?useSSL=false&serverTimezone=UTC
    username: root
    password: '123456'
    driver-class-name: com.mysql.jdbc.Driver

2. Service 代码

在 Service 中使用 BatchExecutor 来实现批量插入、多线程、异步的方式实现百万级数据插入 Hive：

@Service
public class DataInsertService {
    @Autowired
    private BatchExecutor batchExecutor;

    @Autowired
    private DataSource dataSource;

    @Async
    public void insertData(List<Data> dataList) {
        String sql = "INSERT INTO hive_table (id, name, age) VALUES (?, ?, ?)";
        batchExecutor.executeBatch(dataSource, sql, dataList, 1000, new ParameterizedSetter<Data>() {
            @Override
            public void setValues(PreparedStatement ps, Data data) throws SQLException {
                ps.setInt(1, data.getId());
                ps.setString(2, data.getName());
                ps.setInt(3, data.getAge());
            }

            @Override
            public int batchSize() {
                return 1000;
            }
        });
    }
}

说明：

• Data 是要插入的数据实体类
• BatchExecutor 是封装了 JDBC 批量插入的工具类
• ParameterizedSetter 是一个回调接口，用于设置 PreparedStatement 的参数

3. Controller 代码

最后，在 Controller 中调用 DataInsertService 的 insertData 方法即可：

@RestController
public class DataController {
    @Autowired
    private DataInsertService dataInsertService;

    @PostMapping("/insert")
    public void insertData(@RequestBody List<Data> dataList) {
        dataInsertService.insertData(dataList);
    }
}

总结

通过异步、多线程、批量插入的方式，我们可以高效地将百万级数据插入到 Hive 中。

注意：

• BatchExecutor 类需要自行实现，可参考 JDBC 批量插入的实现方式
• 需要确保 Hive 表的结构与 Data 类中的字段匹配
• 为了防止数据丢失，建议使用事务进行操作