例如:
DataSet> input = env.fromElements(1,2,3,4,5,6,7,8,9);
DataSet> sum = input.reduce(new ReduceFunction(),Tuple1>{
public Tuple1 reduce(Tuple1 value1,Tuple1 value2){
return new Tuple1<>(value1.f0 + value2.f0);
}
}
如果上面的reduce转换不是并行操作,我是否需要使用额外的两个转换'partitionByHash'和'mapPartition',如下所示:
DataSet> input = env.fromElements(1,2,3,4,5,6,7,8,9);
DataSet> sum = input.partitionByHash(0).mapPartition(new MapPartitionFunction(),Tuple1>{
public void map(Iterable> values,Collector> out){
long sum = getSum(values);
out.collect(new Tuple1(sum));
}
}).reduce(new ReduceFunction(),Tuple1>{
public Tuple1 reduce(Tuple1 value1,Tuple1 value2){
return new Tuple1<>(value1.f0 + value2.f0);
}
}
以及为什么reduce转换的结果仍然是DataSet的一个实例而不是一个实例 Tuple1
两者,reduce并且reduceGroup是集团为单位的运算和记录组应用.如果未指定使用分组键groupBy,则数据集的所有记录都属于同一组.因此,只有一个组并且最终结果reduce并reduceGroup不能并行计算.
如果reduce转换是可组合的(对于任何ReduceFunction和所有可组合的GroupReduceFunctions 都是如此),Flink可以并行应用组合器.
对你的两个问题有两个答案:
法比安给出了一个很好的解释.如果按键应用,则操作是并行的.否则只有预聚合是并行的.
在第二个示例中,通过引入密钥使其并行.您可以简单地编写(Java 8 Style)而不是使用"mapPartition()"的复杂解决方法
京公网安备 11010802040832号