写点什么

Flink Side Outputs

用户头像
Alex🐒
关注
发布于: 2021 年 06 月 20 日

Side Output 功能从 Flink 1.3.0 版本开始提供,FLINK-4460


除了从 DataStream 操作的结果中获取主数据流之外,还可以产生任意数量额外的旁路输出(side output)结果流。旁路输出的数据类型不需要与主数据流的类型一致,不同旁路输出的类型也可以不同。当你想要拆分数据流时(通常需要复制流),然后从每个流过滤出不需要的数据,此操作将非常有用。


当使用旁路输出时,首先需要定义一个 OutputTag 对象,用来标识一个旁路输出流:

// 这需要是一个匿名的内部类,以便我们分析类型OutputTag<String> outputTag = new OutputTag<String>("side-output") {};
复制代码


注意 OutputTag 是如何根据旁路输出流所包含的元素类型进行类型化的。可以通过以下函数将数据发送到旁路输出:

  • ProcessFunction(1.3.x)

  • KeyedProcessFunction(1.8.x)

  • CoProcessFunction(1.4.x)

  • KeyedCoProcessFunction(1.9.x)

  • ProcessWindowFunction(1.4.x)

  • ProcessAllWindowFunction(1.4.x)


可以使用 Context (在上述函数中会向用户提供使用)将数据发送到由 OutputTag 标识的旁路输出。下面的例子是在 ProcessFunction 中发出旁路输出数据:

DataStream<Integer> input = ...;
final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};
SingleOutputStreamOperator<Integer> mainDataStream = input .process(new ProcessFunction<Integer, Integer>() {
@Override public void processElement( Integer value, Context ctx, Collector<Integer> out) throws Exception { // 发送数据到主要的输出 out.collect(value);
// 发送数据到旁路输出 ctx.output(outputTag, "sideout-" + String.valueOf(value)); } });
复制代码


要获取旁路输出流,可以在 DataStream 上调用 getSideOutput(OutputTag) 来获取,如下面的例子:

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};
SingleOutputStreamOperator<Integer> mainDataStream = ...;
DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);
复制代码



[1]  https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/datastream/side_output/ 

发布于: 2021 年 06 月 20 日阅读数: 3
用户头像

Alex🐒

关注

还未添加个人签名 2020.04.30 加入

还未添加个人简介

评论

发布
暂无评论
Flink Side Outputs