flink使用preparedstatement batch方式提交到mysql

Flink 中的 JDBC 连接通常会使用 PreparedStatement 的批量提交 (Batch Execution) 来提高性能，特别是在处理大量数据的时候。PreparedStatement 是一种预编译的 SQL 语句，它允许你在一次提交中发送一系列相似的 SQL 操作。

以下是使用 Flink JDBC API 执行 PreparedStatement 的批量提交到 MySQL 的步骤：

1. **创建连接**: 首先，你需要通过 `StreamExecutionEnvironment` 获取一个 `JdbcConnection`，设置数据库 URL、用户名和密码。

JdbcOutputFormat<String> jdbcOut = new JdbcOutputFormat<>(
        "jdbc:mysql://localhost:3306/mydatabase", // 数据库连接字符串
        "username", "password", // 用户名和密码
        "table_name" // 输出表名
);

2. **准备批量操作**: 创建一个 `List<PreparedStatement>` 或 `Array` 对象，用于存储预编译的 SQL 语句。这可以是一批 INSERT 语句，例如插入一条记录的数据集合。

List<String> sqlStatements = new ArrayList<>();
for (MyRecord record : recordsToInsert) {
    String sql = "INSERT INTO table_name (column1, column2) VALUES (?, ?)";
    sqlStatements.add(preparedStatement.prepareStatement(sql));
}

3. **填充参数**: 使用 `setXXX` 方法给每个 PreparedStatement 设置对应的值。

4. **执行批量插入**: 使用 `setParameters` 方法批量设置所有 PreparedStatement 的参数，然后执行它们。

for (int i = 0; i < sqlStatements.size(); i++) {
    sqlStatements.get(i).setString(1, record.getColumn1());
    sqlStatements.get(i).setString(2, record.getColumn2());
}

try (AutoCloseable ignored = BatchedExecutionMode.batch(jdbcOut)) {
    for (PreparedStatement statement : sqlStatements) {
        statement.executeBatch();
    }
}

5. **关闭连接**: 确保在操作完成后关闭连接。

注意，批量操作可能会导致更好的性能，但也需要小心管理内存，因为所有的 PreparedStatement 都会保持在内存中直到批量提交完成。此外，如果事务失败，整个批次都会回滚。