如何让Flink Sink算子识别多上游Process Function算子的数据来源

不需要在上游数据中添加额外字段来标识来源，以下两种方案可以避免重复IO，直接实现Sink对上游算子的识别：

1. 侧输出流（Side Output）分流方案 #

利用Flink的侧输出流特性，让每个上游ProcessFunction将数据发送到专属的侧输出流，Sink通过监听不同侧输出流来区分来源：

实现步骤 #

• 为每个上游算子定义唯一的OutputTag：

  private static final OutputTag<YourData> TAG_PROC_A = new OutputTag<>("process-function-a") {};
  private static final OutputTag<YourData> TAG_PROC_B = new OutputTag<>("process-function-b") {};
  

• 上游ProcessFunction处理数据后，将结果发送到对应侧输出流：

  public class ProcFuncA extends ProcessFunction<YourData, YourData> {
      @Override
      public void processElement(YourData value, Context ctx, Collector<YourData> out) {
          // 业务处理逻辑
          ctx.output(TAG_PROC_A, value);
      }
  }
  

• Sink分别绑定不同侧输出流，直接携带来源标识：

  SingleOutputStreamOperator<YourData> mainStream = ...;
  mainStream.getSideOutput(TAG_PROC_A).addSink(new CustomSink("from-proc-a"));
  mainStream.getSideOutput(TAG_PROC_B).addSink(new CustomSink("from-proc-b"));
  

• 自定义Sink接收并使用来源标识：

  public class CustomSink extends RichSinkFunction<YourData> {
      private final String sourceTag;
  
      public CustomSink(String sourceTag) {
          this.sourceTag = sourceTag;
      }
  
      @Override
      public void invoke(YourData value, Context context) {
          // 为数据添加来源标签
          value.setSourceTag(sourceTag);
          // 执行输出逻辑
      }
  }
  

这种方案无需修改原始数据结构，Flink内部对侧输出流的数据复用有优化，IO开销远小于全局添加字段。

2. 自定义分区绑定方案 #

通过自定义分区策略，将每个上游算子的输出固定分配到Sink的特定子任务，Sink在初始化时根据子任务信息映射上游来源：

实现步骤 #

• 自定义分区器，以上游算子名称为分区键：

  public class SourceNamePartitioner implements Partitioner<String> {
      @Override
      public int partition(String key, int numPartitions) {
          return key.hashCode() % numPartitions;
      }
  }
  

• 上游算子输出时绑定分区策略：

  DataStream<YourData> streamA = env.process(new ProcFuncA())
      .partitionCustom(new SourceNamePartitioner(), 
          value -> getRuntimeContext().getOperatorName());
  

• Sink在open方法中获取子任务对应的上游标识：

  public class CustomSink extends RichSinkFunction<YourData> {
      private String upstreamSource;
  
      @Override
      public void open(Configuration parameters) {
          int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask();
          // 提前维护分区与上游算子的映射关系
          this.upstreamSource = switch(subtaskIdx) {
              case 0 -> "proc-func-a";
              case 1 -> "proc-func-b";
              default -> "unknown";
          };
      }
  
      @Override
      public void invoke(YourData value, Context context) {
          value.setSourceTag(upstreamSource);
          // 执行输出逻辑
      }
  }
  

这种方案性能更优，适合低延迟场景，但需要提前规划分区与上游的映射关系，灵活性稍弱。

方案选择建议 #

• 若上游算子数量动态变化，优先选择侧输出流方案，扩展性更强
• 若对性能要求极高且上游算子固定，选择自定义分区方案，避免多流管理开销

内容的提问来源于stack exchange，提问作者Baiqing