ApacheFlink官方文档--流(DataStreamAPI)-旁路输出

旁路输出(side output)

  除了来自数据流算子的主流结果输出之外，可以产生任意数量的流旁路输出结果。旁路输出结果数据类型与主流结果的数据类型以及其他旁路输出结果数据类型可以是完全不同的。当你需要分割数据流时，这个算子非常有用。通常需要复制流，然后从每个数据流中过滤掉不需要的数据。
 当使用旁路输出时，首先需要定义一个OutputTag来标识一个旁路输出流。
Java

西畴ssl适用于网站、小程序/APP、API接口等需要进行数据传输应用场景，ssl证书未来市场广阔！成为创新互联的ssl证书销售渠道，可以享受市场价格4-6折优惠！如果有意向欢迎电话联系或者加微信：13518219792（备注：SSL证书合作）期待与您的合作！

// this needs to be an anonymous inner class, so that we can analyze the type
OutputTag<String> outputTag = new OutputTag<String>("side-output") {};

Scala

val outputTag = OutputTag[String]("side-output")

  注意OutputTag是如何根据旁路输出流包含的元素类型typed的。
 可以通过以下函数发射数据到旁路输出。

• ProcessFunction
• CoProcessFunction
• ProcessWindowFunction
• ProcessAllWindowFunction

  可以使用Context参数（在上述函数中向用户暴露）将数据发送到OutputTag标识的旁路输出。以下是从ProcessFunction发出旁路输出数据的示例：
Java:

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 {
        // emit data to regular output
        out.collect(value);

        // emit data to side output
        ctx.output(outputTag, "sideout-" + String.valueOf(value));
      }
    });

Scala:

val input: DataStream[Int] = ...
val outputTag = OutputTag[String]("side-output")

val mainDataStream = input
  .process(new ProcessFunction[Int, Int] {
    override def processElement(
        value: Int,
        ctx: ProcessFunction[Int, Int]#Context,
        out: Collector[Int]): Unit = {
      // emit data to regular output
      out.collect(value)

      // emit data to side output
      ctx.output(outputTag, "sideout-" + String.valueOf(value))
    }
  })

  要读取旁路输出流，在数据流运算后使用getSideOutput(OutputTag)。此时将会获得键入旁路输出流的结果。
Java:

final OutputTag<String> outputTag = new OutputTag<String>("side-output"){};

SingleOutputStreamOperator<Integer> mainDataStream = ...;

DataStream<String> sideOutputStream = mainDataStream.getSideOutput(outputTag);

Scala:

val outputTag = OutputTag[String]("side-output")

val mainDataStream = ...

val sideOutputStream: DataStream[String] = mainDataStream.getSideOutput(outputTag)

分享文章：ApacheFlink官方文档--流(DataStreamAPI)-旁路输出
文章位置：https://www.cdcxhl.com/article2/gjjpic.html

成都网站建设公司_创新互联，为您提供面包屑导航、网站策划、自适应网站、移动网站建设、、外贸网站建设

广告

声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源： 创新互联