【flink番外篇】1、flink的23种常用算子介绍及详细示例（1）- map、flatmap和filter
  nNPyvzOmRTFq 2023年12月05日 26 0

Flink 系列文章

1、Flink 专栏等系列综合文章链接

(文章目录)

本文主要介绍Flink 的3种常用的operator（map、flatmap和filter）及以具体可运行示例进行说明. 如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。 本文除了maven依赖外，没有其他依赖。

本专题分为五篇，即： 【flink番外篇】1、flink的23种常用算子介绍及详细示例（1）- map、flatmap和filter 【flink番外篇】1、flink的23种常用算子介绍及详细示例（2）- keyby、reduce和Aggregations 【flink番外篇】1、flink的23种常用算子介绍及详细示例（3）-window、distinct、join等 【flink番外篇】1、flink的23种常用算子介绍及详细示例（4）- union、window join、connect、outputtag、cache、iterator、project 【flink番外篇】1、flink的23种常用算子介绍及详细示例（完整版）

一、Flink的23种算子说明及示例

1、maven依赖

下文中所有示例都是用该maven依赖，除非有特殊说明的情况。

<properties>
        <encoding>UTF-8</encoding>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <java.version>1.8</java.version>
        <scala.version>2.12</scala.version>
        <flink.version>1.17.0</flink.version>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- 日志 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.7</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.2</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-common</artifactId>
			<version>3.1.4</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-client</artifactId>
			<version>3.1.4</version>
		</dependency>
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-hdfs</artifactId>
			<version>3.1.4</version>
		</dependency>
    </dependencies>

2、java bean

下文所依赖的User如下

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author alanchan
 *
 */
@Data
@AllArgsConstructor
@NoArgsConstructor
public class User {
	private int id;
	private String name;
	private String pwd;
	private String email;
	private int age;
	private double balance;
}

3、map

[DataStream->DataStream] 这是最简单的转换之一，其中输入是一个数据流，输出的也是一个数据流。 在这里插入图片描述

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
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.datastreamapi.User;

/**
 * @author alanchan
 *
 */
public class TestMapDemo {

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		// env
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		// source

		// transformation
		mapFunction5(env);
		// sink
		// execute
		env.execute();
	}

	// 构造一个list，然后将list中数字乘以2输出，内部匿名类实现
	public static void mapFunction1(StreamExecutionEnvironment env) throws Exception {

		List<Integer> data = new ArrayList<Integer>();
		for (int i = 1; i <= 10; i++) {
			data.add(i);
		}
		DataStreamSource<Integer> source = env.fromCollection(data);

		SingleOutputStreamOperator<Integer> sink = source.map(new MapFunction<Integer, Integer>() {

			@Override
			public Integer map(Integer inValue) throws Exception {
				return inValue * 2;
			}
		});

		sink.print();
//		9> 12
//		4> 2
//		10> 14
//		8> 10
//		13> 20
//		7> 8
//		12> 18
//		11> 16
//		5> 4
//		6> 6
	}

	// 构造一个list，然后将list中数字乘以2输出，lambda实现
	public static void mapFunction2(StreamExecutionEnvironment env) throws Exception {
		List<Integer> data = new ArrayList<Integer>();
		for (int i = 1; i <= 10; i++) {
			data.add(i);
		}
		DataStreamSource<Integer> source = env.fromCollection(data);
		SingleOutputStreamOperator<Integer> sink = source.map(i -> 2 * i);
		sink.print();
//		3> 4
//		4> 6
//		9> 16
//		7> 12
//		10> 18
//		2> 2
//		6> 10
//		5> 8
//		8> 14
//		11> 20
	}

	// 构造User数据源
	public static DataStreamSource<User> source(StreamExecutionEnvironment env) {
		DataStreamSource<User> source = env.fromCollection(Arrays.asList(
				new User(1, "alan1", "1", "1@1.com", 12, 1000), 
				new User(2, "alan2", "2", "2@2.com", 19, 200),
				new User(3, "alan1", "3", "3@3.com", 28, 1500), 
				new User(5, "alan1", "5", "5@5.com", 15, 500), 
				new User(4, "alan2", "4", "4@4.com", 30, 400)
				)
			);
		return source;
	}

	// lambda实现用户对象的balance×2和age+5功能
	public static SingleOutputStreamOperator<User> mapFunction3(StreamExecutionEnvironment env) throws Exception {
		DataStreamSource<User> source = source(env);

		SingleOutputStreamOperator<User> sink = source.map((MapFunction<User, User>) user -> {
			User user2 = user;
			user2.setAge(user.getAge() + 5);
			user2.setBalance(user.getBalance() * 2);

			return user2;
		});
		sink.print();
//		10> User(id=1, name=alan1, pwd=1, email=1@1.com, age=17, balance=2000.0)
//		14> User(id=4, name=alan2, pwd=4, email=4@4.com, age=35, balance=800.0)
//		11> User(id=2, name=alan2, pwd=2, email=2@2.com, age=24, balance=400.0)
//		12> User(id=3, name=alan1, pwd=3, email=3@3.com, age=33, balance=3000.0)
//		13> User(id=5, name=alan1, pwd=5, email=5@5.com, age=20, balance=1000.0)
		return sink;
	}

	// lambda实现balance*2和age+5后，balance》=2000和age》=20的数据过滤出来
	public static SingleOutputStreamOperator<User> mapFunction4(StreamExecutionEnvironment env) throws Exception {

		SingleOutputStreamOperator<User> sink = mapFunction3(env).filter(user -> user.getBalance() >= 2000 && user.getAge() >= 20);
		sink.print();
//		15> User(id=1, name=alan1, pwd=1, email=1@1.com, age=17, balance=2000.0)
//		1> User(id=3, name=alan1, pwd=3, email=3@3.com, age=33, balance=3000.0)
//		16> User(id=2, name=alan2, pwd=2, email=2@2.com, age=24, balance=400.0)
//		3> User(id=4, name=alan2, pwd=4, email=4@4.com, age=35, balance=800.0)
//		2> User(id=5, name=alan1, pwd=5, email=5@5.com, age=20, balance=1000.0)
//		1> User(id=3, name=alan1, pwd=3, email=3@3.com, age=33, balance=3000.0)
		return sink;
	}

	// lambda实现balance*2和age+5后，balance》=2000和age》=20的数据过滤出来并通过flatmap收集
	public static SingleOutputStreamOperator<User> mapFunction5(StreamExecutionEnvironment env) throws Exception {

		SingleOutputStreamOperator<User> sink = mapFunction4(env).flatMap((FlatMapFunction<User, User>) (user, out) -> {
			if (user.getBalance() >= 3000) {
				out.collect(user);
			}
		}).returns(User.class);

		sink.print();
//		8> User(id=5, name=alan1, pwd=5, email=5@5.com, age=20, balance=1000.0)
//		7> User(id=3, name=alan1, pwd=3, email=3@3.com, age=33, balance=3000.0)
//		6> User(id=2, name=alan2, pwd=2, email=2@2.com, age=24, balance=400.0)
//		9> User(id=4, name=alan2, pwd=4, email=4@4.com, age=35, balance=800.0)
//		5> User(id=1, name=alan1, pwd=1, email=1@1.com, age=17, balance=2000.0)
//		7> User(id=3, name=alan1, pwd=3, email=3@3.com, age=33, balance=3000.0)
//		7> User(id=3, name=alan1, pwd=3, email=3@3.com, age=33, balance=3000.0)
		return sink;
	}

}

4、flatmap

[DataStream->DataStream] FlatMap 采用一条记录并输出零个，一个或多个记录。将集合中的每个元素变成一个或多个元素,并返回扁平化之后的结果。 在这里插入图片描述

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.common.functions.FlatMapFunction;
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.util.Collector;

/**
 * @author alanchan
 *
 */
public class TestFlatMapDemo {

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		flatMapFunction3(env);

		env.execute();
	}

	// 构造User数据源
	public static DataStreamSource<String> source(StreamExecutionEnvironment env) {
		List<String> info = new ArrayList<>();
		info.add("i am alanchan");
		info.add("i like hadoop");
		info.add("i like flink");
		info.add("and you ?");

		DataStreamSource<String> dataSource = env.fromCollection(info);

		return dataSource;
	}

	// 将句子以空格进行分割-内部匿名类实现
	public static void flatMapFunction1(StreamExecutionEnvironment env) throws Exception {

		DataStreamSource<String> source = source(env);
		SingleOutputStreamOperator<String> sink = source.flatMap(new FlatMapFunction<String, String>() {
			@Override
			public void flatMap(String value, Collector<String> out) throws Exception {
				String[] splits = value.split(" ");
				for (String split : splits) {
					out.collect(split);
				}
			}
		});
		sink.print();
//		11> and
//		10> i
//		8> i
//		9> i
//		8> am
//		10> like
//		11> you
//		10> flink
//		8> alanchan
//		9> like
//		11> ?
//		9> hadoop
	}

	// lambda实现
	public static void flatMapFunction2(StreamExecutionEnvironment env) throws Exception {
		DataStreamSource<String> source = source(env);
		SingleOutputStreamOperator<String> sink = source.flatMap((FlatMapFunction<String, String>) (input, out) -> {
			String[] splits = input.split(" ");
			for (String split : splits) {
				out.collect(split);
			}
		}).returns(String.class);

		sink.print();
//		6> i
//		8> and
//		8> you
//		8> ?
//		5> i
//		7> i
//		5> am
//		5> alanchan
//		6> like
//		7> like
//		6> hadoop
//		7> flink
	}

	// lambda实现
	public static void flatMapFunction3(StreamExecutionEnvironment env) throws Exception {
		DataStreamSource<String> source = source(env);
		SingleOutputStreamOperator<String> sink = source.flatMap((String input, Collector<String> out) -> Arrays.stream(input.split(" ")).forEach(out::collect))
				.returns(String.class);

		sink.print();
//		8> i
//		11> and
//		10> i
//		9> i
//		10> like
//		11> you
//		8> am
//		11> ?
//		10> flink
//		9> like
//		9> hadoop
//		8> alanchan
	}

}

5、Filter

DataStream → DataStream Filter 函数根据条件判断出结果。按照指定的条件对集合中的元素进行过滤,过滤出返回true/符合条件的元素。 在这里插入图片描述

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
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.datastreamapi.User;

/**
 * @author alanchan
 *
 */
public class TestFilterDemo {
	// 构造User数据源
	public static DataStreamSource<User> sourceUser(StreamExecutionEnvironment env) {
		DataStreamSource<User> source = env.fromCollection(Arrays.asList(
			new User(1, "alan1", "1", "1@1.com", 12, 1000), 
			new User(2, "alan2", "2", "2@2.com", 19, 200),
			new User(3, "alan1", "3", "3@3.com", 28, 1500), 
			new User(5, "alan1", "5", "5@5.com", 15, 500), 
			new User(4, "alan2", "4", "4@4.com", 30, 400)));
		return source;
	}

	// 构造User数据源
	public static DataStreamSource<Integer> sourceList(StreamExecutionEnvironment env) {
		List<Integer> data = new ArrayList<Integer>();
		for (int i = 1; i <= 10; i++) {
			data.add(i);
		}
		DataStreamSource<Integer> source = env.fromCollection(data);

		return source;
	}

	// 过滤出大于5的数字，内部匿名类
	public static void filterFunction1(StreamExecutionEnvironment env) throws Exception {
		DataStream<Integer> source = sourceList(env);

		SingleOutputStreamOperator<Integer> sink = source.map(new MapFunction<Integer, Integer>() {
			public Integer map(Integer value) throws Exception {
				return value + 1;
			}
		}).filter(new FilterFunction<Integer>() {
			@Override
			public boolean filter(Integer value) throws Exception {
				return value > 5;
			}
		});
		sink.print();
//		1> 10
//		14> 7
//		16> 9
//		13> 6
//		2> 11
//		15> 8
	}

	// lambda实现
	public static void filterFunction2(StreamExecutionEnvironment env) throws Exception {
		DataStream<Integer> source = sourceList(env);
		SingleOutputStreamOperator<Integer> sink = source.map(i -> i + 1).filter(value -> value > 5);
		sink.print();
//		12> 7
//		15> 10
//		11> 6
//		13> 8
//		14> 9
//		16> 11
	}

	// 查询user id大于3的记录
	public static void filterFunction3(StreamExecutionEnvironment env) throws Exception {
		DataStream<User> source = sourceUser(env);
		SingleOutputStreamOperator<User> sink = source.filter(user -> user.getId() > 3);
		sink.print();
//		14> User(id=5, name=alan1, pwd=5, email=5@5.com, age=15, balance=500.0)
//		15> User(id=4, name=alan2, pwd=4, email=4@4.com, age=30, balance=400.0)
	}

	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		filterFunction3(env);
		env.execute();

	}

}

本文主要介绍Flink 的3种常用的operator及以具体可运行示例进行说明。 如果需要了解更多内容，可以在本人Flink 专栏中了解更新系统的内容。

本专题分为五篇，即： 【flink番外篇】1、flink的23种常用算子介绍及详细示例（1）- map、flatmap和filter 【flink番外篇】1、flink的23种常用算子介绍及详细示例（2）- keyby、reduce和Aggregations 【flink番外篇】1、flink的23种常用算子介绍及详细示例（3）-window、distinct、join等 【flink番外篇】1、flink的23种常用算子介绍及详细示例（4）- union、window join、connect、outputtag、cache、iterator、project 【flink番外篇】1、flink的23种常用算子介绍及详细示例（完整版）

【版权声明】本文内容来自摩杜云社区用户原创、第三方投稿、转载，内容版权归原作者所有。本网站的目的在于传递更多信息，不拥有版权，亦不承担相应法律责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@moduyun.com
  1. 分享：
最后一次编辑于 2023年12月05日 0

暂无评论

推荐阅读
  KRe60ogUm4le   3天前   7   0   0 flink大数据
  nNPyvzOmRTFq   2023年12月23日   30   0   0 flinkhiveflinkredisHadoophivekafkakafkaHadoopredis
  nNPyvzOmRTFq   2023年12月23日   38   0   0 大数据flinkhiveflink大数据HadoophivekafkakafkaHadoop
  KRe60ogUm4le   3天前   8   0   0 flink大数据
  nNPyvzOmRTFq   2023年12月23日   36   0   0 flinkhivehbaseflinkHadoophivekafkakafkaHadoopHbase
  nNPyvzOmRTFq   2023年12月23日   46   0   0 大数据flinkflink kafkahiveflink大数据hivekafkakafkaflink kafka
  nNPyvzOmRTFq   2023年12月23日   56   0   0 大数据flinkhiveflink大数据hivekafka数据库kafka数据库
  nNPyvzOmRTFq   2023年12月23日   39   0   0 flinkflink kafkaflink windowflink hiveflinkflink windowflink 大数据flink hiveflink kafkaflink 大数据
  KRe60ogUm4le   2024年04月26日   28   0   0 数据集flink
nNPyvzOmRTFq
nNPyvzOmRTFq
作者其他文章 更多
【flink番外篇】4、flink的sink（内置、mysql、kafka、redis、clickhouse、分布式缓存、广播变量）介绍及示例（1） - File、Socket、console

2023-12-23

【flink番外篇】4、flink的sink（内置、mysql、kafka、redis、clickhouse、分布式缓存、广播变量）介绍及示例（2） - jdbc_mysql

2023-12-23

【flink番外篇】3、flink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例 - 完整版

2023-12-23

【flink番外篇】4、flink的sink（内置、mysql、kafka、redis、clickhouse、分布式缓存、广播变量）介绍及示例（3） - redis

2023-12-23

【flink番外篇】4、flink的sink（内置、mysql、kafka、redis、clickhouse、分布式缓存、广播变量）介绍及示例（5） - kafka

2023-12-23

【flink番外篇】5、flink的window(介绍、分类、函数及Tumbling、Sliding、session窗口应用)介绍及示例（1）- 窗口介绍、分类、函数

2023-12-23

40、Flink 的Apache Kafka connector（kafka sink的介绍及使用示例）-2

2023-12-15

【flink番外篇】2、flink的23种算子window join 和interval join 数据倾斜、分区介绍及详细示例-完整版

2023-12-11

【flink番外篇】3、fflink的source（内置、mysql、kafka、redis、clickhouse）介绍及示例（1） - File、Socket、Collection

2023-12-11

21、Flink 的table API与DataStream API 集成（完整版）

2023-12-11

最新推荐 更多
【Leetcode】排序+双指针-16

2024-05-31

【.net 深呼吸】设置序列化中的最大数据量

2024-05-31

【Leetcode】python哈希表

2024-05-31

【Leetcode】链表-21，23

2024-05-31

vue再读20-表格案例--处理无数据的时候的渲染

2024-05-31

你被请来给一个要举办高尔夫比赛的树林砍树，树林由一个 m x n 的矩阵表示， 在这个矩阵中： 0 表示障碍，无法触碰 1 表示地面，可以行走 比 1 大的数 表示有树的单元格

2024-05-31

假设有一个源源吐出不同球的机器，只有装下10个球的袋子，每一个吐出的球，要么放入袋子，要么永远扔掉。

2024-05-31

比jsonpath 更方便的json 数据查询JMESPath 使用

2024-05-31

已知一棵二叉树上所有的值都不一样， 给定这棵二叉树的头节点head， 给定一个整型数组arr，arr里放着不同的值，每个值一定在树上 返回数组里所有值的最低公共祖先。

2024-05-31

给定一个数组arr，给定一个值v。求子数组平均值小于等于v的最长子数组长度。

2024-05-31

线性回归api深度介绍

2024-05-31

给定一个单链表的头节点head，请判断该链表是否为回文结构。

2024-05-31

力扣546，移除盒子。给出一些不同颜色的盒子，盒子的颜色由数字表示，即不同的数字表示不同的颜色。你将经过若干轮操作去去掉盒子，直到所有的盒子都去掉为止。

2024-05-31

给定一个数组arr，长度为N ＞ 1，从中间切一刀，保证左部分和右部分都有数字，一共有N-1种切法，如此多的切法中，每一种都有:绝对值(左部分最大值 – 右部分最大值)。

2024-05-31

【算法】哨兵思想 -- 待进一步总结

2024-05-31

给定两个可能有环也可能无环的单链表，头节点head1和head2。请实现一个函数，如果两个链表相交，请返回相交的 第一个节点。如果不相交，返回null。

2024-05-31

数据统计之日分类商品访问量

2024-05-31

稀疏数组搜索

2024-05-31

魔术索引（返回索引值最小的一个）

2024-05-31

求表达式 f(n)结果末尾0的个数

2024-05-31

常用标签
数据数据数据数据javalinuxlinuxpythonCentOSredisJava区块链技术python区块链信息系统字符串MySQL资讯项目管理mysql教程MySQL