DataFrame のパーティションをどのように定義しますか? 質問する

どうやら不可能ではないようです。それが意味をなすかどうかという疑問が残ります。ほとんどの場合、意味をなさないと私は主張します。

1. 再パーティション化はコストのかかるプロセスです。一般的なシナリオでは、ほとんどのデータをシリアル化、シャッフル、デシリアル化する必要があります。一方、事前にパーティション化されたデータからメリットを得られる操作の数は比較的少なく、内部 API がこの特性を活用するように設計されていない場合はさらに制限されます。

   • いくつかのシナリオでは参加しますが、内部のサポートが必要になります。
   • ウィンドウ関数は、一致するパーティショナーで呼び出されます。上記と同じですが、単一のウィンドウ定義に制限されます。ただし、内部的にすでにパーティション化されているため、事前のパーティション化は冗長になる可能性があります。
   • 単純な集計- 一時バッファのメモリフットプリントを削減することは可能ですが、全体的なコストは大幅に高くなります。 (現在の動作) と(事前パーティション分割)GROUP BYとほぼ同等です。実際には役に立たない可能性があります。groupByKey.mapValues(_.reduce)reduceByKey
   • によるデータ圧縮SqlContext.cacheTable。ランレングス符号化を使用しているようなので、適用するとOrderedRDDFunctions.repartitionAndSortWithinPartitions圧縮率が向上する可能性があります。

2. パフォーマンスはキーの分布に大きく依存します。分布が偏っていると、リソースの使用率が最適ではなくなります。最悪の場合、ジョブを完了することがまったく不可能になります。

3. 高レベルの宣言型APIを使用する主な目的は、低レベルの実装の詳細から自分自身を分離することです。すでに述べたように、フォローそしてロミ・クンツマン最適化は触媒オプティマイザーこれはかなり洗練されたものであり、内部をもっと深く調べなければ、簡単に改善できるとは思えません。

関連概念

JDBC ソースによるパーティション分割:

JDBCデータソースのサポートpredicates口論次のように使用できます。

sqlContext.read.jdbc(url, table, Array("foo = 1", "foo = 3"), props)

述語ごとに 1 つの JDBC パーティションが作成されます。個々の述語を使用して作成されたセットが分離されていない場合は、結果のテーブルに重複が表示されることに注意してください。

partitionBy方法DataFrameWriter:

Sparkは、書き込み時にデータを「パーティション」するために使用できるメソッドをDataFrameWriter提供します。提供された列セットを使用して書き込み時にデータを分離します。partitionBy

val df = Seq(
  ("foo", 1.0), ("bar", 2.0), ("foo", 1.5), ("bar", 2.6)
).toDF("k", "v")

df.write.partitionBy("k").json("/tmp/foo.json")

val df1 = sqlContext.read.schema(df.schema).json("/tmp/foo.json")
df1.where($"k" === "bar")

val cnts = df1.groupBy($"k").sum()

cnts.explain

// == Physical Plan ==
// TungstenAggregate(key=[k#90], functions=[(sum(v#91),mode=Final,isDistinct=false)], output=[k#90,sum(v)#93])
// +- TungstenExchange hashpartitioning(k#90,200), None
//    +- TungstenAggregate(key=[k#90], functions=[(sum(v#91),mode=Partial,isDistinct=false)], output=[k#90,sum#99])
//       +- Scan JSONRelation[k#90,v#91] InputPaths: file:/tmp/foo.json

// Temporarily disable broadcast joins
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)

df.write.bucketBy(42, "k").saveAsTable("df1")
val df2 = Seq(("A", -1.0), ("B", 2.0)).toDF("k", "v2")
df2.write.bucketBy(42, "k").saveAsTable("df2")

// == Physical Plan ==
// *Project [k#41, v#42, v2#47]
// +- *SortMergeJoin [k#41], [k#46], Inner
//    :- *Sort [k#41 ASC NULLS FIRST], false, 0
//    :  +- *Project [k#41, v#42]
//    :     +- *Filter isnotnull(k#41)
//    :        +- *FileScan parquet default.df1[k#41,v#42] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/spark-warehouse/df1], PartitionFilters: [], PushedFilters: [IsNotNull(k)], ReadSchema: struct<k:string,v:int>
//    +- *Sort [k#46 ASC NULLS FIRST], false, 0
//       +- *Project [k#46, v2#47]
//          +- *Filter isnotnull(k#46)
//             +- *FileScan parquet default.df2[k#46,v2#47] Batched: true, Format: Parquet, Location: InMemoryFileIndex[file:/spark-warehouse/df2], PartitionFilters: [], PushedFilters: [IsNotNull(k)], ReadSchema: struct<k:string,v2:double>