【Flink-1.17-教程】-【四】Flink DataStream API(3)转换算子(Transformation)【用户自定义函数(UDF)】

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

用户自定义函数(user-defined functionUDF),即用户可以根据自身需求,重新实现算子的逻辑。

用户自定义函数分为:函数类匿名函数富函数类

1)函数类(Function Classes)

Flink 暴露了所有 UDF 函数的接口,具体实现方式为接口或者抽象类,例如 MapFunctionFilterFunctionReduceFunction 等。所以用户可以自定义一个函数类,实现对应的接口。

需求:用来从用户的点击数据中筛选包含“sensor_1”的内容:

方式一:实现 FilterFunction 接口

public class TransFunctionUDF {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1, 1),
                new WaterSensor("sensor_1", 2, 2),
                new WaterSensor("sensor_2", 2, 2),
                new WaterSensor("sensor_3", 3, 3)
        );
        DataStream<String> filter = stream.filter(new UserFilter());
        filter.print();
        env.execute();
    }

    public static class UserFilter implements
            FilterFunction<WaterSensor> {
        @Override
        public boolean filter(WaterSensor e) throws Exception {
            return e.id.equals("sensor_1");
        }
    }
}

方式二:通过匿名类来实现 FilterFunction 接口

DataStream<String> stream = stream.filter(new FilterFunction<
            WaterSensor>() {
        @Override
        public boolean filter(WaterSensor e) throws Exception {
            return e.id.equals("sensor_1");
        }
    });

方式二的优化:为了类可以更加通用,我们还可以将用于过滤的关键字"home"抽象出来作为类的属性,调用构造方法时传进去

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1, 1),
                new WaterSensor("sensor_1", 2, 2),
                new WaterSensor("sensor_2", 2, 2),
                new WaterSensor("sensor_3", 3, 3)
        );
        DataStream<String> stream = stream.filter(new
                FilterFunctionImpl("sensor_1"));
        public static class FilterFunctionImpl implements
                FilterFunction<WaterSensor> {
            private String id;

            FilterFunctionImpl(String id) {
                this.id = id;
            }

            @Override
            public boolean filter(WaterSensor value) throws Exception {
                return thid.id.equals(value.id);
            }
        }
    }

方式三:采用匿名函数(Lambda)

public class TransFunctionUDF {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1, 1),
                new WaterSensor("sensor_1", 2, 2),
                new WaterSensor("sensor_2", 2, 2),
                new WaterSensor("sensor_3", 3, 3)
        );
//map 函数使用 Lambda 表达式,不需要进行类型声明
        SingleOutputStreamOperator<String> filter =
                stream.filter(sensor -> "sensor_1".equals(sensor.id));
        filter.print();
        env.execute();
    }
}

2)富函数类(Rich Function Classes)

“富函数类”也是 DataStream API 提供的一个函数类的接口,所有的 Flink 函数类都有其 Rich 版 本 。 富函数类一般是以抽象类的形式出现的。例如:RichMapFunctionRichFilterFunctionRichReduceFunction 等。

与常规函数类的不同主要在于,富函数类可以获取运行环境的上下文,并拥有一些生命周期方法,所以可以实现更复杂的功能。

Rich Function 有生命周期的概念。典型的生命周期方法有:

  • open() 方法,是 Rich Function 的初始化方法,也就是会开启一个算子的生命周期。当一个算子的实际工作方法例如 map() 或者 filter() 方法被调用之前,open() 会首先被调用。

  • close() 方法,是生命周期中的最后一个调用的方法,类似于结束方法。一般用来做一些清理工作。

需要注意的是,这里的生命周期方法,对于一个并行子任务来说只会调用一次;而对应的,实际工作方法,例如 RichMapFunction 中的 map(),在每条数据到来后都会触发一次调用。

来看一个例子说明:文章来源地址https://www.toymoban.com/news/detail-818829.html

public class RichFunctionExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        env
                .fromElements(1, 2, 3, 4)
                .map(new RichMapFunction<Integer, Integer>() {
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        System.out.println(" 索 引 是 : " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期开始");
                    }

                    @Override
                    public Integer map(Integer integer) throws
                            Exception {
                        return integer + 1;
                    }

                    @Override
                    public void close() throws Exception {
                        super.close();
                        System.out.println(" 索 引 是 : " + getRuntimeContext().getIndexOfThisSubtask() + " 的任务的生命周期结束");
                    }
                })
                .print();
        env.execute();
    }
}

到了这里,关于【Flink-1.17-教程】-【四】Flink DataStream API(3)转换算子(Transformation)【用户自定义函数(UDF)】的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索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】DataStream API使用之源算子(Source)

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

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

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

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

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

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

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

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

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

    2024年01月16日
    浏览(48)
  • 【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

领红包