请问flink-connector-jdbc在sink时，如何设置批量写入？[阿里云实时计算 Flink版] 暂停朗读为您朗读

在使用 Flink 提供的 JDBC Connector 时，可以通过设置 batchSize 属性来实现批量写入。具体操作如下：

1. 引入依赖

在项目的 pom.xml 文件中添加 JDBC Connector 的依赖：

    org.apache.flink    flink-connector-jdbc_2.11    ${flink.version}

1. 创建 JDBC Sink

在创建 JDBC Sink 时，可以通过 JDBCOutputFormatBuilder 的 setBatchIntervalMillis 方法设置批量写入的时间间隔，单位是毫秒。可以根据实际场景调整时间间隔和批量大小：

JDBCOutputFormatBuilder builder = new JDBCOutputFormatBuilder()        .setDrivername(driverName)        .setDBUrl(dbURL)        .setUsername(userName)        .setPassword(password)        .setQuery(insertQuery)        .setBatchIntervalMillis(batchIntervalMillis)    // 设置批量写入的时间间隔        .setBatchSize(batchSize);                       // 设置批量大小JDBCOutputFormat outputFormat = builder.finish();JDBCTableSink jdbcSink = JDBCTableSink.builder()        .setOutputFormat(outputFormat)        .build();

其中，batchSize 属性可以设置批量写入的大小，即将多个数据行打包成一个批次进行写入。注意，这个值需要根据你的数据量和实际场景进行调整，过小可能会影响写入性能，过大可能会导致内存溢出。

1. 将 Sink 应用到 Flink 流计算中

将 JDBC Sink 应用到 Flink 流计算中，例如：

DataStream inputStream = ...;inputStream.addSink(jdbcSink);

在应用到 Flink 流计算中时，可以通过 setFlushOnCheckpoint(true) 方法设置在Checkpoint时也要进行数据写入，保证数据的完整性和可靠性。

在使用 Flink 的 JDBC sink 时，可以通过配置 batchSize 来设置批量写入的大小。batchSize 表示一次性向 JDBC 批量提交多少条记录，可以设置为整数值。

例如，以下代码片段使用 JDBC sink 向 MySQL 数据库中批量写入数据，每次提交 100 条记录：

DataStream<MyData> stream = ...;FlinkJdbcSink<MyData> sink = JdbcSink.<MyData>batch(        "INSERT INTO myTable (col1, col2, col3) VALUES (?, ?, ?)",        (ps, data) -> {            ps.setString(1, data.getField1());            ps.setInt(2, data.getField2());            ps.setLong(3, data.getField3());        },        JdbcExecutionOptions.builder()                .withBatchSize(100)                .build(),        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()                .withUrl("jdbc:mysql://localhost:3306/myDb")                .withDriverName("com.mysql.jdbc.Driver")                .withUsername("myUser")                .withPassword("myPassword")                .build());stream.addSink(sink);

在上述示例中，使用了 FlinkJdbcSink 的静态方法 batch，并传入了 batchSize 参数。该参数被设置为 100，因此会每次向数据库批量提交 100 条记录。

注意，此时需要保证配置的 batchSize 不超过 JDBC 驱动的最大批量提交大小，否则可能会出现异常。通常情况下，可以根据系统性能和数据量大小综合考虑一个适当的 batchSize 值。

Flink JDBC连接器提供了对MySQL、PostgreSQL和Oracle等常见的数据库读写支持，如果需要配置批量写入可以设置sink.buffer-flush.max-rows 大于0 ，flush数据前，缓存记录的最大值。以及设置sink.buffer-flush.interval 大于0，flush数据的时间间隔。数据在Flink中缓存的时间超过该参数指定的时间后，异步线程将flush数据到数据库中。同时需要注意字段类型映射，参考文档：文档

设置最大缓冲行数 设置最大缓冲时间 默认情况下，JDBC Sink 发送完成后会自动关闭连接，并提交事务。

还有一些与性能相关的调整方面也可以考虑做出：比如，修改输入并行度（即 max.parallelism），或是调整容器资源限制等都可能受益于增加批量处理参数来保证Flink Job稳定运行。

在使用Flink JDBC Connector的Sink写入数据库时,设置批量flush参数是实现高吞吐量的关键。主要有两个相关参数: 1. sink.buffer-flush.max-rows:触发flush并写入数据库的最大行数阈值。默认为1,即每插入1行就flush。 2. sink.buffer-flush.interval:触发flush的最大时间间隔。默认为0,即不按时间隔断flush。 要实现批量写入数据库,需要同时调大这两个参数。例如:

properties# 每5000行或1秒钟触发一次flushsink.buffer-flush.max-rows=5000  sink.buffer-flush.interval=1000

这样,连接器会将插入的数据缓存至内存Buffer,当数据量达到5000行或者超过1秒钟时,会触发flush并批量写入数据库。 adjust这两个参数对吞吐量的影响主要有: 1. sink.buffer-flush.max-rows越大,触发flush的频率越低,插入数据库的批量数据越多,数据库的IO压力越大,从而达到更高的吞吐量。 2. sink.buffer-flush.interval允许在达不到max-rows阈值时,也定期触发flush。这可以避免数据在Buffer中停留时间过长,同时也为数据库提供更均匀的压力,有利于吞吐量的提高。 3. 两个参数都不宜设置过大,否则会给数据库带来过高的压力和较高的延迟。需要根据数据库的处理能力进行评估和调优。 4. interval参数对event time语义的支持很关键。它可以确保无论 max-rows是否达到,数据至少以该频率超越水位线并刷新到数据库。 所以,想实现Flink JDBC Sink的批量写入和高吞吐量,调优这两个flush相关的参数是关键。需要结合具体数据库的性能,选择适当的max-rows和interval值,既能发挥批量写入的优势,又不会过分压榨数据库资源。

可以使用 Flink 官方提供的 BulkFormatBuilder 工具类，来实现 flink-connector-jdbc 的批量写入。

// 这里假设已经定义好了 POJO 类型和 JDBC 配置信息

DataStream records = …; JdbcConnectionOptions connectionOptions = …; JdbcExecutionOptions executionOptions = …;

// 使用 BulkFormatBuilder 构建 JdbcOutputFormatBuilder

JdbcOutputFormatBuilder outputFormatBuilder = JdbcOutputFormat.buildJdbcOutputFormat() .setDrivername(“org.apache.derby.jdbc.EmbeddedDriver”) .setDBUrl(“jdbc:derby:memory:test”) .setUsername(“”) .setPassword(“”) .setQuery(“INSERT INTO test (id, name) VALUES (?, ?);”) .setSqlTypes(new int[] { Types.INTEGER, Types.VARCHAR }) .setBatchIntervalMs(1000) .setBatchSize(500) .setJdbcStatementBuilder(new JdbcStatementBuilder() { public void accept(java.sql.PreparedStatement statement, MyRecord record) { statement.setInt(1, record.id); statement.setString(2, record.name); } });

// 构建 SinkFunction

SinkFunction sink = JdbcSink.sink( outputFormatBuilder, new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl(“jdbc:derby:memory:test”) .withDriverName(“org.apache.derby.jdbc.EmbeddedDriver”) .build(), new JdbcExecutionOptions.JdbcExecutionOptionsBuilder() .withBatchSize(500) .withBatchIntervalMs(1000) .withMaxRetries(3) .build());

// 将 Source 写入 Sink

records.addSink(sink);

在使用 Flink JDBC Sink 时，可以通过以下参数控制批量写入：

1、sink.buffer-flush.max-rows：设置缓冲区大小，即达到该缓冲区大小后批量写入到数据库，默认为 5000； 2、sink.buffer-flush.interval：设置缓冲区时间间隔，即达到该时间间隔后批量写入到数据库，默认为 1s； 3、sink.max-retries：设置写入失败时的最大重试次数，默认为 3； 4、sink.max-parallelism：设置写入的并行度，默认为 1。

如果你想要自定义批量写入的大小，可以通过在参数中添加如下配置实现：

.withParameters(JdbcExecutionOptions.builder().withBatchSize(1000).build()) 其中 withBatchSize() 方法可以设置批量写入的大小，即一次性写入多少条数据。例如上述示例中的 1000 表示一次性写入 1000 条数据。

另外，如果你在使用 Flink JDBC Sink 时出现了 sink.buffer-flush.max-rows 参数不生效的情况，可以尝试将该参数的值设置为较小的值，如 100，看看是否能够生效。如果仍然不生效，可能是因为数据库本身的限制导致无法批量写入。

在使用Flink的JDBC连接器时，可以通过以下两种方式来设置批量写入：

使用BatchingSinkFunction BatchingSinkFunction是一种可选的SinkFunction，它允许将多个记录作为一个批次进行写入。在Flink的JDBC连接器中，您可以通过继承或实现这个类来创建自定义的批量写入SinkFunction。

例如，下面的代码使用BatchingSinkFunction来将数据按照批量大小(100)写入MySQL：

public class MySQLBatchWriter extends BatchingJdbcSinkFunction {

public MySQLBatchWriter() {    super(JdbcExecutionOptions.builder()            .withBatchSize(100)            .build());}@Overrideprotected void prepareStatement(PreparedStatement statement, MyData data) throws SQLException {    statement.setInt(1, data.getId());    statement.setString(2, data.getName());    statement.setDouble(3, data.getValue());}

} 其中，prepareStatement方法用于准备SQL查询语句，而构造函数则使用JdbcExecutionOptions.Builder来设置批量大小。

使用JdbcExecutionOptions JdbcExecutionOptions提供了多个配置选项，其中包括设置批量大小的选项。在使用Flink的JDBC连接器时，您可以通过以下代码来设置批量大小：

JdbcExecutionOptions executionOptions = JdbcExecutionOptions.builder() .withBatchSize(100) .build();

JdbcSink sink = JdbcSink.sink( “INSERT INTO my_table (id, name, value) VALUES (?, ?, ?)”, (statement, data) -> { statement.setInt(1, data.getId()); statement.setString(2, data.getName()); statement.setDouble(3, data.getValue()); }, JdbcExecutionOptions.builder() .withBatchSize(100) .build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder() .withUrl(“jdbc:mysql://localhost:3306/my_database”) .withDriverName(“com.mysql.jdbc.Driver”) .withUsername(“root”) .withPassword(“root”) .build());

dataStream.addSink(sink); 其中，withBatchSize方法用于设置批量大小，并将其传递给JdbcExecutionOptions.Builder。