FlinkSideOutput怎么使用
这篇文章主要介绍“Flink SideOutput怎么使用”,在日常操作中,相信很多人在Flink SideOutput怎么使用问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Flink SideOutput怎么使用”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
创新互联建站长期为上1000+客户提供的网站建设服务,团队从业经验10年,关注不同地域、不同群体,并针对不同对象提供差异化的产品和服务;打造开放共赢平台,与合作伙伴共同营造健康的互联网生态环境。为庄河企业提供专业的成都网站建设、网站制作,庄河网站改版等技术服务。拥有十多年丰富建站经验和众多成功案例,为您定制开发。
SideOutput方法:(侧输出)从主数据流中根据outputTag获取额外的输出流(分流场景下使用)
示例环境
java.version: 1.8.x flink.version: 1.11.1
示例数据源 (项目码云下载)
Flink 系例 之 搭建开发环境与数据
SideOutput.java
import com.flink.examples.DataSource; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.api.java.tuple.Tuple4; 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; import java.util.List; /** * @Description SideOutput方法:(侧输出)从主数据流中根据outputTag获取额外的输出流(分流场景下使用) */ public class SideOutput { /** * 遍历集合,将数据流切分成多个流并打印 * @param args * @throws Exception */ public static void main(String[] args) throws Exception { final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.setParallelism(1); List> tuple3List = DataSource.getTuple3ToList(); //Datastream DataStream > dataStream = env.fromCollection(tuple3List); //自定义man和girl两个性别tag OutputTag > manTag = new OutputTag >("man"){}; OutputTag > girlTag = new OutputTag >("girl"){}; //将所有流数据输入到process做处理 SingleOutputStreamOperator > output = dataStream.process(new ProcessFunction , Tuple4 >() { @Override public void processElement(Tuple3 value, Context ctx, Collector > out) throws Exception { //将数据流按不同的性别划分,创建新的Tuple4,分别绑定到不同性别的tag if (value.f1.equals("man")){ ctx.output(manTag, Tuple4.of(value.f0, value.f1, value.f2, "男")); }else { ctx.output(girlTag, Tuple4.of(value.f0, value.f1, value.f2, "女")); } } }); //获取指定tag的数据流 DataStream > dataStream1 = output.getSideOutput(manTag); DataStream > dataStream2 = output.getSideOutput(girlTag); //打印 dataStream1.print(); dataStream2.print(); env.execute("flink Split job"); } }
打印结果
(张三,man,20,男) (李四,girl,24,女) (王五,man,29,男) (刘六,girl,32,女) (伍七,girl,18,女) (吴八,man,30,男)
到此,关于“Flink SideOutput怎么使用”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!
网站栏目:FlinkSideOutput怎么使用
本文地址:http://myzitong.com/article/posscd.html