Bagaimana cara mendefinisikan partisi DataFrame?

Saya sudah mulai menggunakan Spark SQL dan DataFrames di Spark 1.4.0. Saya ingin mendefinisikan pemartisi khusus di DataFrames, di Scala, tetapi tidak melihat cara melakukan ini.

Salah satu tabel data yang saya kerjakan berisi daftar transaksi, berdasarkan akun, silimar ke contoh berikut.

Account   Date       Type       Amount
1001    2014-04-01  Purchase    100.00
1001    2014-04-01  Purchase     50.00
1001    2014-04-05  Purchase     70.00
1001    2014-04-01  Payment    -150.00
1002    2014-04-01  Purchase     80.00
1002    2014-04-02  Purchase     22.00
1002    2014-04-04  Payment    -120.00
1002    2014-04-04  Purchase     60.00
1003    2014-04-02  Purchase    210.00
1003    2014-04-03  Purchase     15.00

Setidaknya pada awalnya, sebagian besar kalkulasi akan terjadi antara transaksi di dalam akun. Jadi saya ingin data dipartisi sehingga semua transaksi untuk akun berada di partisi Spark yang sama.

Tapi saya tidak melihat cara untuk mendefinisikan ini. Kelas DataFrame memiliki metode yang disebut 'repartition (Int)', di mana Anda dapat menentukan jumlah partisi yang akan dibuat. Tapi saya tidak melihat metode apa pun yang tersedia untuk menentukan pemartisi khusus untuk DataFrame, seperti yang dapat ditentukan untuk RDD.

Data sumber disimpan di Parket. Saya melihat bahwa ketika menulis DataFrame ke Parquet, Anda dapat menentukan kolom untuk dipartisi, jadi mungkin saya bisa memberi tahu Parquet untuk mempartisi datanya dengan kolom 'Akun'. Tapi mungkin ada jutaan akun, dan jika saya memahami Parket dengan benar, ini akan membuat direktori berbeda untuk setiap Akun, jadi itu tidak terdengar seperti solusi yang masuk akal.

Apakah ada cara untuk membuat Spark mempartisi DataFrame ini sehingga semua data untuk sebuah Akun berada di partisi yang sama?

Percikan> = 2.3.0

SPARK-22614 memperlihatkan partisi jarak.

val partitionedByRange = df.repartitionByRange(42, $"k")

partitionedByRange.explain
// == Parsed Logical Plan ==
// 'RepartitionByExpression ['k ASC NULLS FIRST], 42
// +- AnalysisBarrier Project [_1#2 AS k#5, _2#3 AS v#6]
// 
// == Analyzed Logical Plan ==
// k: string, v: int
// RepartitionByExpression [k#5 ASC NULLS FIRST], 42
// +- Project [_1#2 AS k#5, _2#3 AS v#6]
//    +- LocalRelation [_1#2, _2#3]
// 
// == Optimized Logical Plan ==
// RepartitionByExpression [k#5 ASC NULLS FIRST], 42
// +- LocalRelation [k#5, v#6]
// 
// == Physical Plan ==
// Exchange rangepartitioning(k#5 ASC NULLS FIRST, 42)
// +- LocalTableScan [k#5, v#6]

SPARK-22389 memperlihatkan partisi format eksternal di Data Source API v2 .

Percikan> = 1.6.0

Dalam Spark> = 1.6 dimungkinkan untuk menggunakan partisi menurut kolom untuk kueri dan caching. Lihat: SPARK-11410 dan SPARK-4849 menggunakan repartitionmetode:

val df = Seq(
  ("A", 1), ("B", 2), ("A", 3), ("C", 1)
).toDF("k", "v")

val partitioned = df.repartition($"k")
partitioned.explain

// scala> df.repartition($"k").explain(true)
// == Parsed Logical Plan ==
// 'RepartitionByExpression ['k], None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- LogicalRDD [_1#5,_2#6], MapPartitionsRDD[3] at rddToDataFrameHolder at <console>:27
// 
// == Analyzed Logical Plan ==
// k: string, v: int
// RepartitionByExpression [k#7], None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- LogicalRDD [_1#5,_2#6], MapPartitionsRDD[3] at rddToDataFrameHolder at <console>:27
// 
// == Optimized Logical Plan ==
// RepartitionByExpression [k#7], None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- LogicalRDD [_1#5,_2#6], MapPartitionsRDD[3] at rddToDataFrameHolder at <console>:27
// 
// == Physical Plan ==
// TungstenExchange hashpartitioning(k#7,200), None
// +- Project [_1#5 AS k#7,_2#6 AS v#8]
//    +- Scan PhysicalRDD[_1#5,_2#6]

Tidak seperti RDDsSpark Dataset(termasuk Dataset[Row]alias DataFrame) tidak dapat menggunakan pemartisi khusus seperti untuk saat ini. Anda biasanya dapat mengatasinya dengan membuat kolom partisi buatan tetapi itu tidak akan memberi Anda fleksibilitas yang sama.

Spark <1.6.0:

Satu hal yang dapat Anda lakukan adalah memasukkan data masukan prapartisi sebelum Anda membuat file DataFrame

import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.spark.HashPartitioner

val schema = StructType(Seq(
  StructField("x", StringType, false),
  StructField("y", LongType, false),
  StructField("z", DoubleType, false)
))

val rdd = sc.parallelize(Seq(
  Row("foo", 1L, 0.5), Row("bar", 0L, 0.0), Row("??", -1L, 2.0),
  Row("foo", -1L, 0.0), Row("??", 3L, 0.6), Row("bar", -3L, 0.99)
))

val partitioner = new HashPartitioner(5) 

val partitioned = rdd.map(r => (r.getString(0), r))
  .partitionBy(partitioner)
  .values

val df = sqlContext.createDataFrame(partitioned, schema)

Karena DataFramepembuatan dari RDDhanya memerlukan fase peta sederhana, tata letak partisi yang ada harus dipertahankan *:

assert(df.rdd.partitions == partitioned.partitions)

Cara yang sama Anda dapat mempartisi ulang yang ada DataFrame:

sqlContext.createDataFrame(
  df.rdd.map(r => (r.getInt(1), r)).partitionBy(partitioner).values,
  df.schema
)

Jadi sepertinya bukan tidak mungkin. Pertanyaannya tetap apakah itu masuk akal. Saya akan berargumen bahwa seringkali tidak:

1. Proses partisi ulang adalah proses yang mahal. Dalam skenario umum, sebagian besar data harus diserialisasi, dikocok, dan dideserialisasi. Di sisi lain, jumlah operasi yang dapat memanfaatkan data yang dipartisi sebelumnya relatif kecil dan selanjutnya dibatasi jika API internal tidak dirancang untuk memanfaatkan properti ini.

   • bergabung dalam beberapa skenario, tetapi itu akan membutuhkan dukungan internal,
   • fungsi jendela memanggil dengan pemartisi yang cocok. Sama seperti di atas, terbatas pada definisi satu jendela. Ini sudah dipartisi secara internal, jadi pra-partisi mungkin berlebihan,
   • agregasi sederhana dengan GROUP BY- dimungkinkan untuk mengurangi jejak memori dari buffer sementara **, tetapi biaya keseluruhan jauh lebih tinggi. Lebih atau kurang setara dengan groupByKey.mapValues(_.reduce)(perilaku saat ini) vs reduceByKey(pra-partisi). Sepertinya tidak berguna dalam praktik.
   • kompresi data dengan SqlContext.cacheTable. Karena tampaknya ini menggunakan pengkodean panjang proses, penerapan OrderedRDDFunctions.repartitionAndSortWithinPartitionsdapat meningkatkan rasio kompresi.

2. Performa sangat bergantung pada distribusi kunci. Jika miring maka akan menghasilkan pemanfaatan sumber daya yang kurang optimal. Dalam skenario kasus terburuk, tidak mungkin menyelesaikan pekerjaan sama sekali.

3. Inti dari penggunaan API deklaratif tingkat tinggi adalah mengisolasi diri Anda dari detail penerapan tingkat rendah. Seperti yang telah disebutkan oleh @dwysakowicz dan @RomiKuntsman, pengoptimalan adalah tugas dari Catalyst Optimizer . Ini adalah binatang yang cukup canggih dan saya benar-benar ragu Anda dapat dengan mudah memperbaikinya tanpa menyelam lebih dalam ke bagian dalamnya.

Konsep terkait

Mempartisi dengan sumber JDBC :

Sumber data JDBC mendukung predicatesargumen . Ini dapat digunakan sebagai berikut:

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

Ini menciptakan satu partisi JDBC per predikat. Perlu diingat bahwa jika set yang dibuat menggunakan predikat individual tidak terputus-putus, Anda akan melihat duplikat di tabel yang dihasilkan.

partitionBymetode dalamDataFrameWriter :

Spark DataFrameWritermenyediakan partitionBymetode yang dapat digunakan untuk "mempartisi" data saat menulis. Ini memisahkan data saat menulis menggunakan kumpulan kolom yang disediakan

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")

Ini memungkinkan predikat menekan baca untuk kueri berdasarkan kunci:

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

tapi itu tidak sama dengan DataFrame.repartition. Dalam agregasi tertentu seperti:

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

masih membutuhkan TungstenExchange:

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

bucketBymetode diDataFrameWriter (Spark> = 2.0):

bucketBymemiliki aplikasi yang serupa partitionBytetapi hanya tersedia untuk tables ( saveAsTable). Informasi pengelompokan dapat digunakan untuk mengoptimalkan gabungan:

// 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>

* Dengan tata letak partisi yang saya maksud hanya distribusi data. partitionedRDD tidak lagi menjadi pemartisi. ** Dengan asumsi tidak ada proyeksi awal. Jika agregasi hanya mencakup subset kecil dari kolom, mungkin tidak ada keuntungan sama sekali.

Di Spark <1.6 Jika Anda membuat HiveContext, bukan yang lama biasa, SqlContextAnda dapat menggunakan HiveQL DISTRIBUTE BY colX... (memastikan setiap pengecil N mendapatkan rentang x yang tidak tumpang tindih) & CLUSTER BY colX...(pintasan untuk Distribusikan Berdasarkan dan Urutkan Berdasarkan) misalnya;

df.registerTempTable("partitionMe")
hiveCtx.sql("select * from partitionMe DISTRIBUTE BY accountId SORT BY accountId, date")

Tidak yakin bagaimana hal ini cocok dengan api Spark DF. Kata kunci ini tidak didukung dalam SqlContext normal (perhatikan Anda tidak perlu memiliki penyimpanan meta sarang untuk menggunakan HiveContext)

EDIT: Spark 1.6+ sekarang memiliki ini di API DataFrame asli

Jadi untuk memulai dengan beberapa jenis jawaban :) - Anda tidak bisa

Saya bukan seorang ahli, tetapi sejauh yang saya mengerti DataFrames, mereka tidak sama dengan rdd dan DataFrame tidak memiliki hal seperti Partitioner.

Umumnya ide DataFrame adalah menyediakan level abstraksi lain yang menangani masalah itu sendiri. Kueri di DataFrame diterjemahkan ke dalam rencana logis yang selanjutnya diterjemahkan ke operasi di RDD. Pemartisian yang Anda sarankan mungkin akan diterapkan secara otomatis atau setidaknya harus diterapkan.

Jika Anda tidak mempercayai SparkSQL yang akan memberikan beberapa jenis pekerjaan yang optimal, Anda selalu dapat mengubah DataFrame ke RDD [Row] seperti yang disarankan di komentar.

Gunakan DataFrame yang dikembalikan oleh:

yourDF.orderBy(account)

Tidak ada cara eksplisit untuk menggunakan partitionByDataFrame, hanya di PairRDD, tetapi saat Anda mengurutkan DataFrame, itu akan digunakan di LogicalPlan dan itu akan membantu saat Anda perlu membuat kalkulasi pada setiap Akun.

Saya baru saja menemukan masalah yang sama persis, dengan kerangka data yang ingin saya partisi berdasarkan akun. Saya berasumsi bahwa ketika Anda mengatakan "ingin data dipartisi sehingga semua transaksi untuk akun berada di partisi Spark yang sama", Anda menginginkannya untuk skala dan kinerja, tetapi kode Anda tidak bergantung padanya (seperti menggunakan mapPartitions()dll), bukan?

Saya bisa melakukan ini menggunakan RDD. Tetapi saya tidak tahu apakah ini solusi yang dapat diterima untuk Anda. Setelah DF tersedia sebagai RDD, Anda dapat mengajukan permohonan repartitionAndSortWithinPartitionsuntuk melakukan partisi ulang data.

Ini contoh yang saya gunakan:

class DatePartitioner(partitions: Int) extends Partitioner {

  override def getPartition(key: Any): Int = {
    val start_time: Long = key.asInstanceOf[Long]
    Objects.hash(Array(start_time)) % partitions
  }

  override def numPartitions: Int = partitions
}

myRDD
  .repartitionAndSortWithinPartitions(new DatePartitioner(24))
  .map { v => v._2 }
  .toDF()
  .write.mode(SaveMode.Overwrite)