Spark 1.5.2でgroupby操作を改善しました

Question

Sparkを使用してパフォーマンスが低下しています。

私は2つの具体的な質問を持っている：私たちは、いくつかは非常に仕上げ、RDDで行わGROUPBY操作のいくつかは、より多くの時間を取っている。また

段階の数が二回登場していることに気づいたデバッグ

1. すばやくいくつかの時間がかかります

ここにはのスクリーンショットがあります。

現在、ローカルで実行され、シャッフルパーティションが2に設定され、パーティション数が5に設定されていると、データは約100,000レコードになります。

groupby操作といえば、2つの列に基づいてデータフレーム（いくつかの結合の結果）をグループ化してから、結果を得るために関数を適用します。

 val groupedRows = rows.rdd.groupBy(row => (
     row.getAs[Long](Column1), 
     row.getAs[Int](Column2) 
     )) 
     val rdd = groupedRows.values.map(Criteria) 

ここで、Criteriaはグループ化された結果の行に作用する関数です。このグループをどのようにして最適化できますか？

のスクリーンショットです。

Answer 1

私は、既存のdataframeをrddに変換しておらず、実行している複雑なプロセスを実行しないことをお勧めします。

Criteriaの機能を2列（Column1およびColumn2）に実行する場合は、dataframeで直接行うことができます。さらに、Criteriaをinbuilt functionsの組み合わせに減らすことができればそれは素晴らしいことでしょう。ただし、カスタムルールには常にudfの関数を使用できます。私はやってあなたをお勧めします何

はdataframeにgroupByであり、あなたがグループ化されたdataframeから複数の行をしたい場合は、Window機能を使用することができますaggregation機能

rows.groupBy("Column1", "Column2").agg(Criteria function) 

を適用します。詳細here

Answer 2

.groupByis known最も効率的なアプローチではないことにする。

注：この操作は非常に高価である可能性があります。 キーで キーごとに集計（合計や平均など）を行う場合は、PairRDDFunctions.aggregateByKeyまたは を使用して、PairRDDFunctions.reduceByKeyを使用すると、パフォーマンスが大幅に向上します。

hereが説明したように時には.reduceByKeyまたは.aggregateByKeyを使用することをお勧めします：

、これらの機能の両方が正しい答えを生成しますが、 reduceByKeyの例は、大規模なデータセットでより良い動作します。それは です。なぜなら、Sparkは、データをシャッフルする前に各 パーティションの共通キーと出力を組み合わせることができると認識しているからです。

なぜ.reduceByKey、仕事速い.groupByより.aggregateByKey？集約の一部はmapフェーズ中に発生し、reduceフェーズの間にワーカーノードの周りにシャッフルされるデータが少なくなるためです。ここにはhow does aggregateByKey workの良い説明があります。