等待下一个秋

  • Spark
  • Flink
  • Hive
  • 数据仓库
  • 资料文档
  • Java
    • Spring
    • Mybatis
    • SpringBoot
    • 面试题
  • Python
    • Python基础
    • 爬虫
    • Numpy
    • matplotlib
    • Flask
  • 知识星球
  • 副业挣钱
  • 技术杂谈
    • Linux知识
    • Docker
    • Git教程
    • Redis教程
    • mysql
    • 前端
    • R语言
    • 机器学习
  • 其它
    • 关于我
    • 资料下载
    • 收徒弟
专注于Hadoop/Spark/Flink/Hive/数据仓库等
关注公众号:大数据技术派,获取更多学习资料。
  1. 首页
  2. Flink
  3. 正文

第10讲:Flink Side OutPut 分流

2022年1月27日 538点热度 0人点赞 0条评论

Flink系列文章

  1. 第01讲:Flink 的应用场景和架构模型
  2. 第02讲:Flink 入门程序 WordCount 和 SQL 实现
  3. 第03讲:Flink 的编程模型与其他框架比较
  4. 第04讲:Flink 常用的 DataSet 和 DataStream API
  5. 第05讲:Flink SQL & Table 编程和案例
  6. 第06讲:Flink 集群安装部署和 HA 配置
  7. 第07讲:Flink 常见核心概念分析
  8. 第08讲:Flink 窗口、时间和水印
  9. 第09讲:Flink 状态与容错

这一课时将介绍 Flink 中提供的一个很重要的功能:旁路分流器。

分流场景

我们在生产实践中经常会遇到这样的场景,需把输入源按照需要进行拆分,比如我期望把订单流按照金额大小进行拆分,或者把用户访问日志按照访问者的地理位置进行拆分等。面对这样的需求该如何操作呢?

分流的方法

通常来说针对不同的场景,有以下三种办法进行流的拆分。

Filter 分流

image (9).png

Filter 方法我们在第 04 课时中(Flink 常用的 DataSet 和 DataStream API)讲过,这个算子用来根据用户输入的条件进行过滤,每个元素都会被 filter() 函数处理,如果 filter() 函数返回 true 则保留,否则丢弃。那么用在分流的场景,我们可以做多次 filter,把我们需要的不同数据生成不同的流。

来看下面的例子:

复制代码

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    //获取数据源
    List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();
    data.add(new Tuple3<>(0,1,0));
    data.add(new Tuple3<>(0,1,1));
    data.add(new Tuple3<>(0,2,2));
    data.add(new Tuple3<>(0,1,3));
    data.add(new Tuple3<>(1,2,5));
    data.add(new Tuple3<>(1,2,9));
    data.add(new Tuple3<>(1,2,11));
    data.add(new Tuple3<>(1,2,13));

    DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);

    SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> zeroStream = items.filter((FilterFunction<Tuple3<Integer, Integer, Integer>>) value -> value.f0 == 0);

    SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> oneStream = items.filter((FilterFunction<Tuple3<Integer, Integer, Integer>>) value -> value.f0 == 1);

    zeroStream.print();

    oneStream.printToErr();

    //打印结果

    String jobName = "user defined streaming source";

    env.execute(jobName);

}

在上面的例子中我们使用 filter 算子将原始流进行了拆分,输入数据第一个元素为 0 的数据和第一个元素为 1 的数据分别被写入到了 zeroStream 和 oneStream 中,然后把两个流进行了打印。

image (10).png

可以看到 zeroStream 和 oneStream 分别被打印出来。

Filter 的弊端是显而易见的,为了得到我们需要的流数据,需要多次遍历原始流,这样无形中浪费了我们集群的资源。

Split 分流

Split 也是 Flink 提供给我们将流进行切分的方法,需要在 split 算子中定义 OutputSelector,然后重写其中的 select 方法,将不同类型的数据进行标记,最后对返回的 SplitStream 使用 select 方法将对应的数据选择出来。

我们来看下面的例子:

复制代码

public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    //获取数据源

    List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();

    data.add(new Tuple3<>(0,1,0));

    data.add(new Tuple3<>(0,1,1));

    data.add(new Tuple3<>(0,2,2));

    data.add(new Tuple3<>(0,1,3));

    data.add(new Tuple3<>(1,2,5));

    data.add(new Tuple3<>(1,2,9));

    data.add(new Tuple3<>(1,2,11));

    data.add(new Tuple3<>(1,2,13));

    DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);

    SplitStream<Tuple3<Integer, Integer, Integer>> splitStream = items.split(new OutputSelector<Tuple3<Integer, Integer, Integer>>() {

        @Override

        public Iterable<String> select(Tuple3<Integer, Integer, Integer> value) {

            List<String> tags = new ArrayList<>();

            if (value.f0 == 0) {

                tags.add("zeroStream");

            } else if (value.f0 == 1) {

                tags.add("oneStream");

            }

            return tags;

        }

    });

    splitStream.select("zeroStream").print();

    splitStream.select("oneStream").printToErr();

    //打印结果

    String jobName = "user defined streaming source";

    env.execute(jobName);

}

同样,我们把来源的数据使用 split 算子进行了切分,并且打印出结果。

image (11).png

但是要注意,使用 split 算子切分过的流,是不能进行二次切分的,假如把上述切分出来的 zeroStream 和 oneStream 流再次调用 split 切分,控制台会抛出以下异常。

复制代码

Exception in thread "main" java.lang.IllegalStateException: Consecutive multiple splits are not supported. Splits are deprecated. Please use side-outputs.

这是什么原因呢?我们在源码中可以看到注释,该方式已经废弃并且建议使用最新的 SideOutPut 进行分流操作。

image (12).png

SideOutPut 分流

SideOutPut 是 Flink 框架为我们提供的最新的也是最为推荐的分流方法,在使用 SideOutPut 时,需要按照以下步骤进行:

  • 定义 OutputTag
  • 调用特定函数进行数据拆分
    • ProcessFunction
    • KeyedProcessFunction
    • CoProcessFunction
    • KeyedCoProcessFunction
    • ProcessWindowFunction
    • ProcessAllWindowFunction

在这里我们使用 ProcessFunction 来讲解如何使用 SideOutPut:

复制代码

public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    //获取数据源

    List data = new ArrayList<Tuple3<Integer,Integer,Integer>>();

    data.add(new Tuple3<>(0,1,0));

    data.add(new Tuple3<>(0,1,1));

    data.add(new Tuple3<>(0,2,2));

    data.add(new Tuple3<>(0,1,3));

    data.add(new Tuple3<>(1,2,5));

    data.add(new Tuple3<>(1,2,9));

    data.add(new Tuple3<>(1,2,11));

    data.add(new Tuple3<>(1,2,13));

    DataStreamSource<Tuple3<Integer,Integer,Integer>> items = env.fromCollection(data);

    OutputTag<Tuple3<Integer,Integer,Integer>> zeroStream = new OutputTag<Tuple3<Integer,Integer,Integer>>("zeroStream") {};

    OutputTag<Tuple3<Integer,Integer,Integer>> oneStream = new OutputTag<Tuple3<Integer,Integer,Integer>>("oneStream") {};

    SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> processStream= items.process(new ProcessFunction<Tuple3<Integer, Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {

        @Override

        public void processElement(Tuple3<Integer, Integer, Integer> value, Context ctx, Collector<Tuple3<Integer, Integer, Integer>> out) throws Exception {

            if (value.f0 == 0) {

                ctx.output(zeroStream, value);

            } else if (value.f0 == 1) {

                ctx.output(oneStream, value);

            }

        }

    });

    DataStream<Tuple3<Integer, Integer, Integer>> zeroSideOutput = processStream.getSideOutput(zeroStream);

    DataStream<Tuple3<Integer, Integer, Integer>> oneSideOutput = processStream.getSideOutput(oneStream);

    zeroSideOutput.print();

    oneSideOutput.printToErr();

    //打印结果

    String jobName = "user defined streaming source";

    env.execute(jobName);

}

可以看到,我们将流进行了拆分,并且成功打印出了结果。这里要注意,Flink 最新提供的 SideOutPut 方式拆分流是可以多次进行拆分的,无需担心会爆出异常。

image (13).png

总结

这一课时我们讲解了 Flink 的一个小的知识点,是我们生产实践中经常遇到的场景,Flink 在最新的版本中也推荐我们使用 SideOutPut 进行流的拆分。

关注公众号:大数据技术派,回复资料,领取1024G资料。

标签: Flink
最后更新:2022年1月27日

等待下一个秋

待我代码写成,便娶你为妻!专注于Hadoop/Spark/Flink/Hive/数据仓库等,关注公众号:大数据技术派,获取更多学习资料。

打赏 点赞
< 上一篇
下一篇 >

文章评论

取消回复

等待下一个秋

待我代码写成,便娶你为妻!专注于Hadoop/Spark/Flink/Hive/数据仓库等,关注公众号:大数据技术派,获取更多学习资料。

搜一搜
微信
最新 热点 随机
最新 热点 随机
logstash同步mysql数据到elasticsearch Spring IOC 容器源码分析 elasticsearch修改字段类型 curl操作elasticsearch常用命令 Python通过orm操作mysql数据库 Python进程管理——Supervisor
Spring IOC 容器源码分析logstash同步mysql数据到elasticsearch
Flink实时计算topN热榜 程序员大佬都创业了 黑客与画家 前言 黑客与画家 3.不能说的话 十大经典排序算法 十大经典排序算法——基数排序
标签聚合
Redis mysql 算法 挣钱 Java 数据仓库 书籍 Flink 大数据 R语言 Hive Python
文章归档
  • 2022年7月
  • 2022年6月
  • 2022年5月
  • 2022年4月
  • 2022年3月
  • 2022年2月
  • 2022年1月
  • 2021年12月
  • 2021年11月
  • 2021年10月
  • 2021年9月
  • 2021年8月
  • 2021年6月
  • 2021年5月
  • 2021年4月
  • 2021年3月
  • 2021年2月
  • 2021年1月
  • 2020年12月
  • 2020年11月
  • 2020年10月
  • 2020年9月
  • 2020年8月
  • 2020年7月
  • 2020年5月
  • 2020年4月
  • 2020年1月
  • 2019年9月
  • 2019年8月
  • 2019年7月
  • 2019年6月
  • 2019年5月
  • 2019年4月
  • 2019年3月
  • 2019年1月
  • 2018年12月
  • 2017年5月

©2022 ikeguang.com. 保留所有权利。

鄂ICP备2020019097号-1

鄂公网安备 42032202000160号