【Flink-1.17-教程】-【四】Flink DataStream API(5)转换算子(Transformation)【分流】

这篇具有很好参考价值的文章主要介绍了【Flink-1.17-教程】-【四】Flink DataStream API(5)转换算子(Transformation)【分流】。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

所谓“分流”,就是将一条数据流拆分成完全独立的两条、甚至多条流。也就是基于一个 DataStream,定义一些筛选条件,将符合条件的数据拣选出来放到对应的流里。

1)使用 filter 简单实现

其实根据条件筛选数据的需求,本身非常容易实现:只要针对同一条流多次独立调用 .filter() 方法进行筛选,就可以得到拆分之后的流了。

案例需求:读取一个整数数字流,将数据流划分为奇数流和偶数流。

public class SplitByFilterDemo {
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);

        DataStreamSource<String> socketDS = env.socketTextStream("hadoop102", 7777);

        /**
         * TODO 使用filter来实现分流效果
         * 缺点: 同一个数据,要被处理两遍(调用两次filter)
          */

        SingleOutputStreamOperator<String> even = socketDS.filter(value -> Integer.parseInt(value) % 2 == 0);
        SingleOutputStreamOperator<String> odd = socketDS.filter(value -> Integer.parseInt(value) % 2 == 1);


        even.print("偶数流");
        odd.print("奇数流");


        env.execute();
    }
}

这种实现非常简单,但代码显得有些冗余——我们的处理逻辑对拆分出的三条流其实是一样的,却重复写了三次。而且这段代码背后的含义,是将原始数据流 stream 复制三份,然后对每一份分别做筛选;这明显是不够高效的。我们自然想到,能不能不用复制流,直接用一个算子就把它们都拆分开呢?

2)使用侧输出流实现

关于处理函数中侧输出流的用法,我们已经在 flatmap 课节做了详细介绍。简单来说,只需要调用上下文 ctx.output() 方法,就可以输出任意类型的数据了。而侧输出流的标记和提取,都离不开一个“输出标签”OutputTag),指定了侧输出流的 id 和类型。

代码实现:将 WaterSensor 按照 id 类型进行分流。

准备好自定义的 MapFunction:

public class WaterSensorMapFunction implements MapFunction<String,WaterSensor> {
    @Override
    public WaterSensor map(String value) throws Exception {
        String[] datas = value.split(",");
        return new WaterSensor(datas[0], Long.valueOf(datas[1]), Integer.valueOf(datas[2]));
    }
}

实现:

public class SideOutputDemo {
    public static void main(String[] args) throws Exception {
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("hadoop102", 7777)
                .map(new WaterSensorMapFunction());


        /**
         * TODO 使用侧输出流 实现分流
         * 需求: watersensor的数据,s1、s2的数据分别分开
         *
         * TODO 总结步骤:
         *    1、使用 process算子
         *    2、定义 OutputTag对象
         *    3、调用 ctx.output
         *    4、通过主流 获取 测流
         */

        /**
         * 创建OutputTag对象
         * 第一个参数: 标签名
         * 第二个参数: 放入侧输出流中的 数据的 类型,Typeinformation
         */
        OutputTag<WaterSensor> s1Tag = new OutputTag<>("s1", Types.POJO(WaterSensor.class));
        OutputTag<WaterSensor> s2Tag = new OutputTag<>("s2", Types.POJO(WaterSensor.class));

        SingleOutputStreamOperator<WaterSensor> process = sensorDS
                .process(
                        new ProcessFunction<WaterSensor, WaterSensor>() {
                            @Override
                            public void processElement(WaterSensor value, Context ctx, Collector<WaterSensor> out) throws Exception {
                                String id = value.getId();
                                if ("s1".equals(id)) {
                                    // 如果是 s1,放到侧输出流s1中
                                    /**
                                     * 上下文ctx 调用ouput,将数据放入侧输出流
                                     * 第一个参数: Tag对象
                                     * 第二个参数: 放入侧输出流中的 数据
                                     */
                                    ctx.output(s1Tag, value);
                                } else if ("s2".equals(id)) {
                                    // 如果是 s2,放到侧输出流s2中

                                    ctx.output(s2Tag, value);
                                } else {
                                    // 非s1、s2的数据,放到主流中
                                    out.collect(value);
                                }

                            }
                        }
                );
        // 从主流中,根据标签 获取 侧输出流
        SideOutputDataStream<WaterSensor> s1 = process.getSideOutput(s1Tag);
        SideOutputDataStream<WaterSensor> s2 = process.getSideOutput(s2Tag);

        // 打印主流
        process.print("主流-非s1、s2");

        //打印 侧输出流
        s1.printToErr("s1");
        s2.printToErr("s2");

        env.execute();
    }
}

要点:

1、使用 process();(是最底层 API)。

2、process 每次处理一条数据

3、定义 OutputTag 对象:

(1)第一个参数:标签名

(2)第二个参数:放入侧输出流中的数据的类型,Typeinformation

4、调用 ctx.output();

5、通过主流获取测流文章来源地址https://www.toymoban.com/news/detail-820048.html

到了这里,关于【Flink-1.17-教程】-【四】Flink DataStream API(5)转换算子(Transformation)【分流】的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 大数据学习之Flink算子、了解(Transformation)转换算子(基础篇三)

    目录 Transformation转换算子(基础篇三) 三、转换算子(Transformation) 1.基本转换算子 1.1 映射(Map) 1.2 过滤(filter) 1.3 扁平映射(flatmap) 1.4基本转换算子的例子 2.聚合算子(Aggregation) 2.1 按键分区(keyBy) 2.2 简单聚合 2.3 归约聚合(reduce) 3.用户自定义函数(UDF) 3.1 函

    2024年02月20日
    浏览(40)
  • flink重温笔记(三):Flink 流批一体 API 开发——Transformation 重要算子操作

    前言:今天是学习 flink 第三天啦,学习了高级 api 开发中11 中重要算子,查找了好多资料理解其中的原理,以及敲了好几个小时代码抓紧理解原理。 Tips:虽然学习进度有点慢,希望自己继续努力,不断猜想 api 原理,通过敲代码不断印证自己的想法,转码大数据之路一定会越

    2024年02月19日
    浏览(42)
  • Flink核心API之DataStream(基础常用算子)

    (一)Flink核心API Flink中提供了4种不同层次的API,每种API在简洁和易表达之间有自己的权衡,适用于不同的场景。目前上面3个会用得比较多。 低级API(Stateful Stream Processing):提供了对时间和状态的细粒度控制,简洁性和易用性较差, 主要应用在一些复杂事件处理逻辑上。 核心

    2024年01月22日
    浏览(43)
  • 【Flink】DataStream API使用之源算子(Source)

    创建环境之后,就可以构建数据的业务处理逻辑了,Flink可以从各种来源获取数据,然后构建DataStream进项转换。一般将数据的输入来源称为数据源(data source),而读取数据的算子就叫做源算子(source operator)。所以,Source就是整个程序的输入端。 Flink中添加source的方式,是

    2024年02月10日
    浏览(36)
  • Flink|《Flink 官方文档 - DataStream API - 算子 - 窗口》学习笔记

    学习文档:《Flink 官方文档 - DataStream API - 算子 - 窗口》 学习笔记如下: 窗口(Window):窗口是处理无界流的关键所在。窗口可以将数据流装入大小有限的 “桶” 中,再对每个 “桶” 加以处理。 Keyed Windows 在 Keyed Windows 上使用窗口时,要调用 keyBy(...) 而后再调用 window(..

    2024年01月18日
    浏览(44)
  • 大数据学习之Flink算子、了解DataStream API(基础篇一)

    注: 本文只涉及DataStream 原因:随着大数据和流式计算需求的增长,处理实时数据流变得越来越重要。因此,DataStream由于其处理实时数据流的特性和能力,逐渐替代了DataSet成为了主流的数据处理方式。 目录 DataStream API (基础篇) 前摘: 一、执行环境 1. 创建执行环境 2. 执

    2024年01月23日
    浏览(51)
  • 【Flink-1.17-教程】-【一】Flink概述、Flink快速入门

    在准备好所有的开发环境之后,我们就可以开始开发自己的第一个 Flink 程序了。首先我们要做的,就是在 IDEA 中搭建一个 Flink 项目的骨架。我们会使用 Java 项目中常见的 Maven 来进行依赖管理。 1、创建工程 (1)打开 IntelliJ IDEA,创建一个 Maven 工程。 (2)将这个 Maven 工程命

    2024年01月16日
    浏览(47)
  • 【Flink-1.17-教程】-【二】Flink 集群搭建、Flink 部署、Flink 运行模式

    集群规划: 具体安装部署步骤如下: 1、下载并解压安装包 (1)下载安装包 flink-1.17.0-bin-scala_2.12.tgz,将该 jar 包上传到 hadoop102 节点服务器的 /opt/software 路径上。 (2)在 /opt/software 路径上解压 flink-1.17.0-bin-scala_2.12.tgz 到 /opt/module 路径上。 2、修改集群配置 (1)进入 conf 路

    2024年01月18日
    浏览(60)
  • Flink流批一体计算(17):PyFlink DataStream API之StreamExecutionEnvironment

    目录 StreamExecutionEnvironment Watermark watermark策略简介 使用 Watermark 策略 内置水印生成器 处理空闲数据源 算子处理 Watermark 的方式 创建DataStream的方式 通过list对象创建 ​​​​​​使用DataStream connectors创建 使用Table SQL connectors创建 StreamExecutionEnvironment 编写一个 Flink Python DataSt

    2024年02月11日
    浏览(42)
  • Flink复习3-2-4-6-1(v1.17.0): 应用开发 - DataStream API - 状态和容错 - 数据类型&序列化 - 概述

    Apache Flink handles data types and serialization in a unique way, containing its own type descriptors, generic type extraction, and type serialization framework. This document describes the concepts and the rationale behind them. Apache Flink以独特的方式处理数据类型和序列化,包含自己的类型描述符、泛型类型提取和类型序列化

    2024年02月12日
    浏览(49)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包