Flink入门(五)--Flink算子

Map 

DataStream → DataStream 

一个接受一个元素并产生一个元素的函数。

示例

dataStream.map { x => x * 2 }

FlatMap 

DataStream → DataStream 

一个接受一个元素并产生零个、一个或多个元素的函数。

例如

dataStream.flatMap { str => str.split(" ") }

Filter 

DataStream → DataStream 

对于每个元素，设定一个布尔函数，并保留那些使函数返回true的元素。

例如 

dataStream.filter { _ != 0 }

KeyBy 

DataStream → KeyedStream 

逻辑上将流划分为不相交的分区。所有具有相同键的记录都被分配到同一个分区中。在内部，keyBy() 是通过哈希分区来实现的。指定键的方式有多种。

注意：没有实现hashcode()方法的POJO类和任何类型的数组都无法作为Key！！！

Reduce

KeyedStream → WindowedStream

该操作会连续地将当前元素与上一个reduce操作的结果（即最后一个reduced值）进行合并，并发出新的合并后的值。这种操作通常用于计算流数据的累积或滚动汇总。

 例如

keyedStream.reduce { _ + _ }

Window 

KeyedStream → WindowedStream 

在已经分区的KeyedStreams上可以定义窗口。窗口根据某些特性（例如，在过去5秒内到达的数据）将每个键中的数据分组。

例如 

dataStream.keyBy(_._1).window(TumblingEventTimeWindows.of(Time.seconds(5)))

对于窗口有关的知识点可以参考我的另一篇博文

Flink入门(四) -- Flink中的窗口_flink 窗口概念 使用场景-CSDN博客

WindowAll 

DataStream → AllWindowedStream

窗口可以在常规数据流（DataStream）上定义。窗口会根据某些特性（例如，在过去5秒内到达的数据）将所有流事件进行分组。

 例如

dataStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))

Tips：在许多情况下，这是一个非并行转换。对于windowAll操作符，所有记录都将被收集到一个任务中。

Window和WindowAll的异同

 Window Apply

WindowedStream → DataStream

Window Apply 是一个操作，它允许你应用一个函数到整个窗口上。这意味着你可以定义一个自定义函数来处理窗口内的所有元素，而不是仅仅对每个元素进行独立的操作。这个操作的结果是产生一个新的 DataStream，其中包含了函数处理每个窗口后的结果。

 例如

windowedStream.apply { WindowFunction }// applying an AllWindowFunction on non-keyed window stream
allWindowedStream.apply { AllWindowFunction }

 Union

与sql中union类似

DataStream* → DataStream

两个或多个数据流的联合操作会创建一个新的数据流，该数据流包含所有原始数据流中的所有元素。需要注意的是，如果你将一个数据流与自身进行联合，那么在结果数据流中，每个元素将会出现两次。（不去重不排序）

Join 

Join two data streams on a given key and a common window.

dataStream.join(otherStream).where(<key selector>).equalTo(<key selector>).window(TumblingEventTimeWindows.of(Time.seconds(3))).apply { ... }

Interval Join

KeyedStream,KeyedStream → DataStream

例如

假设你有两个数据流：

订单流（Order Stream）：包含订单的详细信息，每个订单都有一个唯一的订单ID、用户ID、订单时间戳（下单时间）和订单金额等。
支付流（Payment Stream）：包含支付的详细信息，每个支付都有一个唯一的支付ID、对应的订单ID、支付时间戳和支付金额等。

你的任务是分析订单的支付情况，包括支付是否及时（例如，是否在订单下单后的几分钟内完成支付）。这里，intervalJoin 就可以派上用场了。

// this will join the two streams so that
// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2
keyedStream.intervalJoin(otherKeyedStream).between(Time.milliseconds(0), Time.milliseconds(20000)) // lower and upper bound.upperBoundExclusive(true) // optional.lowerBoundExclusive(true) // optional.process(new IntervalJoinFunction() {...})

partition

• 自定义分区

• DataStream→DataStream 使用用户定义的分区程序为每个数据元选择目标任务。

dataStream.partitionCustom(partitioner, "someKey");
dataStream.partitionCustom(partitioner, 0);

• 随机分区

• DataStream→DataStream 根据均匀分布随机分配数据元。

dataStream.shuffle();

• Rebalance （循环分区）

• DataStream→DataStream 分区数据元循环，每个分区创建相等的负载。在存在数据倾斜时用于性能优化。

dataStream.rebalance();

   · rescaling

元素以轮询方式分区到下游操作的一个子集。这在您希望拥有这样的管道时非常有用，例如，从源的每个并行实例分发到几个映射器的子集以分散负载，但又不想触发rebalance()方法所带来的全面重新平衡。这取决于其他配置值（如TaskManager的插槽数），可能只需要本地数据传输，而不需要通过网络传输数据。

上游操作发送元素的下游操作子集取决于上游和下游操作的并行度。例如，如果上游操作有2个并行度，而下游操作有6个并行度，那么一个上游操作会将元素分发到三个下游操作，而另一个上游操作会将元素分发到另外三个下游操作。另一方面，如果下游操作有2个并行度，而上游操作有6个并行度，那么三个上游操作会将元素分发到一个下游操作，而另外三个上游操作会将元素分发到另一个下游操作。

在不同并行度不是彼此倍数的情况下，一个或多个下游操作将从上游操作接收到不同数量的输入。

dataStream.rescale()