【Spark Java API】Action(1)—reduce、aggregate
6294 ワード
reduce
公式ドキュメントの説明:
関数のプロトタイプ:
マッピング関数fに従ってRDD中の要素を二元計算(交換則と結合則を満たす)し、計算結果を返す.
ソース分析:
ソースコードから分かるように、reduce関数はRDDの要素に対してreduceLeft関数操作を行うことに相当し、reduceLeft関数はリストの左側から右側にreduce関数を適用する.その後、結果はdriver側でマージ処理されるため、パーティション数が多すぎるか、カスタム関数が複雑すぎるとdriver側の負荷が重くなる.
例:
aggregate
公式ドキュメントの説明:
関数のプロトタイプ:
Aggregateは、各区分の各要素をマージし、パーティション結果をmerge処理します.この関数が最終的に返すタイプはRDDの要素タイプと一致する必要はありません.
ソース分析:
ソースコードからaggregate関数は、各パーティションに対してscala集合を用いてaggregateを操作し、comb()を用いて前の各パーティション結果を集約することがわかる.
例:
公式ドキュメントの説明:
Reduces the elements of this RDD using the specified commutative and associative binary operator.
関数のプロトタイプ:
def reduce(f: JFunction2[T, T, T]): T
マッピング関数fに従ってRDD中の要素を二元計算(交換則と結合則を満たす)し、計算結果を返す.
ソース分析:
def reduce(f: (T, T) => T): T = withScope {
val cleanF = sc.clean(f)
val reducePartition: Iterator[T] => Option[T] = iter => {
if (iter.hasNext) {
Some(iter.reduceLeft(cleanF))
} else {
None
}
}
var jobResult: Option[T] = None
val mergeResult = (index: Int, taskResult: Option[T]) => {
if (taskResult.isDefined) {
jobResult = jobResult match {
case Some(value) => Some(f(value, taskResult.get))
case None => taskResult
}
}
}
sc.runJob(this, reducePartition, mergeResult)
// Get the final result out of our Option, or throw an exception if the RDD was empty
jobResult.getOrElse(throw new UnsupportedOperationException("empty collection"))
}
ソースコードから分かるように、reduce関数はRDDの要素に対してreduceLeft関数操作を行うことに相当し、reduceLeft関数はリストの左側から右側にreduce関数を適用する.その後、結果はdriver側でマージ処理されるため、パーティション数が多すぎるか、カスタム関数が複雑すぎるとdriver側の負荷が重くなる.
例:
JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
List data = Arrays.asList(5, 1, 1, 4, 4, 2, 2);
JavaRDD javaRDD = javaSparkContext.parallelize(data,3);
Integer reduceRDD = javaRDD.reduce(new Function2() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~" + reduceRDD);
aggregate
公式ドキュメントの説明:
Aggregate the elements of each partition, and then the results for all the partitions, using given combine functions and a neutral "zero value". This function can return a different result type, U, than the type of this RDD, T. Thus, we need one operation for merging a T into an U and one operation for merging two U's, as in scala.TraversableOnce. Both of these functions are allowed to modify and return their first argument instead of creating a new U to avoid memory allocation.
関数のプロトタイプ:
def aggregate[U](zeroValue: U)(seqOp: JFunction2[U, T, U], combOp: JFunction2[U, U, U]): U
Aggregateは、各区分の各要素をマージし、パーティション結果をmerge処理します.この関数が最終的に返すタイプはRDDの要素タイプと一致する必要はありません.
ソース分析:
def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U = withScope {
// Clone the zero value since we will also be serializing it as part of tasks
var jobResult = Utils.clone(zeroValue, sc.env.serializer.newInstance())
val cleanSeqOp = sc.clean(seqOp)
val cleanCombOp = sc.clean(combOp)
val aggregatePartition = (it: Iterator[T]) => it.aggregate(zeroValue)(cleanSeqOp, cleanCombOp)
val mergeResult = (index: Int, taskResult: U) => jobResult = combOp(jobResult, taskResult)
sc.runJob(this, aggregatePartition, mergeResult)
jobResult
}
ソースコードからaggregate関数は、各パーティションに対してscala集合を用いてaggregateを操作し、comb()を用いて前の各パーティション結果を集約することがわかる.
例:
JavaSparkContext javaSparkContext = new JavaSparkContext(sparkConf);
List data = Arrays.asList(5, 1, 1, 4, 4, 2, 2);
JavaRDD javaRDD = javaSparkContext.parallelize(data,3);
Integer aggregateRDD = javaRDD.aggregate(2, new Function2() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
}, new Function2() {
@Override
public Integer call(Integer v1, Integer v2) throws Exception {
return v1 + v2;
}
});
System.out.println("~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~" + aggregateRDD);