Flink--API 之Transformation-转换算子的使用解析

目录

一、常用转换算子详解

（一）map 算子

（二）flatMap 算子

（三）filter 算子

（四）keyBy 算子

元组类型

POJO

（五）reduce 算子

二、合并与连接操作

（一）union 算子

（二）connect 算子

三、侧输出流（Side Outputs）

四、总结

        在大数据处理领域，Apache Flink 凭借其强大的流处理和批处理能力备受青睐。而转换算子作为 Flink 编程模型中的关键部分，能够对数据进行灵活多样的处理操作，满足各种复杂业务场景需求。本文将深入介绍 Flink 中常见的转换算子，包括 map、flatMap、filter、keyBy、reduce 等，并结合详细代码示例讲解其使用方法，同时探讨 union、connect 等合并连接操作以及侧输出流等特性，帮助读者全面掌握 Flink 转换算子的精髓。

一、常用转换算子详解

（一）map 算子

功能概述
        map 算子主要用于对输入流中的每个元素进行一对一的转换操作，基于用户自定义的映射逻辑将输入元素转换为新的输出元素。

代码示例
假设我们有一份访问日志数据，格式如下：

86.149.9.216 10001 17/05/2015:10:05:30 GET /presentations/logstash-monitorama-2013/images/github-contributions.png
83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css
83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css
83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css
83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css
83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css
83.149.9.216 10002 17/05/2015:10:06:53 GET /presentations/logstash-monitorama-2013/css/print/paper.css
10.0.0.1 10003 17/05/2015:10:06:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css
10.0.0.1 10003 17/05/2015:10:07:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css
10.0.0.1 10003 17/05/2015:10:08:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css
10.0.0.1 10003 17/05/2015:10:09:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css
10.0.0.1 10003 17/05/2015:10:10:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css
10.0.0.1 10003 17/05/2015:10:16:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css
10.0.0.1 10003 17/05/2015:10:26:53 POST /presentations/logstash-monitorama-2013/css/print/paper.css

        我们要将其转换为一个 LogBean 对象，包含访问 ip、用户 userId、访问时间戳 timestamp、访问方法 method、访问路径 path 等字段。

假如需要用到日期工具类，可以导入lang3包：

<dependency><groupId>org.apache.commons</groupId><artifactId>commons-lang3</artifactId><version>3.12.0</version>
</dependency>

以下是代码实现：


import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
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 java.text.SimpleDateFormat;
import java.util.Date;public class MapDemo {@Data@AllArgsConstructor@NoArgsConstructorstatic class LogBean{String ip;      // 访问ipint userId;     // 用户idlong timestamp; // 访问时间戳String method;  // 访问方法String path;    // 访问路径}public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStream<String> fileStream = env.readTextFile("F:\\BD230801\\FlinkDemo\\datas\\a.log");//3. transformation-数据处理转换// 此处也可以将数据放入到tuple中，tuple可以支持到tuple25DataStream<LogBean> mapStream = fileStream.map(new MapFunction<String, LogBean>() {@Overridepublic LogBean map(String line) throws Exception {String[] arr = line.split(" ");String ip = arr[0];int userId = Integer.valueOf(arr[1]);String createTime = arr[2];// 如何将一个时间字符串变为时间戳// 17/05/2015:10:05:30/*SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");Date date = simpleDateFormat.parse(createTime);long timeStamp = date.getTime();*/// 要想使用这个common.lang3 下的工具类，需要导入包Date date = DateUtils.parseDate(createTime, "dd/MM/yyyy:HH:mm:ss");long timeStamp = date.getTime();String method = arr[3];String path = arr[4];LogBean logBean = new LogBean(ip, userId, timeStamp, method, path);return logBean;}});//4. sink-数据输出mapStream.print();//5. execute-执行env.execute();}
}

        在上述代码中，通过 map 函数的自定义逻辑，将每行日志字符串按空格拆分后，进行相应字段提取与时间戳转换，最终封装成 LogBean 对象输出。

第二个版本：

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.RuntimeExecutionMode;
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 java.text.SimpleDateFormat;
import java.util.Date;@Data
@AllArgsConstructor
class LogBean{private String ip;      // 访问ipprivate int userId;     // 用户idprivate long timestamp; // 访问时间戳private String method;  // 访问方法private String path;    // 访问路径
}
public class Demo04 {// 将数据转换为javaBeanpublic static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStreamSource<String> streamSource = env.readTextFile("datas/a.log");//3. transformation-数据处理转换SingleOutputStreamOperator<LogBean> map = streamSource.map(new MapFunction<String, LogBean>() {@Overridepublic LogBean map(String line) throws Exception {String[] arr = line.split("\\s+");//时间戳转换  17/05/2015:10:06:53String time = arr[2];SimpleDateFormat format = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");Date date = format.parse(time);long timeStamp = date.getTime();return new LogBean(arr[0],Integer.parseInt(arr[1]),timeStamp,arr[3],arr[4]);}});//4. sink-数据输出map.print();//5. execute-执行env.execute();}
}

（二）flatMap 算子

功能概述
        flatMap 算子可以将输入流中的每个元素转换为零个、一个或多个输出元素。它适用于需要对输入元素进行展开、拆分等操作的场景。

代码示例
        假设有数据格式为“张三,苹果手机,联想电脑,华为平板”这样的文本文件，我们要将其转换为“张三有苹果手机”“张三有联想电脑”“张三有华为平板”等形式。

flatmap.log文件如：

张三,苹果手机,联想电脑,华为平板
李四,华为手机,苹果电脑,小米平板

代码如下：

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class FlatMapDemo {public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据//2. source-加载数据DataStream<String> fileStream = env.readTextFile("F:\\BD230801\\FlinkDemo\\datas\\flatmap.log");//3. transformation-数据处理转换DataStream<String> flatMapStream = fileStream.flatMap(new FlatMapFunction<String, String>() {@Overridepublic void flatMap(String line, Collector<String> collector) throws Exception {//张三,苹果手机,联想电脑,华为平板String[] arr = line.split(",");String name = arr[0];for (int i = 1; i < arr.length; i++) {String goods = arr[i];collector.collect(name+"有"+goods);}}});//4. sink-数据输出flatMapStream.print();//5. execute-执行env.execute();}
}

        这里在 flatMap 函数内部，按逗号拆分每行数据，遍历拆分后的数组（除第一个元素作为名称外），通过 collector 将新组合的字符串收集输出。

（三）filter 算子

功能概述
        filter 算子依据用户定义的过滤条件，对输入流元素进行筛选，满足条件的元素继续向下游传递，不满足的则被过滤掉。

代码示例
        读取map算子中的访问日志数据，过滤出访问 IP 是 83.149.9.216 的访问日志，代码实现如下：

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
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.environment.StreamExecutionEnvironment;import java.util.Date;public class FilterDemo {@Data@AllArgsConstructor@NoArgsConstructorstatic class LogBean{String ip;      // 访问ipint userId;     // 用户idlong timestamp; // 访问时间戳String method;  // 访问方法String path;    // 访问路径}public static void main(String[] args) throws Exception {// 准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);// 加载数据DataStream<String> fileStream = env.readTextFile("F:\\BD230801\\FlinkDemo\\datas\\a.log");// 数据处理转换DataStream<String> filterStream = fileStream.filter(new FilterFunction<String>() {@Overridepublic boolean filter(String line) throws Exception {String ip = line.split(" ")[0];return ip.equals("83.149.9.216");}});// 数据输出filterStream.print();// 执行env.execute();}
}

        在 filter 函数中，通过拆分每行日志获取 IP 地址，并与目标 IP 进行比较决定是否保留该条日志数据。

（四）keyBy 算子

功能概述
        keyBy 算子在流处理中用于对数据按照指定的键进行分组，类似于 SQL 中的 group by，后续可基于分组进行聚合等操作，支持对元组类型和 POJO 类型数据按不同方式指定分组键。

流处理中没有groupBy,而是keyBy

KeySelector对象可以支持元组类型，也可以支持POJO[Entry、JavaBean]

元组类型

单个字段keyBy

//用字段位置（已经被废弃）
wordAndOne.keyBy(0)//用字段表达式
wordAndOne.keyBy(v -> v.f0)

多个字段keyBy

//用字段位置
wordAndOne.keyBy(0, 1);//用KeySelector
wordAndOne.keyBy(new KeySelector<Tuple2<String, Integer>, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> getKey(Tuple2<String, Integer> value) throws Exception {return Tuple2.of(value.f0, value.f1);}
});

类似于sql中的group by

select sex,count(1) from student group by sex;

group by 后面也可以跟多个字段进行分组，同样 keyBy 也支持使用多个列进行分组 

POJO

public class PeopleCount {private String province;private String city;private Integer counts;public PeopleCount() {}//省略其他代码。。。
}

单个字段keyBy

source.keyBy(a -> a.getProvince());

多个字段keyBy

source.keyBy(new KeySelector<PeopleCount, Tuple2<String, String>>() {@Overridepublic Tuple2<String, String> getKey(PeopleCount value) throws Exception {return Tuple2.of(value.getProvince(), value.getCity());}
});

代码示例
        假设有数据表示不同球类的数量，格式为 Tuple2（球类名称，数量），如 Tuple2.of("篮球", 1) 等，需求是统计篮球、足球各自的总数量。代码如下：

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class KeyByDemo {public static void main(String[] args) throws Exception {// 准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);// 加载数据DataStream<Tuple2<String, Integer>> dataStream = env.fromElements(Tuple2.of("篮球", 1),Tuple2.of("篮球", 2),Tuple2.of("篮球", 3),Tuple2.of("足球", 3),Tuple2.of("足球", 2),Tuple2.of("足球", 3));// 数据处理转换/*KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> tuple2) throws Exception {return tuple2.f0;}});*/KeyedStream<Tuple2<String, Integer>, String> keyedStream = dataStream.keyBy(tuple -> tuple.f0);keyedStream.sum(1).print();// 执行env.execute();}
}

这里通过 lambda 表达式指定 Tuple2 的第一个元素（球类名称）作为分组键，对数据分组后使用 sum 聚合统计每种球类的数量总和。

pojo演示

package com.bigdata.day02;import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;/*** @基本功能:* @program:FlinkDemo* @author: 闫哥* @create:2024-05-13 14:32:52**/
public class Demo07 {@Data@AllArgsConstructorstatic class Ball{private String ballName;private int num;}public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据//3. transformation-数据处理转换//4. sink-数据输出// 以下演示数据是pojoDataStreamSource<Ball> ballSource = env.fromElements(new Ball("篮球", 1),new Ball("篮球", 2),new Ball("篮球", 3),new Ball("足球", 3),new Ball("足球", 2),new Ball("足球", 3));ballSource.keyBy(ball -> ball.getBallName()).print();ballSource.keyBy(new KeySelector<Ball, String>() {@Overridepublic String getKey(Ball ball) throws Exception {return ball.getBallName();}});//5. execute-执行env.execute();}
}

（五）reduce 算子

功能概述
        reduce 算子可对一个数据集或一个分组进行聚合计算，将多个元素逐步合并为一个最终元素，常用于求和、求最值等场景，sum 底层其实也是基于 reduce 实现。

代码示例
        读取访问日志，统计每个 IP 地址的访问 PV 数量，代码如下：

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.Date;public class ReduceDemo {@Data@AllArgsConstructor@NoArgsConstructorstatic class LogBean{String ip;      // 访问ipint userId;     // 用户idlong timestamp; // 访问时间戳String method;  // 访问方法String path;    // 访问路径}public static void main(String[] args) throws Exception {//1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//2. source-加载数据DataStream<String> fileStream = env.readTextFile("F:\\BD230801\\FlinkDemo\\datas\\a.log");//3. transformation-数据处理转换// 此处也可以将数据放入到tuple中，tuple可以支持到tuple25DataStream<LogBean> mapStream = fileStream.map(new MapFunction<String, LogBean>() {@Overridepublic LogBean map(String line) throws Exception {String[] arr = line.split(" ");String ip = arr[0];int userId = Integer.valueOf(arr[1]);String createTime = arr[2];// 如何将一个时间字符串变为时间戳// 17/05/2015:10:05:30/*SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");Date date = simpleDateFormat.parse(createTime);long timeStamp = date.getTime();*/// 要想使用这个common.lang3 下的工具类，需要导入包Date date = DateUtils.parseDate(createTime, "dd/MM/yyyy:HH:mm:ss");long timeStamp = date.getTime();String method = arr[3];String path = arr[4];LogBean logBean = new LogBean(ip, userId, timeStamp, method, path);return logBean;}});DataStream<Tuple2<String, Integer>> mapStream2 = mapStream.map(new MapFunction<LogBean, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(LogBean logBean) throws Exception {return Tuple2.of(logBean.getIp(), 1);}});//4. sink-数据输出KeyedStream<Tuple2<String,Integer>, String> keyByStream = mapStream2.keyBy(tuple -> tuple.f0);// sum的底层是 reduce// keyByStream.sum(1).print();//  [ ("10.0.0.1",1),("10.0.0.1",1),("10.0.0.1",1) ]keyByStream.reduce(new ReduceFunction<Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {// t1 => ("10.0.0.1",10)// t2 => ("10.0.0.1",1)return Tuple2.of(t1.f0,  t1.f1 + t2.f1);}}).print();//5. execute-执行env.execute();}
}

        先将日志数据处理成包含 IP 和计数 1 的 Tuple2 格式，按 IP 分组后，在 reduce 函数中对相同 IP 的计数进行累加，得到每个 IP 的访问 PV 数。

二、合并与连接操作

（一）union 算子

功能概述
        union 算子能够合并多个同类型的流，将多个 DataStream 合并成一个 DataStream，但注意合并时流的类型必须一致，且不会对数据去重。

代码示例

import org.apache.flink.api.common.RuntimeExecutionMode;
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 UnionConnectDemo {public static void main(String[] args) throws Exception {// 准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);// 加载数据DataStreamSource<String> stream1 = env.fromElements("hello", "nihao", "吃甘蔗的人");DataStreamSource<String> stream2 = env.fromElements("hello", "kong ni qi wa", "看电子书的人");// 合并流DataStream<String> unionStream = stream1.union(stream2);unionStream.print();// 执行env.execute();}
}

上述代码将两个包含字符串元素的流进行合并输出。

（二）connect 算子

功能概述
        connect 算子可连接 2 个不同类型的流，连接后形成 ConnectedStreams，内部两个流保持各自的数据和形式独立，之后需通过自定义处理逻辑（如 CoMapFunction 等）处理后再输出，且处理后的数据类型需相同。

代码示例

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;public class UnionConnectDemo {public static void main(String[] args) throws Exception {// 准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);// 加载数据DataStreamSource<String> stream1 = env.fromElements("hello", "nihao", "吃甘蔗的人");DataStreamSource<Long> stream3 = env.fromSequence(1, 10);// 连接流ConnectedStreams<String, Long> connectStream = stream1.connect(stream3);// 处理流DataStream<String> mapStream = connectStream.map(new CoMapFunction<String, Long, String>() {@Overridepublic String map1(String value) throws Exception {return value;}@Overridepublic String map2(Long value) throws Exception {return Long.toString(value);}});// 输出mapStream.print();// 执行env.execute();}
}

        这里连接了一个字符串流和一个长整型序列流，通过 CoMapFunction 分别将字符串按原样、长整型转换为字符串后合并输出。

三、侧输出流（Side Outputs）

功能概述
        侧输出流可根据自定义规则对输入流数据进行分流，将满足不同条件的数据输出到不同的“分支”流中，方便后续针对性处理。

代码示例
以下示例对流中的数据按照奇数和偶数进行分流并获取分流后的数据：

import org.apache.flink.api.common.RuntimeExecutionMode;
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 SideOutputExample {public static void main(String[] args) throws Exception {// 1. env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);// 侧道输出流// 从0到100生成一个Long类型的数据流作为示例数据DataStreamSource<Long> streamSource = env.fromSequence(0, 100);// 定义两个标签，用于区分偶数和奇数的侧输出流OutputTag<Long> tag_even = new OutputTag<Long>("偶数", TypeInformation.of(Long.class));OutputTag<Long> tag_odd = new OutputTag<Long>("奇数", TypeInformation.of(Long.class));// 2. source-加载数据// 使用ProcessFunction来处理每个元素，决定将其输出到哪个侧输出流或者主输出流SingleOutputStreamOperator<Long> process = streamSource.process(new ProcessFunction<Long, Long>() {@Overridepublic void processElement(Long value, ProcessFunction<Long, Long>.Context ctx, Collector<Long> out) throws Exception {// value代表每一个数据，判断是否为偶数if (value % 2 == 0) {// 如果是偶数，将其输出到偶数的侧输出流中ctx.output(tag_even, value);} else {// 如果是奇数，将其输出到奇数的侧输出流中ctx.output(tag_odd, value);}}});// 3. 获取奇数的所有数据，从主输出流中获取对应标签（tag_odd）的侧输出流数据DataStream<Long> sideOutput = process.getSideOutput(tag_odd);sideOutput.print("奇数：");// 获取所有偶数数据，同样从主输出流中获取对应标签（tag_even）的侧输出流数据DataStream<Long> sideOutput2 = process.getSideOutput(tag_even);sideOutput2.print("偶数：");// 4. sink-数据输出（这里通过打印展示了侧输出流的数据，实际应用中可对接其他下游操作）// 5. 执行任务env.execute();}
}

在上述代码中：

  • 首先，我们创建了 StreamExecutionEnvironment 来准备 Flink 的执行环境，并设置了运行模式为自动（AUTOMATIC）。
  • 接着，通过 fromSequence 方法生成了一个从 0 到 100 的 Long 类型的数据流作为示例的输入数据。
  • 然后，定义了两个 OutputTag，分别用于标记偶数和奇数的侧输出流，并且指定了对应的类型信息（这里都是 Long 类型）。
  • 在 process 函数中，针对每个输入的元素（value），通过判断其是否能被 2 整除来决定将其输出到对应的侧输出流中。如果是偶数，就通过 ctx.output(tag_even, value) 将其发送到偶数侧输出流；如果是奇数，则通过 ctx.output(tag_odd, value) 发送到奇数侧输出流。
  • 最后，通过 getSideOutput 方法分别获取奇数和偶数侧输出流的数据，并进行打印输出，以此展示了侧输出流的分流及获取数据的完整流程。实际应用场景中，这些侧输出流的数据可以进一步对接不同的业务逻辑进行相应处理，比如奇数流进行一种聚合计算，偶数流进行另一种统计分析等。

        这样，利用侧输出流的特性，我们可以很灵活地根据自定义条件对数据进行分流处理，满足多样化的数据处理需求。

四、总结

        本文围绕 Apache Flink 转换算子展开，旨在助力读者洞悉其核心要点与多样应用，以灵活处理复杂业务数据。

  1. 常用转换算子：
    • map：基于自定义逻辑，对输入元素逐一变换，如剖析日志字符串、提取关键信息并转换格式，封装为定制对象输出，契合精细化处理需求。
    • flatMap：将输入元素按需拆分为零个及以上输出元素，凭借拆分、重组操作，挖掘数据深层价值，适配展开、细化数据场景。
    • filter：依设定条件甄别筛选，精准把控数据流向，剔除不符元素，保障下游数据贴合业务关注点。
    • keyBy：类比 SQL 分组，依指定键归拢数据，为聚合奠基，以不同方式适配多元数据类型，实现分类统计。
    • reduce：聚焦数据集或分组，渐进聚合，可求和、求最值等，sum 操作底层亦仰仗于此，高效整合数据得出汇总结果。
  2. 合并与连接操作：
    • union：整合同类型多流为一，操作简便，唯需留意类型一致，虽不除重但拓宽数据维度。
    • connect：桥接不同类型流，借自定义逻辑协同处理，统一输出类型，达成跨流数据交互融合。
  3. 侧输出流特性：基于自定义规则巧妙分流，借OutputTag标记、ProcessFunction判定，将数据导向不同 “分支”，按需对接各异业务逻辑，于复杂场景中尽显灵活应变优势，全方位满足数据处理多元化诉求。

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

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

相关文章

解锁Scala编程：深入文本分析与数据处理的艺术

解锁Scala编程：深入文本分析与数据处理的艺术

引言&#xff1a; 在数据科学的世界里&#xff0c;Scala以其强大的并发能力和简洁的语法&#xff0c;成为处理大规模数据集的理想选择。本文将带您踏上一段Scala编程的探索之旅&#xff0c;从基础的文本分析到复杂的数据处理&#xff0c;每一步都精心设计&#xff0c;让您在实…
阅读更多...
Top 10 Tools to Level Up Your Prompt Engineering Skills

Top 10 Tools to Level Up Your Prompt Engineering Skills

此文章文字是转载翻译&#xff0c;图片是自已用AI 重新生成的。文字内容来自 https://www.aifire.co/p/top-10-ai-prompt-engineering-tools 供记录学习使用。 Introduction to AI Prompt Engineering AI Prompt Engineering 简介 1&#xff0c;Prompt Engineering 提示工程…
阅读更多...
【RK3588 Linux 5.x 内核编程】-内核线程与读写自旋锁

【RK3588 Linux 5.x 内核编程】-内核线程与读写自旋锁

内核线程与读写自旋锁 文章目录 内核线程与读写自旋锁1、读写自旋锁介绍2、Linux内核中的读写自旋锁2.1 初始化2.1.1 静态方法2.1.2 动态方法3、读写自旋锁加锁与解锁3.1 在用户上下文之间锁定3.1.1 读锁3.1.2 写锁3.2 下半部分之间的锁定3.3 在用户上下文和下半部分之间锁定3.…
阅读更多...
Rust语言俄罗斯方块（漂亮的界面案例+详细的代码解说+完美运行）

Rust语言俄罗斯方块（漂亮的界面案例+详细的代码解说+完美运行）

tetris-demo A Tetris example written in Rust using Piston in under 500 lines of code 项目地址: https://gitcode.com/gh_mirrors/te/tetris-demo 项目介绍 "Tetris Example in Rust, v2" 是一个用Rust语言编写的俄罗斯方块游戏示例。这个项目不仅是一个简单…
阅读更多...
Spring Boot 与 Spring Cloud Alibaba 版本兼容对照

Spring Boot 与 Spring Cloud Alibaba 版本兼容对照

版本选择要点 Spring Boot 3.x 与 Spring Cloud Alibaba 2022.0.x Spring Boot 3.x 基于 Jakarta EE&#xff0c;javax.* 更换为 jakarta.*。 需要使用 Spring Cloud 2022.0.x 和 Spring Cloud Alibaba 2022.0.x。 Alibaba 2022.0.x 对 Spring Boot 3.x 的支持在其发行说明中…
阅读更多...
Oracle, PostgreSQL 字符串排序不一致及调整

Oracle, PostgreSQL 字符串排序不一致及调整

有一个使用了不同数据库的应&#xff0c;Oracle 和 PostgreSQL&#xff0c;数据库中的记录完全相同&#xff0c;相同的查询语句(相同的排序&#xff0c;至少从字面上来说是的)取到的记录排序却不同&#xff0c;从而产生了 Bug。 简单演示一下默认排序各自在这两种数据库中的行为…
阅读更多...
（免费送源码）计算机毕业设计原创定制：Java+ssm+JSP+Ajax SSM棕榈校园论坛的开发

（免费送源码）计算机毕业设计原创定制：Java+ssm+JSP+Ajax SSM棕榈校园论坛的开发

摘要 随着计算机科学技术的高速发展,计算机成了人们日常生活的必需品&#xff0c;从而也带动了一系列与此相关产业&#xff0c;是人们的生活发生了翻天覆地的变化&#xff0c;而网络化的出现也在改变着人们传统的生活方式&#xff0c;包括工作&#xff0c;学习&#xff0c;社交…
阅读更多...
Ubuntu Opencv 源码包安装

Ubuntu Opencv 源码包安装

说明&#xff1a; ubuntu20.04 建议 使用 opencv-4.6.0版本 ubuntu18.04 建议 使用 opencv-4.5.2-版本 安装包准备 1、下载源码包 OpenCV官网 下载相关版本源码 Sources # 克隆方式 OpenCV 源码git clone https://github.com/opencv/opencv.gitcd opencvgit checkout 4.5.2 …
阅读更多...
Linux 下自动化之路：达梦数据库定期备份并推送至 GitLab 全攻略

Linux 下自动化之路：达梦数据库定期备份并推送至 GitLab 全攻略

目录 环境准备 生成SSH 密钥对 数据库备份并推送到gitlab脚本 设置定时任务 环境准备 服务器要有安装达梦数据库&#xff08;达梦安装这里就不示例了&#xff09;&#xff0c;git 安装Git 1、首先&#xff0c;确保包列表是最新的&#xff0c;运行以下命令&#xff1a; …
阅读更多...
极狐GitLab 17.6 正式发布几十项与 DevSecOps 相关的功能【一】

极狐GitLab 17.6 正式发布几十项与 DevSecOps 相关的功能【一】

GitLab 是一个全球知名的一体化 DevOps 平台&#xff0c;很多人都通过私有化部署 GitLab 来进行源代码托管。极狐GitLab 是 GitLab 在中国的发行版&#xff0c;专门为中国程序员服务。可以一键式部署极狐GitLab。 学习极狐GitLab 的相关资料&#xff1a; 极狐GitLab 官网极狐…
阅读更多...
＜项目代码＞YOLOv8 停车场空位识别＜目标检测＞

＜项目代码＞YOLOv8 停车场空位识别＜目标检测＞

YOLOv8是一种单阶段&#xff08;one-stage&#xff09;检测算法&#xff0c;它将目标检测问题转化为一个回归问题&#xff0c;能够在一次前向传播过程中同时完成目标的分类和定位任务。相较于两阶段检测算法&#xff08;如Faster R-CNN&#xff09;&#xff0c;YOLOv8具有更高的…
阅读更多...
【智能制造-46】人机工程（工厂自动化）

【智能制造-46】人机工程（工厂自动化）

工作空间设计 设备布局规划 根据人体测量学数据&#xff0c;合理安排自动化设备、生产线和工作区域的布局。例如&#xff0c;考虑工人的操作空间和活动范围&#xff0c;确保他们能够舒适地接近和操作设备。在汽车装配车间&#xff0c;机器人和工人的工作区域应划分明确&#…
阅读更多...
Spring Boot 集成 Knife4j 的 Swagger 文档

Spring Boot 集成 Knife4j 的 Swagger 文档

在开发微服务应用时&#xff0c;API 文档的生成和维护是非常重要的一环。Swagger 是一个非常流行的 API 文档工具&#xff0c;可以帮助我们自动生成 RESTful API 的文档&#xff0c;并提供了一个友好的界面供开发者测试 API。本文将介绍如何在 Spring Boot 项目中集成 Knife4j …
阅读更多...
一线大厂面试集锦

一线大厂面试集锦

String 为什么要设计成不可变的 String被设计成不可变的有以下几个原因: 线程安全:由于String是不可变的，多个线程可以同时访问同一个String对象而无需担心数据被修改。这使得String在多线程环境下是线程安全1. 的。 2.缓存Hash值:由于String是不可变的，它的hashcode可以…
阅读更多...
JavaScript：遍历一个对象数组，获取每个对象中 mainKeyId 对应的值，并以这个值作为新属性的键名，给该对象添加一个新属性。

JavaScript：遍历一个对象数组，获取每个对象中 mainKeyId 对应的值，并以这个值作为新属性的键名，给该对象添加一个新属性。

假设我们有一个对象数组&#xff0c;每个对象包含一个 mainKeyId 字段&#xff0c;以 mainKeyId 的值作为新的键名&#xff0c;并给它添加一个新的属性。可以通过以下代码实现&#xff1a; 示例代码&#xff1a; const arr [{ mainKeyId: key1, name: Item 1, value: 10 },{…
阅读更多...
php CURL请求502

php CURL请求502

增加 pm.max_children 的值 当前值为 50&#xff0c;如果并发请求较多&#xff0c;可能会耗尽子进程资源&#xff0c;导致 502。增加到 100 或更高&#xff08;根据服务器内存和负载能力调整&#xff09;。 2. 调整 request_terminate_timeout 当前为 100 秒&#xff0c;适合…
阅读更多...
微信小程序中会议列表页面的前后端实现

微信小程序中会议列表页面的前后端实现

题外话&#xff1a;想通过集成腾讯IM来解决即时聊天的问题&#xff0c;如果含语音视频&#xff0c;腾讯组件一年5万起步&#xff0c;贵了&#xff01;后面我们改为自己实现这个功能&#xff0c;这里只是个总结而已。 图文会诊需求 首先是个图文列表界面 同个界面可以查看具体…
阅读更多...
git（Linux）

git（Linux）

1.git 三板斧 基本准备工作&#xff1a; 把远端仓库拉拉取到本地了 .git --> 本地仓库 git在提交的时候&#xff0c;只会提交变化的部分 就可以在当前目录下新增代码了 test.c 并没有被仓库管理起来 怎么添加&#xff1f; 1.1 git add test.c 也不算完全添加到仓库里面&…
阅读更多...
【动手学电机驱动】STM32-FOC（8）MCSDK Profiler 电机参数辨识

【动手学电机驱动】STM32-FOC（8）MCSDK Profiler 电机参数辨识

STM32-FOC&#xff08;1&#xff09;STM32 电机控制的软件开发环境 STM32-FOC&#xff08;2&#xff09;STM32 导入和创建项目 STM32-FOC&#xff08;3&#xff09;STM32 三路互补 PWM 输出 STM32-FOC&#xff08;4&#xff09;IHM03 电机控制套件介绍 STM32-FOC&#xff08;5&…
阅读更多...
对于相对速度的重新理解 - 插一句

对于相对速度的重新理解 - 插一句

因为发了太多的公式&#xff0c;系统提示插入图片太频繁&#xff0c;下一个部分稍后再写。 这里要强调一下&#xff1a; 狭义相对论的平方和形式&#xff0c;其实就是因为分不清虚数单位的大小才写成这个样子。或者用物理语言来说&#xff0c;就是认为所有惯性系的“光速”都…
阅读更多...
最新文章

Copyright @ 2022~2023