输出算子(sink)

Flink作为数据处理框架，最终还是要把计算处理的结果写入外部存储，为外部应用提供支持。
Alt text

1. 连接到外部系统

Flink的DataStream API专门提供了向外部写入数据的方法：addSink。与addSource类似，addSink方法对应着一个"Sink"算子，主要就是用来实现与外部系统连接、并将数据提交写入的；Flink程序中所有对外的输出操作，一般都是利用Sink算子完成的。
Flink1.12以前，Sink算子的创建是通过调用DataStream的.addSink()方法实现的。

java

stream.addSink(new SinkFunction(…));

addSink方法同样需要传入一个参数，实现的是SinkFunction接口。在这个接口中只需要重写一个方法invoke()，用来将指定的值写入到外部系统中。这个方法在每条数据记录到来时都会调用。
Flink1.12开始，同样重构了Sink架构，使用也是新的API:

java

stream.sinkTo(…)

当然，Sink多数情况下同样并不需要我们自己实现。之前我们一直在使用的print方法其实就是一种Sink，它表示将数据流写入标准控制台打印输出。Flink官方为我们提供了一部分的框架的Sink连接器。如下图所示，列出了Flink官方目前支持的第三方系统连接器：
Alt text 我们可以看到，像Kafka之类流式系统，Flink提供了完美对接，source/sink两端都能连接，可读可写；而对于Elasticsearch、JDBC等数据存储系统，则只提供了输出写入的sink连接器。除此以外，就需要用户自定义实现sink连接器了。

2. 输出到文件

Flink专门提供了一个流式文件系统的连接器：FileSink，为批处理和流处理提供了一个统一的Sink，它可以将分区文件写入Flink支持的文件系统。
FileSink支持行编码(Row-encoded)和批量编码(Bulk-encoded)格式。这两种不同的方式都有各自的构建器(builder)，可以直接调用FileSink的静态方法：

行编码： FileSink.forRowFormat(basePath，rowEncoder)。
批量编码： FileSink.forBulkFormat(basePath，bulkWriterFactory)。

java

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
    env.setParallelism(1);

    // 必须开启checkpoint，否则文件一直都是 .inprogress状态
    env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);
    // 设置 数据源为生成器
    DataGeneratorSource<String> generatorSource =
            new DataGeneratorSource<>(
                    (GeneratorFunction<Long, String>) value -> "Number: " + value,
                    Long.MAX_VALUE,
                    RateLimiterStrategy.perSecond(1000),
                    Types.STRING
            );
    DataStreamSource<String> genSource = env.fromSource(generatorSource, WatermarkStrategy.noWatermarks(), "genSource");

    FileSink fileSink = FileSink
            // 输出行式存储的文件，指定路径、指定编码
            .forRowFormat(new Path("e://tmp//flink_sink"),
                        new SimpleStringEncoder("utf8"))
            // 输出文件的一些配置： 文件名的前缀、后缀
            .withOutputFileConfig(
                    OutputFileConfig.builder()
                            .withPartPrefix("dataGen")
                            .withPartSuffix(".log").build())
            // 按照目录分桶：如下，就是每个小时一个目录
            .withBucketAssigner(new DateTimeBucketAssigner())
            // 文件滚动策略:  1分钟 或 10KB
            .withRollingPolicy(DefaultRollingPolicy.builder()
                    .withRolloverInterval(Duration.ofMinutes(1))
                    .withMaxPartSize(MemorySize.parse("10KB")).build())
                    .build();

    genSource.sinkTo(fileSink);
    env.execute();
}

运行结果：
Alt text

3. 输出到Kafka

添加Kafka连接器依赖
由于我们已经测试过从Kafka数据源读取数据，连接器相关依赖已经引入，这里就不重复介绍了。
启动Kafka集群
编写输出到Kafka的代码
输出无key的record:

java

public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
	env.setParallelism(1);

	// 必须开启checkpoint，否则文件一直都是 .inprogress状态
	// 如果是精准一次，必须开启checkpoint(后续章节介绍)
	env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

	DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop102", 7777);

	KafkaSink<String> stringKafkaSink = KafkaSink.<String>builder()
			// 指定 kafka 的地址和端口
			.setBootstrapServers("hadoop104:9092")
			// 指定序列化器：指定Topic名称、具体的序列化
			.setRecordSerializer(KafkaRecordSerializationSchema.<String>builder()
					.setTopic("ws")
					.setValueSerializationSchema(new SimpleStringSchema()).build())
			// 写到kafka的一致性级别： 精准一次、至少一次
			.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
			// 如果是精准一次，必须设置 事务的前缀
			.setTransactionalIdPrefix("socketMsg-")
			// 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
			.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10*60*1000+"").build();
	socketTextStream.sinkTo(stringKafkaSink);
	env.execute();
}

运行结果：
Alt text

Kafka Sink要点

如果要使用精准一次写入Kafka，需要满足以下条件，缺一不可
1、开启checkpoint(后续介绍)
2、设置事务前缀
3、设置事务超时时间： checkpoint间隔 < 事务超时时间 < max的15分钟

自定义序列化器，实现带key的record:

java

public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
	env.setParallelism(1);

	// 必须开启checkpoint，否则文件一直都是 .inprogress状态
	// 如果是精准一次，必须开启checkpoint(后续章节介绍)
	env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

	DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop102", 7777);

	KafkaSink<String> stringKafkaSink = KafkaSink.<String>builder()
			// 指定 kafka 的地址和端口
			.setBootstrapServers("hadoop104:9092")
			// 指定序列化器：指定Topic名称、具体的序列化
			.setRecordSerializer(new KafkaRecordSerializationSchema<String>(){
					/**
					 * 如果要指定写入kafka的key，可以自定义序列化器：
					 * 1、实现 一个接口，重写 序列化 方法
					 * 2、指定key，转成 字节数组
					 * 3、指定value，转成 字节数组
					 * 4、返回一个 ProducerRecord对象，把key、value放进去
					 */
					@Nullable
					@Override
					public ProducerRecord<byte[], byte[]> serialize(String element, KafkaSinkContext context, Long timestamp) {
					String[] split = element.split(",");
					byte[] key = split[0].getBytes(StandardCharsets.UTF_8);
					byte[] value = element.getBytes(StandardCharsets.UTF_8);
					return new ProducerRecord<>("ws", key, value);
					}
			})
			// 写到kafka的一致性级别： 精准一次、至少一次
			.setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
			// 如果是精准一次，必须设置 事务的前缀
			.setTransactionalIdPrefix("socketMsg-")
			// 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
			.setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10*60*1000+"").build();
	socketTextStream.sinkTo(stringKafkaSink);
	env.execute();
}

4. 输出到MySQL(JDBC)

4.1 添加相关依赖

xml

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.27</version>
</dependency>
<dependency>
	<groupId>org.apache.flink</groupId>
	<artifactId>flink-connector-jdbc</artifactId>
	<version>3.1.2-1.17</version>
</dependency>

4.2 建表ws

启动MySQL，在test库下建表ws:

sql

CREATE TABLE `ws` (
  `id` varchar(100) NOT NULL,
  `ts` bigint(20) DEFAULT NULL,
  `vc` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8

4.4 编写输出到MySQL的示例代码

java

public static void main(String[] args) throws Exception {
	StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
	env.setParallelism(1);

	// 必须开启checkpoint，否则文件一直都是 .inprogress状态
	// 如果是精准一次，必须开启checkpoint(后续章节介绍)
	env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

	SingleOutputStreamOperator<WaterSensor> streamOperator = env.socketTextStream("hadoop102", 7777)
			.map(new MapFunction<String, WaterSensor>() {

				@Override
				public WaterSensor map(String value) throws Exception {
					String[] split = value.split(",");
					return new WaterSensor(split[0], Long.valueOf(split[1]), Integer.valueOf(split[2]));
				}
			});

	String insertSql = "INSERT INTO test.ws (id, ts, vc) VALUES(?, ?, ?);";
	SinkFunction<WaterSensor> mysqlSink = JdbcSink.sink(
			insertSql,
			new JdbcStatementBuilder<WaterSensor>() {

				@Override
				public void accept(PreparedStatement ps, WaterSensor waterSensor) throws SQLException {
					ps.setString(1, waterSensor.getId());
					ps.setLong(2, waterSensor.getTs());
					ps.setInt(3, waterSensor.getVc());
				}
			},
			JdbcExecutionOptions.builder()
					// 重试次数
					.withMaxRetries(3)
					// 批次的大小：条数
					.withBatchSize(10)

					// 批次的时间
					.withBatchIntervalMs(3000).build(),
			new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
					.withDriverName("com.mysql.cj.jdbc.Driver")
					.withUrl("jdbc:mysql://hadoop104:3307/test?useSSL=false&characterEncoding=utf-8&useUnicode=true")
					.withUsername("root")
					.withPassword("123456")
					.build()
	);
	streamOperator.addSink(mysqlSink);
	env.execute();
}

运行最终结果(可以查看web页面的日志辅助排查问题)：
Alt text

总结

只能用老的sink写法： addsink
JDBCSink的4个参数:
- 第一个参数：执行的sql，一般就是 insert into
- 第二个参数：预编译sql, 对占位符填充值
- 第三个参数：执行选项 ---》攒批、重试
- 第四个参数：连接选项 ---》 url、用户名、密码

5. 自定义Sink输出

如果我们想将数据存储到我们自己的存储设备中，而Flink并没有提供可以直接使用的连接器，就只能自定义Sink进行输出了。与Source类似，Flink为我们提供了通用的SinkFunction接口和对应的RichSinkDunction抽象类，只要实现它，通过简单地调用DataStream的.addSink()方法就可以自定义写入任何外部存储。

java

stream.addSink(new MySinkFunction<String>());

在实现SinkFunction的时候，需要重写的一个关键方法invoke()，在这个方法中我们就可以实现将流里的数据发送出去的逻辑。
这种方式比较通用，对于任何外部存储系统都有效；不过自定义Sink想要实现状态一致性并不容易，所以一般只在没有其它选择时使用。实际项目中用到的外部连接器Flink官方基本都已实现，而且在不断地扩充，因此自定义的场景并不常见。

最后更新于: 2025-03-19 18:00

登录后即可评论