大数据-玩转数据-Flink-Transform

一、Transform

转换算子可以把一个或多个DataStream转成一个新的DataStream.程序可以把多个复杂的转换组合成复杂的数据流拓扑.
在这里插入图片描述

二、基本转换算子

2.1、map（映射）

将数据流中的数据进行转换, 形成新的数据流，消费一个元素并产出一个元素

package com.lyh.flink05;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;public class Transform_map {public static void main(String[] args) throws Exception {ExecutionEnvironment svn = ExecutionEnvironment.getExecutionEnvironment();DataSource<Integer> s_num = svn.fromElements(1, 2, 3, 4, 5);s_num.map(new MapFunction<Integer, Integer>() {@Overridepublic Integer map(Integer values) throws Exception {return values*values;}}).print();}
}

2.2、filter（过滤）

根据指定的规则将满足条件（true）的数据保留，不满足条件(false)的数据丢弃

package com.lyh.flink05;import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;public class Transform_filter {public static void main(String[] args) throws Exception {ExecutionEnvironment svn = ExecutionEnvironment.getExecutionEnvironment();DataSource<Integer> elements = svn.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9);
//        elements.filter(new FilterFunction<Integer>() {
//            @Override
//            public boolean filter(Integer integer) throws Exception {
//                if (integer % 2 == 0 )
//                        return false;
//                        else {
//                            return true;
//                }
//            }
//        }).print();elements.filter(value -> value % 2 != 0).print();}
}

2.3、flatMap（扁平映射）

消费一个元素并产生零个或多个元素

package com.lyh.flink05;import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.util.Collector;import java.util.concurrent.ExecutionException;public class flatMap {public static void main(String[] args) throws Exception {ExecutionEnvironment svn = ExecutionEnvironment.getExecutionEnvironment();DataSource<Integer> dataSource = svn.fromElements(1, 2, 3);dataSource.flatMap(new FlatMapFunction<Integer, Integer>() {@Overridepublic void flatMap(Integer integer, Collector<Integer> collector) throws Exception {collector.collect(integer*integer);collector.collect(integer*integer*integer);}}).print();}
}

三、聚合算子

3.1、keyBy（按键分区）

把流中的数据分到不同的分区(并行度)中.具有相同key的元素会分到同一个分区中

package com.lyh.flink05;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class keyBy_s {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)env.fromElements(Tuple2.of(2L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(2L, 4L), Tuple2.of(1L, 2L)).keyBy(0) // 以数组的第一个元素作为key.map((MapFunction<Tuple2<Long, Long>, String>) longLongTuple2 -> "key:" + longLongTuple2.f0 + ",value:" + longLongTuple2.f1).print();env.execute("execute");}
}

3.2、sum,min,max,minBy,maxBy（简单聚合）

KeyedStream的每一个支流做聚合。执行完成后，会将聚合的结果合成一个流返回，所以结果都是DataStream

package com.lyh.flink05;import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class keyBy_s {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// this can be used in a streaming program like this (assuming we have a StreamExecutionEnvironment env)env.fromElements(Tuple2.of(2L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L), Tuple2.of(2L, 4L), Tuple2.of(1L, 2L)).keyBy(0) // 以数组的第一个元素作为key.sum(1).print();env.execute("execute");}
}

3.3、reduce（归约聚合）

一个分组数据流的聚合操作，合并当前的元素和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是只返回最后一次聚合的最终结果。
在这里插入图片描述

package com.lyh.flink05;import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class keyByReduce {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);env.fromElements(Tuple2.of(1L,2L),Tuple2.of(2L,4L),Tuple2.of(2L,9L),Tuple2.of(1L,9L),Tuple2.of(1L,2L),Tuple2.of(2L,3L)).keyBy(0).reduce(new ReduceFunction<Tuple2<Long, Long>>() {@Overridepublic Tuple2<Long, Long> reduce(Tuple2<Long, Long> values1, Tuple2<Long, Long> values2) throws Exception {return new Tuple2<>(values1.f0,values1.f1+values2.f1);}}).print();env.execute();}
}

3.4、process（底层处理）

process算子在Flink算是一个比较底层的算子, 很多类型的流上都可以调用, 可以从流中获取更多的信息(不仅仅数据本身)，process 用法比较灵活，后面再做专门研究。

package com.lyh.flink05;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
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 Process_s {public static void main(String[] args) throws Exception {// 获取环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<Integer> streamSource = env.fromElements(1, 2, 3, 4, 5);SingleOutputStreamOperator<Integer> processed = streamSource.process(new ProcessFunction<Integer, Integer>() {@Overridepublic void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {if (value % 3 == 0) {//测流数据ctx.output(new OutputTag<Integer>("3%0", TypeInformation.of(Integer.class)), value);}if (value % 3 == 1) {//测流数据ctx.output(new OutputTag<Integer>("3%1", TypeInformation.of(Integer.class)), value);}//主流 ,数据out.collect(value);}});DataStream<Integer> output0 = processed.getSideOutput(new OutputTag<>("3%0",TypeInformation.of(Integer.class)));DataStream<Integer> output1 = processed.getSideOutput(new OutputTag<>("3%1",TypeInformation.of(Integer.class)));output1.print();env.execute();}
}

四、合流算子

4.1、connect（连接）

在某些情况下，我们需要将两个不同来源的数据流进行连接，实现数据匹配，比如订单支付和第三方交易信息，这两个信息的数据就来自于不同数据源，连接后，将订单支付和第三方交易信息进行对账，此时，才能算真正的支付完成。
Flink中的connect算子可以连接两个保持他们类型的数据流，两个数据流被connect之后，只是被放在了一个同一个流中，内部依然保持各自的数据和形式不发生任何变化，两个流相互独立。

package com.lyh.flink05;import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.common.protocol.types.Field;public class connect_s {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<Integer> data1 = env.fromElements(1, 2, 3, 4, 5);DataStreamSource<String> data2 = env.fromElements("a", "b", "c");ConnectedStreams<Integer, String> data3 = data1.connect(data2);data3.getFirstInput().print("data1");data3.getSecondInput().print("data2");env.execute();}
}

4.2、union（合并）

package com.lyh.flink05;import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class union_s {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(1);DataStreamSource<Integer> data1 = env.fromElements(1, 2, 3);DataStreamSource<Integer> data2 = env.fromElements(555,666);DataStreamSource<Integer> data3 = env.fromElements(999);DataStream<Integer> data = data1.union(data2).union(data3);data.print();env.execute();}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/28204.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Jenkins build包时虽然单元测试失败了，但是仍然可以成功build包（最终结束时build success）

Jenkins build包时虽然单元测试失败了，但是仍然可以成功build包（最终结束时build success）

1.尝试方案1&#xff1a; 尽管单元测试失败&#xff0c;Jenkins Maven仍然可以获得成功-Java 学习之路 将 -Dmaven.test.failure.ignorefalse 添加到 MAVEN_OPTS artifactoryMaven {goals "-U clean install -Dmaven.test.skipfalse -DallowSnapshotstrue -Dmaven.te…
阅读更多...
R119772795 记录详情

R119772795 记录详情

题目描述 在奶牛回家休息和娱乐之前&#xff0c;Farmer John 希望它们通过玩游戏获得一些智力上的刺激。游戏板由 n 个相同的孔组成&#xff0c;这些孔最初都是空的。一头母牛要么用石头盖住一个洞&#xff0c;要么揭开一个先前被盖住的洞。游戏状态的定义是哪些洞被石头覆盖&…
阅读更多...
Unity 3D中使用tilemap创建关卡地图，瓦片间隙有漏缝

Unity 3D中使用tilemap创建关卡地图，瓦片间隙有漏缝

我们使用一张图片来作为Sprite图集&#xff0c;创建地形图&#xff1a; 运行后&#xff0c;会发现&#xff0c;瓦片之间似乎总是有间距。 检查了图片发现&#xff0c;并不是图片边界存在间隙。 最后发现问题是出在图片资源中的线性过滤属性值&#xff1a; 在设计界面就能够看…
阅读更多...
ETHERCAT转CANOPEN协议网关canopen分类

ETHERCAT转CANOPEN协议网关canopen分类

1.产品功能 JM-ECT-COP是自主研发的一款 ETHERCAT 从站功能的通讯网关。该产品主要功能是将 ETHERCAT 网络和 CANOPEN 网络连接起来。 本网关连接到 ETHERCAT 总线中做为从站使用&#xff0c;连接到 CANOPEN 总线中做为主站使用。 3.技术参数 ETHERCAT 技术参数 网关做为 E…
阅读更多...
Kotlin理解内置函数

Kotlin理解内置函数

目录 一 内置函数1.1 apply 函数1.2 let 函数1.3 run函数1.4 with函数1.5 also函数1.6 takeIf函数1.7 takeUnless函数1.8 总结 Kotlin内置函数包括&#xff1a;let、run、with、apply、also&#xff0c;这些函数都是在Any类中定义的扩展函数&#xff0c;所以任何对象都可以调用…
阅读更多...
【设计模式——学习笔记】23种设计模式——中介者模式Observer（原理讲解+应用场景介绍+案例介绍+Java代码实现）

【设计模式——学习笔记】23种设计模式——中介者模式Observer（原理讲解+应用场景介绍+案例介绍+Java代码实现）

文章目录 案例引入案例一普通实现中介者模式 案例二 介绍基础介绍登场角色尚硅谷 《图解设计模式》 案例实现案例一&#xff1a;智能家庭类图实现 案例二&#xff1a;登录页面逻辑实现说明类图实现 总结文章说明 案例引入 案例一 普通实现 在租房过程中&#xff0c;客户可能…
阅读更多...
【BASH】回顾与知识点梳理（十一）

【BASH】回顾与知识点梳理（十一）

【BASH】回顾与知识点梳理 十一 十一. 八至十章知识点总结及练习11.1 总结11.2 练习情境模拟题一&#xff1a;透过 grep 搜寻特殊字符串&#xff0c;并配合数据流重导向来处理大量的文件搜寻问题。情境模拟题二&#xff1a;使用管线命令配合正规表示法建立新指令与新变量。 该系…
阅读更多...
ftp访问ubuntu文件系统

ftp访问ubuntu文件系统

ftp访问ubuntu文件系统 安装vsftpd服务器 sudo apt-get install vsftpd启动ftp服务 sudo service vsftpd start编辑vsftdp的配置文件 sudo vim /etc/vsftpd.conf找到write_enable字段并修改， 设定可以进行写操作，保存并退出 write_enable=YES从新启动ftp服务
阅读更多...
PHP最简单自定义自己的框架定义常量自动生成目录（三）

PHP最简单自定义自己的框架定义常量自动生成目录（三）

1、框架入口增加模块定义&#xff0c;实现多模块功能 index.php 定义模块 <?php //定义当前请求模块 define("MODULE",index); require "./core/KJ.php"; 创建后台模块admin.php <?php define("MODULE",admin); require "./cor…
阅读更多...
每日一题——最小花费爬楼梯

每日一题——最小花费爬楼梯

题目 给定一个整数数组 cost &#xff0c;其中 cost[i] 是从楼梯第i 个台阶向上爬需要支付的费用&#xff0c;下标从0开始。一旦你支付此费用&#xff0c;即可选择向上爬一个或者两个台阶。 你可以选择从下标为 0 或下标为 1 的台阶开始爬楼梯。 请你计算并返回达到楼梯顶部的…
阅读更多...
计算机网络（8） --- IP与IP协议

计算机网络（8） --- IP与IP协议

计算机网络&#xff08;7&#xff09; --- UDP协议和TCP协议_哈里沃克的博客-CSDN博客UDP协议和TCP协议https://blog.csdn.net/m0_63488627/article/details/132125374?spm1001.2014.3001.5501 目录 1.IP与IP协议 IP作用 协议​编辑 2.网段划分 DHCP划分 CIDR划分 特殊…
阅读更多...
ubuntu下载安装libevent

ubuntu下载安装libevent

ubuntu下载安装libevent libevent 官网上下载最新稳定版本&#xff0c;然后拖到你的linux系统中&#xff0c;解压。或者可是使用wget在线下载。本文下载的是最新版本&#xff0c;其他版本同操作 tar xf libevent-2.1.12-stable.tar.gz 进入解压后的目录&#xff0c;执行 ./con…
阅读更多...
yarn 迁移 pnpm 记录

yarn 迁移 pnpm 记录

迁移步骤 1.全局安装pnpm npm -g i pnpm2. 使用 pnpm import 由yarn.lock导出pnpm.lock 3. 添加 script 到 package.json 确保pnpm唯一包管理 "scripts": {"preinstall": "npx only-allow pnpm", ... }4. 更换.yarnrc 为 .npmrc 5. 运行 p…
阅读更多...
Java基础入门篇——赋值运算符（九）

Java基础入门篇——赋值运算符（九）

赋值运算符的作用是将一个数值、变量或者表达式的值赋给某一个变量。在赋值过程中&#xff0c;运算顺序从右往左&#xff0c;将右边表达式的结果赋值给左边的变量。 一、赋值运算符 【】赋值&#xff0c;用法&#xff1a;a10&#xff0c;将10赋值给变量a&#xff1b; 【】相…
阅读更多...
ETCD备份还原

ETCD备份还原

备份只需要在一个节点上备就可以了&#xff0c;每个节点上的数据是同步的&#xff1b;但是数据恢复是需要在每个主节点上进行 1.查看证书位置 #查看etcd证书 [rootk8s-master01 manifests]# cat /etc/kubernetes/manifests/kube-apiserver.yaml |grep etcd- --etcd-cafile/et…
阅读更多...
【果树农药喷洒机器人】Part5：基于深度相机与分割掩膜的果树冠层体积探测方法

【果树农药喷洒机器人】Part5：基于深度相机与分割掩膜的果树冠层体积探测方法

文章目录 一、引言二、树冠体积测量对比方法2.1冠层体积人工测量法2.2冠层体积拟合测量法 三、基于深度相机与分割掩膜探测树冠体积方法3.1像素值与深度值的转换3.2树冠体积视觉探测法3.3实验分析 总结 一、引言 果树靶标探测是实现农药精准喷施的关键环节&#xff0c;本章以果…
阅读更多...
Syncfusion Essential Edit for WPF Crack

Syncfusion Essential Edit for WPF Crack

Syncfusion Essential Edit for WPF Crack 在任何WPF应用程序中启用语法高亮显示。 Syncfusion Essential Edit for WPF是一款具有所有基本功能的编辑器&#xff0c;如文本编辑、剪切、复制和粘贴。它允许用户从各种文件格式打开文件并将其保存为各种文件格式。Syncfusion Esse…
阅读更多...
java.lang.IllegalStateException: 不允许修改锁定的参数映射

java.lang.IllegalStateException: 不允许修改锁定的参数映射

问题描述&#xff1a; 这个问题是在我写javaweb项目&#xff0c;做敏感词过滤时出现的 需求是&#xff1a;如果是由 getParameter(String s) 得到的数据&#xff0c;可以直接修改value值&#xff0c;将含有敏感词的部分替换为 *** request.getParameterMap() 方法返回一个包…
阅读更多...
2023牛客暑期多校第三场部分题解

2023牛客暑期多校第三场部分题解

索引 ABDEGHIJ A 直接输出两个数的差即可。再判一下无解的情况。 B 其实思路还挺顺的&#xff0c;首先拿的牌肯定是一段增一段减一段增一段减……的序列&#xff0c;并且 > n >n >n 的开头和 ≤ n \leq n ≤n 的开头两种序列是对称的&#xff0c;我们随便考虑一种…
阅读更多...
Windows命名管道学习记录

Windows命名管道学习记录

前言 编程菜鸟&#xff0c;第一次接触进程间通信&#xff0c;网上windows命名管道的资料太少了&#xff0c;很多还是错的&#xff0c;学起来很费劲&#xff0c;现在也只能说跑的通程序了&#xff0c;记录一下防止忘记&#xff0c;理解没有很深&#xff0c;可能会有错误&#x…
阅读更多...
最新文章

Copyright @ 2022~2023