洞悉Flink运算符：keyBy、reduce和aggregations的妙用

2024-01-09 20:06:50

深入剖析keyBy算子

keyBy算子是Flink中基于Key的分组转换的基石，它允许用户根据指定的字段将数据流划分为多个分组。这种分组操作对于数据聚合、窗口计算等场景至关重要。

keyBy算子的使用

使用keyBy算子非常简单，只需指定需要作为分组字段的字段名即可。例如，以下代码将数据流中的“user_id”字段作为分组字段：

DataStream<Tuple2<String, Long>> keyedStream = dataStream
  .keyBy(tuple -> tuple.f0);

keyBy算子的优点

keyBy算子具有以下优点：

分组效率高：keyBy算子利用哈希表进行分组，分组效率非常高，可以满足实时数据处理的需求。
分组结果准确：keyBy算子保证相同Key的数据流元素会被分配到同一个分组中，分组结果准确无误。
灵活的分组字段选择：keyBy算子支持任意字段作为分组字段，为用户提供了极大的灵活性。

reduce算子和aggregations算子的强强联合

reduce算子和aggregations算子是Flink中用于数据聚合的算子。reduce算子允许用户在每个分组内对数据流元素进行累积计算，而aggregations算子则提供了更丰富的聚合函数，例如求和、求平均值、求最大值等。

reduce算子的使用

reduce算子的使用也非常简单，只需指定聚合函数和聚合字段即可。例如，以下代码使用reduce算子计算每个分组中数据流元素的和：

DataStream<Long> reducedStream = keyedStream
  .reduce((tuple1, tuple2) -> tuple1.f1 + tuple2.f1);

aggregations算子的使用

aggregations算子的使用与reduce算子类似，只需指定聚合函数和聚合字段即可。例如，以下代码使用aggregations算子计算每个分组中数据流元素的平均值：

DataStream<Double> aggregatedStream = keyedStream
  .aggregations(Aggregations.SUM(1), Aggregations.COUNT(0))
  .map(aggregation -> aggregation.f0 / aggregation.f1);

reduce算子和aggregations算子的区别

reduce算子和aggregations算子的主要区别在于聚合函数的选择。reduce算子只支持自定义的聚合函数，而aggregations算子提供了更丰富的内置聚合函数，例如求和、求平均值、求最大值等。

实例演示：keyBy、reduce和aggregations算子的协同合作

为了更好地理解keyBy、reduce和aggregations算子的使用方法，我们来看一个实例。假设我们有一个数据流，其中包含用户ID、商品ID和购买数量。我们想要计算每个用户购买的总金额。

我们可以使用以下代码实现这个需求：

DataStream<Tuple3<String, String, Long>> dataStream = ...;

DataStream<Tuple2<String, Long>> keyedStream = dataStream
  .keyBy(tuple -> tuple.f0);

DataStream<Tuple2<String, Long>> reducedStream = keyedStream
  .reduce((tuple1, tuple2) -> Tuple2.of(tuple1.f0, tuple1.f2 + tuple2.f2));

DataStream<Tuple2<String, Long>> aggregatedStream = keyedStream
  .aggregations(Aggregations.SUM(2))
  .map(aggregation -> Tuple2.of(aggregation.f0, aggregation.f1));