flink通过ProcessFunction来分流，可以将一份流进行拆分、复制等操作，比如下面的代码通过读取一个基本的文本流，将流分别做处理后进行输出：

案例代码

package wordcount;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;public class manyOutWordCount {public static void main(String[] args) throws Exception {// 1.创建流式执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2.从文件中读取数据DataStream<String> dataStream = env.readTextFile("src/main/resources/hello.txt");// 执行环境并行度设置3env.setParallelism(3);// 3.按照空格分词,流的类型是new Tuple2<>(wordLine, 1)DataStream<Tuple2<String, Integer>> sensorStream = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {@Overridepublic void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {String[] wordString = value.split(" ");for (String wordLine : wordString) {out.collect(new Tuple2<>(wordLine, 1));}}});//旁路输出，拆分流final OutputTag<Tuple2<String, Integer>> sideStream = new OutputTag<Tuple2<String, Integer>>("te") {};SingleOutputStreamOperator<Tuple2<String, Integer>> mainDataStream = sensorStream.process(new ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {@Overridepublic void processElement(Tuple2<String, Integer> value, ProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>.Context ctx, Collector<Tuple2<String, Integer>> out) throws Exception {out.collect(new Tuple2<>(value.f0, 2)); // 这里把 mainDataStream 的输出变为 Tuple(单词,2)ctx.output(sideStream, value); // 这里把 sideStream 的输出变为 Tuple(单词,1)}});DataStream<Tuple2<String, Integer>> sideOutput = mainDataStream.getSideOutput(sideStream);//获取sideOutput的数据sideOutput.print();mainDataStream.print();//执行env.execute();}
}

其中数据hello.txt的文件内容是：

hello world
hello flink
hello spark
When we have shuffled off this mortal coil
When we have shuffled off this mortal coil
ack
hello world
hello flink
hello spark
When we have shuffled off this mortal coil
When we have shuffled off this mortal coil
ack

flink java旁路输出(Side Output)，对原始流进行分流、复制相关推荐

java 输出赌赢unicod_JAVA基础整理-90.Java输入/输出（I/O）流
Java流的概念:什么是输入/输出流? 在 Java 中所有数据都是使用流读写的.流是一组有序的数据序列,将数据从一个地方带到另一个地方.根据数据流向的不同,可以分为输入(Input)流和输出(Out ...
Flink流计算处理-旁路输出
使用Flink做流数据处理时,除了主流数据输出,还自定义侧流输出即旁路输出,以实现灵活的数据拆分. 定义旁路输出标签首先需要定义一个OutputTag,代码如下: // 这需要是一个匿名的内部类,以 ...
java学习输出文档
知识地图一.类设计 1.面向对象的设计思想的理解面向对象vs面向过程(洗衣服) 封装: 就是把内部的东西保护起来,不被外界所看到. 继承: 就是用于类的扩展多态: 概念:同一操作作用 ...
Flink - Java篇
文章目录前言一.概述 1 Flink是什么 2 架构分层 3 数据处理流水线 4 运行组件 TaskManager JobManager ResourceManager Dispatcher 5 ...
php内容缓存输出,PHP使用缓存即时输出内容(output buffering)的方法
PHP使用缓存即时输出内容(output buffering)的方法 PHP使用缓存即时输出内容(output buffering)的方法.分享给大家供大家参考.具体如下: $buffer = ini ...
Flink FileSink 自定义输出路径——BucketingSink
今天看到有小伙伴在问,就想着自己实现一下. 问题: Flink FileSink根据输入数据指定输出位置,比如讲对应日期的数据输出到对应目录输入数据: 20190716 输出到路径 20190716 ...
在Mac上使用idea搭建flink java开发环境
1.环境本文档记录的是使用flink的java API简单地创建应用的过程. 前置条件:需要安装Java.maven和flink. 1.1 Java环境变量 Java需要jdk.path.class ...
java倒序输出数字的方法
1.在输入框中输入一个整数,比如要输入"5",需要输出倒序,可以使用数字键盘进行输入,也可以使用文本编辑器进行输入. 2.在命令行中输入"6",如图所示. 3. ...
Word处理控件Aspose.Words功能演示：使用 Java 处理 Word 文档的原始版本或修订版本
Aspose.Words是一种高级Word文档处理API,用于执行各种文档管理和操作任务.API支持生成,修改,转换,呈现和打印文档,而无需在跨平台应用程序中直接使用Microsoft Word. A ...

flink java旁路输出(Side Output)，对原始流进行分流、复制

案例代码

flink java旁路输出(Side Output)，对原始流进行分流、复制相关推荐

最新文章

热门文章