flink Reduce、GroupReduce、GroupCombine笔记

1、reduce操作，在分组的dataset上使用，也可以在不分组的dataset上使用

应用于分组DataSet的Reduce转换使用用户定义的reduce函数将每个组减少为单个元素。对于每组输入元素，reduce函数连续地将元素对组合成一个元素，直到每个组只剩下一个元素。

注意，对于ReduceFunction，返回对象的key字段应与输入值匹配。这是因为reduce是可隐式组合（combine）的，并且从combine运算符发出的对象在传递给reduce运算符时再次按key分组。

1.1 使用key表达式的dataset进行reduce

key表达式指定DataSet的每个元素的一个或多个字段。每个key表达式都是公共字段的名称或getter方法。用点被用于向下钻取对象。key表达式“*”选择所有字段。以下代码显示如何使用key表达式对POJO DataSet进行分组，并使用reduce函数对其进行规约。

// some ordinary POJO

public class WC {

  public String word;

  public int count;

  // [...]

}

// ReduceFunction that sums Integer attributes of a POJO

public class WordCounter implements ReduceFunction<WC> {

  @Override

  public WC reduce(WC in1, WC in2) {

    return new WC(in1.word, in1.count + in2.count);

  }

}

// [...]

DataSet<WC> words = // [...]

DataSet<WC> wordCounts = words

                         // DataSet grouping on field "word"

                         .groupBy("word")

                         // apply ReduceFunction on grouped DataSet

                         .reduce(new WordCounter());

1.2 使用KeySelector函数的dataset上进行reduce

key选择器函数从DataSet的每个元素中提取键值。提取的key用于对DataSet进行分组。以下代码显示如何使用键选择器函数对POJO DataSet进行分组，并使用reduce函数对其进行规约操作。

// some ordinary POJO

public class WC {

  public String word;

  public int count;

  // [...]

}

// ReduceFunction that sums Integer attributes of a POJO

public class WordCounter implements ReduceFunction<WC> {

  @Override

  public WC reduce(WC in1, WC in2) {

    return new WC(in1.word, in1.count + in2.count);

  }

}

// [...]

DataSet<WC> words = // [...]

DataSet<WC> wordCounts = words

                         // DataSet grouping on field "word"

                         .groupBy(new SelectWord())

                         // apply ReduceFunction on grouped DataSet

                         .reduce(new WordCounter());

public class SelectWord implements KeySelector<WC, String> {

  @Override

  public String getKey(Word w) {

    return w.word;

  }

}

1.3 在Tuple元组上应用的reduce，可以使用数字来指明字段位置，类似索引

字段位置键指定一个或多个字段用于分组

DataSet<Tuple3<String, Integer, Double>> tuples = // [...]

DataSet<Tuple3<String, Integer, Double>> reducedTuples = tuples

                                         // group DataSet on first and second field of Tuple

                                         .groupBy(0, 1)

                                         // apply ReduceFunction on grouped DataSet

                                         .reduce(new MyTupleReducer());

1.4 在整个数据集上应用reduce

Reduce转换可以将用户定义的reduce函数应用于DataSet的所有元素。 reduce函数随后将元素对组合成一个元素，直到只剩下一个元素。

使用Reduce转换规约完整的DataSet意味着最终的Reduce操作不能并行完成。但是，reduce函数可以自动组合，因此Reduce转换不会限制大多数用例的可伸缩性

以下代码显示如何对Integer DataSet的所有元素求和：

// ReduceFunction that sums Integers

public class IntSummer implements ReduceFunction<Integer> {

  @Override

  public Integer reduce(Integer num1, Integer num2) {

    return num1 + num2;

  }

}

// [...]

DataSet<Integer> intNumbers = // [...]

DataSet<Integer> sum = intNumbers.reduce(new IntSummer());

2、分组reduce，即GroupReduce

应用于分组DataSet的GroupReduce调用用户定义的group-reduce函数转换每个分组。
这与Reduce的区别在于用户定义的函数会立即获得整个组。在组的所有元素上使用Iterable调用该函数，并且可以返回任意数量的结果元素。

2.1 GroupReduce对于分组的键于redeuce相同

以下代码显示如何从Integer分组的DataSet中删除重复的字符串。

public class DistinctReduce

         implements GroupReduceFunction<Tuple2<Integer, String>, Tuple2<Integer, String>> {

  @Override

  public void reduce(Iterable<Tuple2<Integer, String>> in, Collector<Tuple2<Integer, String>> out) {

    Set<String> uniqStrings = new HashSet<String>();

    Integer key = null;

    // add all strings of the group to the set

    for (Tuple2<Integer, String> t : in) {

      key = t.f0;

      uniqStrings.add(t.f1);

    }

    // emit all unique strings.

    for (String s : uniqStrings) {

      out.collect(new Tuple2<Integer, String>(key, s));

    }

  }

}

// [...]

DataSet<Tuple2<Integer, String>> input = // [...]

DataSet<Tuple2<Integer, String>> output = input

                           .groupBy(0)            // group DataSet by the first tuple field

                           .reduceGroup(new DistinctReduce());  // apply GroupReduceFunction

2.2 将GroupReduce应用于排序分组的数据集

group-reduce函数使用Iterable访问组的元素。Iterable可以按指定的顺序分发组的元素（可选）。在许多情况下，这可以帮助降低用户定义的组减少功能的复杂性并提高其效率。

下面的代码显示了如何删除由Integer分组并按String排序的DataSet中的重复字符串的另一个示例。

// GroupReduceFunction that removes consecutive identical elements

public class DistinctReduce

         implements GroupReduceFunction<Tuple2<Integer, String>, Tuple2<Integer, String>> {

  @Override

  public void reduce(Iterable<Tuple2<Integer, String>> in, Collector<Tuple2<Integer, String>> out) {

    Integer key = null;

    String comp = null;

    for (Tuple2<Integer, String> t : in) {

      key = t.f0;

      String next = t.f1;

      // check if strings are different

      if (com == null || !next.equals(comp)) {

        out.collect(new Tuple2<Integer, String>(key, next));

        comp = next;

      }

    }

  }

}

// [...]

DataSet<Tuple2<Integer, String>> input = // [...]

DataSet<Double> output = input

                         .groupBy(0)                         // group DataSet by first field

                         .sortGroup(1, Order.ASCENDING)      // sort groups on second tuple field

                         .reduceGroup(new DistinctReduce());

3、可组合的GroupReduce功能

与reduce函数相比，group-reduce函数不是可隐式组合的。为了使group-reduce函数可组合，它必须实现GroupCombineFunction接口。

要点：GroupCombineFunction接口的通用输入和输出类型必须等于GroupReduceFunction的通用输入类型，如以下示例所示：

// Combinable GroupReduceFunction that computes a sum.

public class MyCombinableGroupReducer implements

  GroupReduceFunction<Tuple2<String, Integer>, String>,

  GroupCombineFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>

{

  @Override

  public void reduce(Iterable<Tuple2<String, Integer>> in,

                     Collector<String> out) {

    String key = null;

    int sum = 0;

    for (Tuple2<String, Integer> curr : in) {

      key = curr.f0;

      sum += curr.f1;

    }

    // concat key and sum and emit

    out.collect(key + "-" + sum);

  }

  @Override

  public void combine(Iterable<Tuple2<String, Integer>> in,

                      Collector<Tuple2<String, Integer>> out) {

    String key = null;

    int sum = 0;

    for (Tuple2<String, Integer> curr : in) {

      key = curr.f0;

      sum += curr.f1;

    }

    // emit tuple with key and sum

    out.collect(new Tuple2<>(key, sum));

  }

}

4、GroupCombine 分组连接

GroupCombine转换是可组合GroupReduceFunction中组合步骤的通用形式。它在某种意义上被概括为允许将输入类型I组合到任意输出类型O.
相反，GroupReduce中的组合步骤仅允许从输入类型I到输出类型I的组合。这是因为reduce步骤中，GroupReduceFunction期望输入类型为I.

在一些应用中，期望在执行附加变换（例如，减小数据大小）之前将DataSet组合成中间格式。这可以通过CombineGroup转换能以非常低的成本实现。

注意：分组数据集上的GroupCombine在内存中使用贪婪策略执行，该策略可能不会一次处理所有数据，而是以多个步骤处理。
它也可以在各个分区上执行，而无需像GroupReduce转换那样进行数据交换。这可能会导致输出的是部分结果，
所以GroupCombine是不能替代GroupReduce操作的，尽管它们的操作内容可能看起来都一样。

以下示例演示了如何将CombineGroup转换用于备用WordCount实现。

DataSet<String> input = [..] // The words received as input

DataSet<Tuple2<String, Integer>> combinedWords = input

  .groupBy(0) // group identical words

  .combineGroup(new GroupCombineFunction<String, Tuple2<String, Integer>() {

    public void combine(Iterable<String> words, Collector<Tuple2<String, Integer>>) { // combine

        String key = null;

        int count = 0;

        for (String word : words) {

            key = word;

            count++;

        }

        // emit tuple with word and count

        out.collect(new Tuple2(key, count));

    }

});

DataSet<Tuple2<String, Integer>> output = combinedWords

  .groupBy(0)                              // group by words again

  .reduceGroup(new GroupReduceFunction() { // group reduce with full data exchange

    public void reduce(Iterable<Tuple2<String, Integer>>, Collector<Tuple2<String, Integer>>) {

        String key = null;

        int count = 0;

        for (Tuple2<String, Integer> word : words) {

            key = word;

            count++;

        }

        // emit tuple with word and count

        out.collect(new Tuple2(key, count));

    }

});

巴特西

flink Reduce、GroupReduce、GroupCombine笔记

最新文章

热门文章