4. Flink DataStream API 进阶：Sink算子、多流操作与分区策略
好，咱们继续往下走。上一章我们把 DataStream API 的 Source 和 Transformation 算子捋了一遍，你大概已经能写出一个能跑起来的 Flink 任务了。但一个完整的流处理应用，光有输入和计算还不够——数据算完了，你得把它发出去，对吧？
这一章，我们就来聊聊“出口”和“分流”的那些事。说白了，就是 Sink 算子、多流操作 和 分区策略。这三个东西，在实际项目中几乎是天天见。我个人习惯把它们放在一起讲，因为它们共同决定了你的数据最终流向哪里、怎么合并、以及如何在集群里分布。
4.1 Sink 算子：数据最终去哪了？
Sink，就是数据流的终点。Flink 内置了很多现成的 Sink，比如写到文件、写到 Kafka、写到数据库。但有时候，内置的满足不了你，那就得自己写一个。
4.1.1 写入文件：StreamingFileSink
先说说写文件。你可能会想：“写文件还不简单？每条数据来的时候，open 一个 FileOutputStream，write 一下，完事。” 嗯，你要是真这么干，在分布式环境下会出大问题——多个并行子任务同时写一个文件，数据就乱套了。
Flink 提供了 StreamingFileSink，它专门解决这个问题。它会按你指定的规则（比如按时间、按文件大小）滚动文件，并且保证 Exactly-Once 语义。
核心思路：StreamingFileSink 会把数据先写到临时文件，等 checkpoint 完成后再正式提交（commit）。这样即使任务挂了，也能从 checkpoint 恢复，不会丢数据。
来看个例子，把数据按天分区写到 HDFS：
DataStream<String> input = ...;
StreamingFileSink<String> sink = StreamingFileSink
.forRowFormat(
new Path("hdfs://namenode:8020/data/logs"),
new SimpleStringEncoder<String>("UTF-8")
)
.withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd"))
.withRollingPolicy(
DefaultRollingPolicy.builder()
.withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
.withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
.withMaxPartSize(1024 * 1024 * 128) // 128MB
.build()
)
.build();
input.addSink(sink);这里我解释几个关键点：
- BucketAssigner：决定数据写到哪个“桶”（目录）里。上面用的是按天分桶，你也可以自定义。
- RollingPolicy：决定什么时候滚动文件。我建议你根据业务场景来配，比如日志类数据，15分钟滚动一次比较常见；如果是大吞吐的指标数据，可以按128MB文件大小来切。
我的经验：我曾经在一个日志采集项目里，默认用了按文件大小滚动，结果因为某天流量突增，小文件特别多，搞得 HDFS NameNode 压力很大。后来改成按时间+大小双重策略，才稳住。所以，文件滚动策略一定要结合你的数据量级来调。
4.1.2 写入 Kafka：FlinkKafkaProducer
Kafka 是 Flink 最常用的“下游”之一。Flink 和 Kafka 的集成非常成熟，FlinkKafkaProducer 支持 Exactly-Once 语义（需要配合 Kafka 的事务机制）。
写 Kafka 的代码很简洁：
DataStream<String> stream = ...;
Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("transaction.timeout.ms", "60000");
FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>(
"my-output-topic",
new SimpleStringSchema(),
props,
FlinkKafkaProducer.Semantic.EXACTLY_ONCE
);
stream.addSink(kafkaSink);这里有个坑，我必须提醒你：
注意：使用 Exactly-Once 语义时，Kafka 的 transaction.timeout.ms 必须大于 Flink 的 checkpoint 间隔，否则会频繁报错。我曾经因为没配这个参数，线上任务每几分钟就挂一次，排查了半天才发现是事务超时。
4.1.3 自定义 Sink：RichSinkFunction
当内置 Sink 满足不了你时，比如你要写入 Redis、Elasticsearch、或者某个自研的存储系统，那就得自己实现 RichSinkFunction。
自定义 Sink 的核心是三个方法：
open()：初始化连接、打开文件等。invoke()：每条数据到达时调用，你在这里写具体的写入逻辑。close()：清理资源，关闭连接。
举个写入 Redis 的例子：
public class RedisSink extends RichSinkFunction<Tuple2<String, String>> {
private transient Jedis jedis;
@Override
public void open(Configuration parameters) throws Exception {
// 在 open 里初始化连接，避免每条数据都创建连接
jedis = new Jedis("redis-host", 6379);
jedis.auth("password");
}
@Override
public void invoke(Tuple2<String, String> value, Context context) throws Exception {
// 写入 Redis，key 是 value.f0，value 是 value.f1
jedis.set(value.f0, value.f1);
}
@Override
public void close() throws Exception {
if (jedis != null) {
jedis.close();
}
}
}
// 使用
DataStream<Tuple2<String, String>> stream = ...;
stream.addSink(new RedisSink());避坑指南：我曾经在自定义 Sink 里犯过一个低级错误——在 invoke() 方法里每次都 new 一个连接，结果连接数爆炸，直接把 Redis 打挂了。记住：连接一定要在 open() 里初始化，并且考虑连接池。
4.2 多流操作：把多条流玩出花
实际业务中，很少只有一条数据流。比如你要把用户点击流和用户画像流合并，或者把订单流和支付流关联。Flink 提供了 union、connect、coMap 等操作来处理多流场景。
4.2.1 Union：简单粗暴的合并
union 是最简单的多流操作。它可以把多个 同类型 的数据流合并成一个。注意，必须是同类型——数据类型要一样。
DataStream<String> stream1 = ...;
DataStream<String> stream2 = ...;
DataStream<String> stream3 = ...;
DataStream<String> unioned = stream1.union(stream2, stream3);合并后的流，会包含所有输入流的数据，顺序是不确定的（取决于哪个流先来数据）。
适用场景：比如你有多个数据源都在上报同一种日志，用 union 把它们合并到一起，统一处理。
4.2.2 Connect 与 CoMap：不同类型流的“联姻”
如果两条流的数据类型不一样，union 就无能为力了。这时候要用 connect。connect 可以把两条流连接起来，然后通过 CoMapFunction 或 CoFlatMapFunction 分别处理。
举个例子，把订单流和支付流关联：
DataStream<Order> orderStream = ...;
DataStream<Payment> paymentStream = ...;
ConnectedStreams<Order, Payment> connected = orderStream.connect(paymentStream);
DataStream<String> result = connected.map(new CoMapFunction<Order, Payment, String>() {
@Override
public String map1(Order order) throws Exception {
// 处理订单流的数据
return "Order: " + order.orderId;
}
@Override
public String map2(Payment payment) throws Exception {
// 处理支付流的数据
return "Payment: " + payment.paymentId;
}
});这里要注意：CoMapFunction 的两个方法 map1 和 map2 是独立调用的，每条数据只会触发其中一个。说白了，connect 只是把两条流“绑”在一起，但数据还是各自处理。如果你想做真正的“关联”（比如 join），需要结合状态或窗口。
我的习惯：我通常用 connect + KeyedBroadcastProcessFunction 来做“动态配置流”和“业务数据流”的关联。比如把规则配置流 broadcast 出去，然后业务流去查这个广播状态。这样规则可以动态更新，不用重启任务。
4.3 分区策略：数据在集群里怎么跑？
分区策略决定了数据如何分配给下游的并行子任务。选对了分区策略，能显著提升性能；选错了，可能造成数据倾斜，甚至 OOM。
Flink 提供了多种分区策略，我挑三个最常用的讲：rebalance、rescale、broadcast。
4.3.1 Rebalance：轮询分发
rebalance 使用 Round-Robin 算法，把数据轮流发给下游的每个并行子任务。它能保证每个子任务收到的数据量大致相等。
stream.rebalance().map(...);什么时候用？当你发现数据倾斜时，rebalance 是最快的解决办法。但它也有代价——数据会跨网络传输，如果上下游的并行度一样，其实没必要用。
4.3.2 Rescale：本地轮询
rescale 和 rebalance 有点像，也是轮询。但区别在于：rescale 只在“本地”轮询，不会跨 TaskManager 传输数据。
举个例子：上游有 4 个并行子任务，下游有 4 个。如果它们都在同一个 TaskManager 上，rescale 就是本地传输，效率很高。但如果上下游不在同一个 TaskManager，rescale 会尽量保持数据在本地。
性能对比：rescale 比 rebalance 更轻量，因为它避免了不必要的网络开销。但 rebalance 的负载均衡更均匀。
4.3.3 Broadcast：广播给所有人
broadcast 会把每条数据发送给下游的 所有 并行子任务。这通常用于分发小数据量的“配置”或“规则”。
stream.broadcast().map(...);千万别滥用：如果你把大数据量的流 broadcast 出去，每个子任务都会收到全量数据，内存会爆炸。我曾经见过一个新手把整个数据流 broadcast，结果 10 个并行度，内存直接翻了 10 倍，任务秒挂。
4.4 本章知识体系总览
为了让你更直观地理解这一章的内容，我画了一张图，把 Sink、多流操作、分区策略串起来：
这张图把整个流程串起来了：数据从源头进来，经过多流操作（合并或连接），然后通过分区策略分发到下游，最后通过 Sink 输出。每一步都有多种选择，选对了，你的任务就稳了。
4.5 小结
这一章的内容，说白了就是三个字：出口、合并、分发。
- Sink 是数据的最终归宿，写文件用 StreamingFileSink，写 Kafka 用 FlinkKafkaProducer，特殊场景自己实现 RichSinkFunction。记住：连接要复用，事务要配好。
- 多流操作 帮你处理多条数据流的关系。同类型用 union，不同类型用 connect + coMap。如果想做真正的关联，别忘了结合状态。
- 分区策略 决定了数据怎么跑。数据倾斜用 rebalance，追求性能用 rescale，小配置用 broadcast。千万别把大流 broadcast 出去，血的教训。
嗯，这一章就到这里。代码不多，但每个点都是实战中会踩的坑。你把这些搞明白了，Flink 的 DataStream API 就算入门了八成。剩下的两成，咱们后面慢慢聊。