怎样设计网站,郑州网站推广公司地址,台州网络优化,昆山做网站的那家好Flink数据算子转换有很多类型#xff0c;各位看官看好#xff0c;接下来#xff0c;演示其中的十八种类型。
1.Map#xff08;映射转换#xff09; DataStream → DataStream 将函数作用在集合中的每一个元素上,并返回作用后的结果#xff0c;其中输入是一个数据流… Flink数据算子转换有很多类型各位看官看好接下来演示其中的十八种类型。
1.Map映射转换 DataStream → DataStream 将函数作用在集合中的每一个元素上,并返回作用后的结果其中输入是一个数据流输出的也是一个数据流
DataStreamInteger dataStream //加载数据源
dataStream.map(new MapFunctionInteger, Integer() {Overridepublic Integer map(Integer age) throws Exception {return 2 age;}
});2.Flatmap扁平映射转换 DataStream → DataStream FlatMap 将集合中的每个元素变成一个或多个元素,并返回扁平化之后的结果即采用一条记录并输出零个一个或多个记录。
//加载数据源dataStream.flatMap(new FlatMapFunctionString, String() {Overridepublic void flatMap(String value, CollectorString out)throws Exception {for(String word: value.split(,)){out.collect(word);}}
});3.Filter过滤转换 DataStream → DataStream 按照指定的条件对集合中的元素进行过滤,过滤出返回true/符合条件的元素。
// 过滤出年龄大于18的数据
dataStream.filter(new FilterFunctionInteger() {Overridepublic boolean filter(Integer age) throws Exception {return age 18;}
});Keyby分组转换 DataStream → KeyedStream 按照指定的key来对流中的数据进行分组在逻辑上是基于 key 对流进行分区。在内部它使用 hash 函数对流进行分区。它返回 KeyedDataStream 数据流。
KeyedStreamStudent, Integer keyBy student.keyBy(new KeySelectorStudent, Integer() {Overridepublic Integer getKey(Student value) throws Exception {return value.age;}
});
4.Reduce归约转换 KeyedStream → DataStream 对集合中的元素进行聚合Reduce 返回单个的结果值并且 reduce 操作每处理一个元素总是创建一个新值。常用的方法有 average, sum, min, max, count使用 reduce 方法都可实现。
keyedStream.reduce(new ReduceFunctionInteger() {Overridepublic Integer reduce(Integer value1, Integer value2)throws Exception {return value1 * value2;}
});
5.Aggregations聚合转换 KeyedStream → DataStream 在分组后的数据集上进行聚合操作如求和、计数、最大值、最小值等。这些函数可以应用于 KeyedStream 以获得 Aggregations 聚合。
DataStreamTuple2String, Integer dataStream ...; // 加载数据源dataStream.keyBy(0) // 对元组的第一个元素进行分组 .sum(1); // 对元组的第二个元素求和
6.Window分组开窗转换
KeyedStream → WindowedStream Flink 定义数据片段以便可能处理无限数据流。 这些切片称为窗口将数据流划分为不重叠的窗口并在每个窗口上执行转换操作常用于对时间窗口内的数据进行处理。 此切片有助于通过应用转换处理数据块。 要对流进行窗口化需要分配一个可以进行分发的键和一个描述要对窗口化流执行哪些转换的函数 要将流切片到窗口我们可以使用 Flink 自带的窗口分配器。 我们有选项如 tumbling windows, sliding windows, global 和 session windows。 Flink 还允许您通过扩展 WindowAssginer 类来编写自定义窗口分配器。
inputStream.keyBy(0).window(Time.seconds(10));上述案例是数据分组后是以 10 秒的时间窗口聚合
7.WindowAll开窗转换 DataStream → AllWindowedStream 类似于 Window 操作但是对整个数据流应用窗口操作而不是对每个 key 分别应用。 windowAll 函数允许对常规数据流进行分组。 通常这是非并行数据转换因为它在非分区数据流上运行。 唯一的区别是它们处理窗口数据流。 所以窗口缩小就像 Reduce 函数一样Window fold 就像 Fold 函数一样并且还有聚合。 // 创建一个简单的数据流 DataStreamTuple2String, Integer dataStream ...; // 请在此处填充你的数据源 // 定义一个 ReduceFunction用于在每个窗口内进行求和操作 ReduceFunctionTuple2String, Integer reduceFunction new ReduceFunctionTuple2String, Integer() { Override public Tuple2String, Integer reduce(Tuple2String, Integer value1, Tuple2String, Integer value2) throws Exception { return new Tuple2(value1.f0, value1.f1 value2.f1); } }; // 使用 WindowAll 方法指定时间窗口和 ReduceFunction dataStream.windowAll(Time.of(5, TimeUnit.SECONDS), reduceFunction) .print(); // 输出结果到 stdout (for debugging)
8.Union合并转换 DataStream* → DataStream 将多个数据流合并为一个数据流。union算子可以合并多个同类型的数据流并生成同类型的数据流即可以将多个DataStream[T]合并为一个新的DataStream[T]。数据将按照先进先出First In First Out的模式合并且不去重
//通过一些 key 将同一个 window 的两个数据流 join 起来。inputStream.join(inputStream1).where(0).equalTo(1).window(Time.seconds(5)) .apply (new JoinFunction () {...});
// 以上示例是在 5 秒的窗口中连接两个流其中第一个流的第一个属性的连接条件等于另一个流的第二个属性。
9.Connect / CoMap / CoFlatMap连接转换 DataStream,DataStream → DataStream 连接两个数据流并对连接后的数据流进行转换操作。 ConnectConnect 算子用于连接两个数据流这两个数据流的类型可以不同。Connect 算子会将两个数据流连接成一个 ConnectedStreams 对象但并不对元素做任何转换操作。Connect 算子通常用于需要将两个不同类型的数据流进行关联处理的场景。 CoMapCoMap 算子用于对 ConnectedStreams 中的每一个数据流应用一个 map 函数将它们分别转换为另一种类型。CoMap 会将两个数据流中的元素分别转换为不同的类型因此在使用 CoMap 时需要分别指定两个不同的 map 函数。 CoFlatMapCoFlatMap 算子和 CoMap 类似也是用于对 ConnectedStreams 中的每一个数据流应用一个 flatMap 函数将它们分别转换为另一种类型。不同之处在于CoFlatMap 生成的元素个数可以是 0、1 或多个因此适用于需要将每个输入元素转换为零个、一个或多个输出元素的情况。
// 创建两个数据流
DataStreamType1 dataStream1 ... // 从某个地方获取 Type1 类型的数据流
DataStreamType2 dataStream2 ... // 从某个地方获取 Type2 类型的数据流// 使用 Connect 算子连接两个数据流
ConnectedStreamsType1, Type2 connectedStreams dataStream1.connect(dataStream2);// 使用 CoMap 对每个数据流进行单独的转换
SingleOutputStreamOperatorResultType1 resultStream1 connectedStreams.map(new CoMapFunctionType1, ResultType1() {Overridepublic ResultType1 map1(Type1 value) throws Exception {// 对 Type1 数据流的转换逻辑// ...return transformedResult1;}
});SingleOutputStreamOperatorResultType2 resultStream2 connectedStreams.map(new CoMapFunctionType2, ResultType2() {Overridepublic ResultType2 map2(Type2 value) throws Exception {// 对 Type2 数据流的转换逻辑// ...return transformedResult2;}
});// 使用 CoFlatMap 对连接后的数据流进行转换
SingleOutputStreamOperatorResultType resultStream connectedStreams.flatMap(new CoFlatMapFunctionType1, Type2, ResultType() {Overridepublic void flatMap1(Type1 value, CollectorResultType out) throws Exception {// 对 Type1 数据流的转换逻辑// 将转换后的结果发射出去out.collect(transformedResult1);}Overridepublic void flatMap2(Type2 value, CollectorResultType out) throws Exception {// 对 Type2 数据流的转换逻辑// 将转换后的结果发射出去out.collect(transformedResult2);}
});// 执行任务
env.execute(Connect and CoMap Example);10.Join连接转换 KeyedStream,KeyedStream → DataStream 可以使用 join 算子来实现两个数据流的连接转换操作
java
// 创建两个数据流
DataStreamType1 inputStream1 ... // 从某个地方获取 Type1 类型的数据流
DataStreamType2 inputStream2 ... // 从某个地方获取 Type2 类型的数据流// 使用 keyBy 将两个数据流按照相同的字段进行分区
KeyedStreamType1, KeyType keyedStream1 inputStream1.keyBy(keySelector);
KeyedStreamType2, KeyType keyedStream2 inputStream2.keyBy(keySelector);// 使用 join 进行连接转换
SingleOutputStreamOperatorOutputType resultStream keyedStream1.join(keyedStream2).where(keySelector1).equalTo(keySelector2).window(windowAssigner).apply(new JoinFunctionType1, Type2, OutputType() {Overridepublic OutputType join(Type1 value1, Type2 value2) throws Exception {// 执行连接转换逻辑// ...return transformedResult;}});// 执行任务
env.execute(Join Example); 上述事例有两个输入数据流 inputStream1 和 inputStream2它们的元素类型分别为 Type1 和 Type2。对这两个数据流进行连接转换操作并输出连接后的结果。首先使用 keyBy 对两个数据流进行分区然后使用 join 算子将两个分区后的数据流按照指定的条件进行连接。在 join 方法中我们需要指定连接条件和窗口分配器并通过 apply 方法应用一个 JoinFunction 对连接后的数据进行转换操作。在 JoinFunction 的 join 方法中我们可以编写具体的连接转换逻辑然后返回转换后的结果。 Split / Select将一个数据流拆分为多个数据流然后对不同的数据流进行选择操作。
此功能根据条件将流拆分为两个或多个流。 当您获得混合流并且您可能希望单独处理每个数据流时可以使用此方法。
11.Apply窗口中的元素自定义转换 WindowedStream → DataStream AllWindowedStream → DataStream 当使用 Flink 的 apply 方法时将一个自定义的函数应用于流中的每个元素并生成一个新的流。这个自定义的函数可以是 MapFunction、FlatMapFunction、FilterFunction 等接口的实现。
// 创建输入数据流
DataStreamType1 inputStream ... // 从某个地方获取 Type1 类型的数据流// 使用 apply 方法应用自定义函数
SingleOutputStreamOperatorOutputType resultStream inputStream.apply(new MyMapFunction());// 定义自定义的 MapFunction
public class MyMapFunction implements MapFunctionType1, OutputType {Overridepublic OutputType map(Type1 value) {// 执行转换操作OutputType transformedValue ... // 对输入元素进行一些转换操作return transformedValue;}
}// 执行任务
env.execute(Apply Example);
12.Iterate迭代转换 DataStream → IterativeStream → DataStream 允许在数据流上进行迭代计算通常用于实现迭代算法。iterate函数允许您定义一个迭代处理的核心逻辑并通过closeWith方法指定迭代结束的条件。 // 定义迭代逻辑DataSetLong iteration initialInput.iterate(1000) // 指定迭代上限.map(new MapFunctionLong, Long() {Overridepublic Long map(Long value) throws Exception {// 迭代处理逻辑这里简单地加1return value 1;}});// 指定迭代结束条件DataSetLong result iteration.closeWith(iteration.filter(value - value 10)); 在这个示例中使用iterate函数来定义迭代逻辑其中map函数对每个元素进行加1操作。接着我们使用closeWith方法来指定迭代结束的条件即当元素的值大于等于10时结束迭代。 需要注意的是在实际的迭代处理中需要根据具体业务逻辑来定义迭代的处理过程和结束条件。另外还需要注意迭代过程中的性能和资源消耗以及迭代次数的控制避免出现无限循环等问题。
13.CoGroup分组连接转换 DataStream,DataStream → DataStream 将两个或多个数据流中的元素进行连接操作通常基于相同的键进行连接。
import org.apache.flink.api.common.functions.CoGroupFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;import java.util.ArrayList;
import java.util.List;public class CoGroupExample {public static void main(String[] args) throws Exception {final ExecutionEnvironment env ExecutionEnvironment.getExecutionEnvironment();// 创建第一个数据集ListTuple2Integer, String firstDataSet new ArrayList();firstDataSet.add(new Tuple2(1, A));firstDataSet.add(new Tuple2(2, B));// 创建第二个数据集ListTuple2Integer, String secondDataSet new ArrayList();secondDataSet.add(new Tuple2(1, X));secondDataSet.add(new Tuple2(3, Y));// 将数据集转化为Flink的DataSetorg.apache.flink.api.java.DataSetTuple2Integer,String first env.fromCollection(firstDataSet);org.apache.flink.api.java.DataSetTuple2Integer,String second env.fromCollection(secondDataSet);// 使用CoGroup算子进行连接first.coGroup(second).where(0) // 第一个数据集的连接字段.equalTo(0) // 第二个数据集的连接字段.with(new MyCoGroupFunction()) // 指定自定义的CoGroupFunction.print();env.execute();}// 自定义CoGroupFunctionpublic static class MyCoGroupFunction implements CoGroupFunctionTuple2Integer, String, Tuple2Integer, String, String {Overridepublic void coGroup(IterableTuple2Integer, String first, IterableTuple2Integer, String second, CollectorString out) {ListString valuesFromFirst new ArrayList();for (Tuple2Integer, String t : first) {valuesFromFirst.add(t.f1);}ListString valuesFromSecond new ArrayList();for (Tuple2Integer, String t : second) {valuesFromSecond.add(t.f1);}// 对两个数据集的分组进行连接操作for (String s1 : valuesFromFirst) {for (String s2 : valuesFromSecond) {out.collect(s1 - s2);}}}}
}在这个示例中首先创建了两个简单的数据集firstDataSet和secondDataSet然后将它们转换为Flink的DataSet对象。接着使用CoGroup算子对这两个数据集进行分组连接操作其中通过where和equalTo指定了连接字段通过with方法指定了自定义的CoGroupFunction。最后在CoGroupFunction中实现了对两个数据集分组的连接逻辑并通过Collector将结果输出。
14.Cross笛卡尔积转换 计算两个数据流的笛卡尔积。 DataStream,DataStream → DataStream
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;import java.util.ArrayList;
import java.util.List;public class CrossExample {public static void main(String[] args) throws Exception {final ExecutionEnvironment env ExecutionEnvironment.getExecutionEnvironment();// 创建第一个数据集ListInteger firstDataSet new ArrayList();firstDataSet.add(1);firstDataSet.add(2);// 创建第二个数据集ListString secondDataSet new ArrayList();secondDataSet.add(A);secondDataSet.add(B);// 将数据集转化为Flink的DataSetDataSetInteger first env.fromCollection(firstDataSet);DataSetString second env.fromCollection(secondDataSet);// 使用Cross算子进行笛卡尔积操作DataSetTuple2Integer, String result first.cross(second);// 打印结果result.print();env.execute();}
}在这个示例中首先创建了两个简单的数据集firstDataSet和secondDataSet然后将它们转换为Flink的DataSet对象。接着使用Cross算子对这两个数据集进行笛卡尔积操作得到了一个包含所有可能组合的新数据集。
15.Project投影转换 DataStream → DataStream 对数据集进行投影操作选择特定的字段或属性。Project算子用于从数据集中选择或投影出特定的字段。
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple3;import java.util.ArrayList;
import java.util.List;public class ProjectExample {public static void main(String[] args) throws Exception {final ExecutionEnvironment env ExecutionEnvironment.getExecutionEnvironment();// 创建数据集ListTuple3Integer, String, Double inputDataSet new ArrayList();inputDataSet.add(new Tuple3(1, Alice, 1000.0));inputDataSet.add(new Tuple3(2, Bob, 1500.0));inputDataSet.add(new Tuple3(3, Charlie, 2000.0));// 将数据集转化为Flink的DataSetDataSetTuple3Integer, String, Double input env.fromCollection(inputDataSet);// 使用Project算子进行字段投影DataSetTuple2Integer, String projectedDataSet input.project(0, 1); // 选择字段0和字段1// 打印结果projectedDataSet.print();env.execute();}
}在这个示例中首先创建了一个包含整数、字符串和双精度浮点数的元组数据集inputDataSet。然后将它们转换为Flink的DataSet对象。接着使用Project算子对数据集进行字段投影选择了字段0和字段1。最后打印出了字段投影后的结果。
16.Connect连接转换 DataStream,DataStream → ConnectedStreams connect提供了和union类似的功能用来连接两个数据流它与union的区别在于connect只能连接两个数据流union可以连接多个数据流。connect所连接的两个数据流的数据类型可以不一致union所连接的两个数据流的数据类型必须一致。 两个DataStream经过connect之后被转化为ConnectedStreamsConnectedStreams会对两个流的数据应用不同的处理方法且双流之间可以共享状态。
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.tuple.Tuple2;import java.util.ArrayList;
import java.util.List;public class ConnectExample {public static void main(String[] args) throws Exception {final ExecutionEnvironment env ExecutionEnvironment.getExecutionEnvironment();// 创建第一个数据集ListInteger firstDataSet new ArrayList();firstDataSet.add(1);firstDataSet.add(2);firstDataSet.add(3);// 创建第二个数据集ListString secondDataSet new ArrayList();secondDataSet.add(A);secondDataSet.add(B);secondDataSet.add(C);// 将数据集转化为Flink的DataSetDataSetInteger first env.fromCollection(firstDataSet);DataSetString second env.fromCollection(secondDataSet);// 使用Map将Integer类型转换为Tuple2类型DataSetTuple2Integer, String firstMapped first.map(new MapFunctionInteger, Tuple2Integer, String() {Overridepublic Tuple2Integer, String map(Integer value) {return new Tuple2(value, default);}});// 使用Connect算子将两个数据集连接在一起DataSetTuple2Integer, String connectedDataSet firstMapped.connect(second).map(new MapFunctionInteger, Tuple2Integer, String() {Overridepublic Tuple2Integer, String map(Integer value) {return new Tuple2(value, connected);}});// 打印结果connectedDataSet.print();env.execute();}
}在这个示例中首先创建了两个简单的数据集firstDataSet和secondDataSet然后将它们转换为Flink的DataSet对象。接着使用Map算子将第一个数据集中的整数类型转换为Tuple2类型。然后使用Connect算子将转换后的第一个数据集与第二个数据集连接在一起最后再对连接后的数据集进行处理。最终打印出了连接后的结果。
17.IntervalJoin时间窗口连接转换 KeyedStream,KeyedStream → DataStream IntervalJoin算子用于在两个数据流之间执行基于时间窗口的连接操作。
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.streaming.api.datastream.IntervalJoinOperator;
import org.apache.flink.streaming.api.windowed.TimeWindow;
import org.apache.flink.streaming.api.windowing.time.Time;public class IntervalJoinExample {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 创建第一个数据流DataStreamTuple2String, Integer firstStream ... // 从数据源获取第一个数据流// 创建第二个数据流DataStreamTuple2String, String secondStream ... // 从数据源获取第二个数据流// 定义时间窗口大小Time windowSize Time.seconds(10);// 使用IntervalJoin算子进行时间窗口连接IntervalJoinOperatorTuple2String, Integer, Tuple2String, String, String joinedStream firstStream.intervalJoin(secondStream).between(Time.seconds(-3), Time.seconds(3)) // 定义连接窗口范围.upperBoundExclusive() // 指定上界为不包含.lowerBoundExclusive() // 指定下界为不包含.process(new MyIntervalJoinFunction());// 打印结果joinedStream.print();// 执行任务env.execute(Interval Join Example);}// 自定义IntervalJoinFunctionpublic static class MyIntervalJoinFunction implements JoinFunctionTuple2String, Integer, Tuple2String, String, String {Overridepublic String join(Tuple2String, Integer first, Tuple2String, String second) {// 在这里实现连接后的处理逻辑return Joined: first.toString() and second.toString();}}
}在这个示例中首先创建了两个数据流firstStream和secondStream这些数据流可以来自各种数据源例如Kafka、Socket等。然后使用IntervalJoin算子将这两个数据流在时间窗口上进行连接操作通过定义连接窗口的范围来指定两个数据流之间的连接条件。最后定义了自定义的JoinFunction来处理连接后的数据。最终打印出了连接后的结果并执行Flink任务。
18.Split / Select拆分和选择转换 DataStream → DataStream Split 和 Select 算子用于将单个数据流拆分为多个流并选择其中的部分流进行处理。
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;public class SplitSelectExample {public static void main(String[] args) throws Exception {final StreamExecutionEnvironment env StreamExecutionEnvironment.getExecutionEnvironment();// 创建数据流DataStreamString inputDataStream ... // 从数据源获取数据流// 使用 Split 算子将数据流拆分为多个流SplitStreamString splitStream inputDataStream.split(new OutputSelectorString() {Overridepublic IterableString select(String value) {ListString output new ArrayList();if (value.contains(category1)) {output.add(category1);} else if (value.contains(category2)) {output.add(category2);} else {output.add(other);}return output;}});// 选择拆分后的流中的部分流进行处理DataStreamString category1Stream splitStream.select(category1);DataStreamString category2Stream splitStream.select(category2);// 对每个流进行进一步处理DataStreamString processedCategory1Stream category1Stream.map(new MyMapperFunction());DataStreamString processedCategory2Stream category2Stream.filter(new MyFilterFunction());// 将处理后的结果合并为一个流DataStreamString resultStream processedCategory1Stream.union(processedCategory2Stream);// 打印结果resultStream.print();// 执行任务env.execute(Split and Select Example);}// 自定义 Mapper 函数public static class MyMapperFunction implements MapFunctionString, String {Overridepublic String map(String value) {// 在这里实现对流中元素的转换操作return Processed Category1: value;}}// 自定义 Filter 函数public static class MyFilterFunction implements FilterFunctionString {Overridepublic boolean filter(String value) {// 在这里实现过滤逻辑return value.length() 10;}}
}在这个示例中首先创建了一个输入数据流inputDataStream然后使用 Split 算子将数据流拆分为三个不同的流category1、category2 和 other。接着使用 Select 算子选择了category1和category2两个流并对它们分别应用了自定义的 Mapper 函数和 Filter 函数进行处理。最后将处理后的结果合并为一个流并打印出来。 更多消息资讯请访问昂焱数据https://www.ayshuju.com
