Iskra> = 2.3.0
SPARK-22614 ujawnia podział zakresu.
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 ujawnia zewnętrzne partycjonowanie formatu w Data Source API v2 .
Iskra> = 1.6.0
W Spark> = 1.6 możliwe jest użycie partycjonowania według kolumn do zapytań i buforowania. Widzieć: SPARK-11410 i zapłonem 4849 stosując repartitionmetodę:
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]
W przeciwieństwie do RDDsSpark Dataset(w tym Dataset[Row]aka DataFrame) nie można teraz używać niestandardowego partycjonera. Zazwyczaj można temu zaradzić, tworząc sztuczną kolumnę partycjonowania, ale nie zapewni to takiej samej elastyczności.
Iskra <1.6.0:
Jedną rzeczą, którą możesz zrobić, jest wstępne podzielenie danych wejściowych na partycje przed utworzeniem pliku 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)
Ponieważ DataFrametworzenie z poziomu RDDwymaga tylko prostej fazy mapy, należy zachować istniejący układ partycji *:
assert(df.rdd.partitions == partitioned.partitions)
W ten sam sposób możesz ponownie podzielić istniejące DataFrame:
sqlContext.createDataFrame(
df.rdd.map(r => (r.getInt(1), r)).partitionBy(partitioner).values,
df.schema
)
Więc wygląda na to, że nie jest to niemożliwe. Pozostaje pytanie, czy w ogóle ma to sens. Będę argumentował, że w większości przypadków tak nie jest:
Zmiana partycji to kosztowny proces. W typowym scenariuszu większość danych musi być serializowana, tasowana i deserializowana. Z drugiej strony liczba operacji, które mogą skorzystać na danych wstępnie podzielonych na partycje, jest stosunkowo niewielka i jest dodatkowo ograniczona, jeśli wewnętrzny interfejs API nie jest zaprojektowany do wykorzystania tej właściwości.
- łączy się w niektórych scenariuszach, ale wymagałoby wewnętrznego wsparcia,
- wywołania funkcji okna z odpowiednim partycjonerem. Tak samo jak powyżej, ograniczone do definicji pojedynczego okna. Jest już jednak podzielony wewnętrznie, więc wstępne partycjonowanie może być zbędne,
- proste agregacje z
GROUP BY- możliwe jest zmniejszenie rozmiaru pamięci tymczasowych buforów **, ale ogólny koszt jest znacznie wyższy. Mniej więcej odpowiednik groupByKey.mapValues(_.reduce)(obecne zachowanie) vs reduceByKey(wstępne partycjonowanie). W praktyce mało prawdopodobne.
- kompresja danych z
SqlContext.cacheTable. Ponieważ wygląda na to, że używa kodowania długości serii, zastosowanie OrderedRDDFunctions.repartitionAndSortWithinPartitionsmoże poprawić współczynnik kompresji.
Wydajność w dużym stopniu zależy od dystrybucji kluczy. Jeśli jest przekrzywiony, spowoduje to nieoptymalne wykorzystanie zasobów. W najgorszym przypadku ukończenie pracy w ogóle nie będzie możliwe.
- Celem używania deklaratywnego interfejsu API wysokiego poziomu jest odizolowanie się od szczegółów implementacji niskiego poziomu. Jak już wspomnieli @dwysakowicz i @RomiKuntsman , optymalizacja jest zadaniem Catalyst Optimizer . Jest to dość wyrafinowana bestia i naprawdę wątpię, by można było ją łatwo poprawić bez głębszego zanurzenia się w jej wnętrzności.
Pojęcia pokrewne
Partycjonowanie ze źródłami JDBC :
Źródła danych JDBC obsługują predicatesargument . Można go używać w następujący sposób:
sqlContext.read.jdbc(url, table, Array("foo = 1", "foo = 3"), props)
Tworzy jedną partycję JDBC na predykat. Pamiętaj, że jeśli zestawy utworzone przy użyciu indywidualnych predykatów nie są rozłączne, w wynikowej tabeli zobaczysz duplikaty.
partitionBy metoda w DataFrameWriter :
Spark DataFrameWriterudostępnia partitionBymetodę, której można użyć do „partycjonowania” danych podczas zapisu. Oddziela dane przy zapisie za pomocą dostarczonego zestawu kolumn
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")
Umożliwia to przesunięcie predykatu w dół podczas odczytu dla zapytań opartych na kluczu:
val df1 = sqlContext.read.schema(df.schema).json("/tmp/foo.json")
df1.where($"k" === "bar")
ale to nie jest równoważne DataFrame.repartition. W szczególności agregacje takie jak:
val cnts = df1.groupBy($"k").sum()
nadal będzie wymagać 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
bucketBymetoda wDataFrameWriter (Spark> = 2.0):
bucketByma podobne aplikacje, partitionByale jest dostępny tylko dla tabel ( saveAsTable). Informacje o zasobnikach można wykorzystać do optymalizacji sprzężeń:
// 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>
* Przez układ partycji mam na myśli tylko dystrybucję danych. partitionedRDD nie ma już programu do partycjonowania. ** Zakładając brak wczesnej prognozy. Jeśli agregacja obejmuje tylko niewielki podzbiór kolumn, prawdopodobnie nie ma żadnego zysku.