dataset.countがシャッフルを引き起こすのはなぜですか？

Question

は、ここに私のデータフレームである（2.2をスパーク）：

根本的なRDDは、2つのパーティション

私はdf.countを行う

を持って、生産DAGは です

df.rdd.countを実行すると、DAGは次のようになります。

QUES：Countがスパークでのアクションで、公式の定義は「データフレーム内の行数を返します。」。さて、私がデータフレーム上でカウントを実行すると、なぜシャッフルが起こっていますか？また、基本的なRDDで同じことをしてもシャッフルは起こりません。

なぜシャッフルが起こるのかはわかりません。私はここでカウントのソースコードを調べようとしましたspark githubしかし、それは完全に私に意味をなさない。 "groupby"はアクションの犯人に供給されていますか？

PS。 df.coalesce（1）.countはシャッフルを起こさない

Answer 1

sparkがデータフレーム操作を行っているときは、最初にすべてのパーティションの部分カウントを計算し、次にそれらを合計する別のステージがあります。これは、大規模なデータフレームに特に適しています。ここでは、複数のエグゼキュータへのカウントの分配が実際にパフォーマンスに追加されます。これを確認する

場所は、以下の物理的な計画の記述のいくつかの並べ替えを持っているでしょうスパークUIのSQLタブです：

*HashAggregate(keys=[], functions=[count(1)], output=[count#202L]) 
+- Exchange SinglePartition 
    +- *HashAggregate(keys=[], functions=[partial_count(1)], output=[count#206L]) 

Answer 2

データフレームのカウント動作がシャッフルで結果GROUPBYを使用しているようです。あなたはRDDのカウント機能を見れば、それは配列として各パーティションの合計を返し、その後、合計に.SUM使用するか、各パーティションに集約関数に渡されますが、以下のhttps://github.com/apache/spark/blob/master/sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

* Returns the number of rows in the Dataset. 
* @group action 
* @since 1.6.0 
*/ 
def count(): Long = withAction("count", groupBy().count().queryExecution) { 
plan => 
plan.executeCollect().head.getLong(0) 
} 

からのコードです配列の要素このリンクから

コードスニペット： https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/rdd/RDD.scala

/** 
* Return the number of elements in the RDD. 
*/ 
def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum