Cómo definir la partición de DataFrame?

Spark > = 2.3.0

SPARK-22614 expone particiones de rango.

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 expone particiones de formato externo en la API de origen de datos v2.

Spark > = 1.6.0

En Spark >= 1.6 es posible usar particiones por columna para consultas y almacenamiento en caché. Véase: SPARK-11410 y SPARK-4849 utilizando el método repartition:

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]

A diferencia de RDDs Spark Dataset (incluyendo Dataset[Row] a.k.a DataFrame) no puede utilice particionador personalizado como por ahora. Por lo general, puede abordar eso creando una columna de partición artificial, pero no le dará la misma flexibilidad.

Spark

Una cosa que puede hacer es pre-particionar los datos de entrada antes de crear un 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)

Dado que DataFrame la creación a partir de un RDD requiere solo una fase de mapa simple, se debe preservar el diseño de la partición existente*:

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

De la misma manera que puede reparticionar existente DataFrame:

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

Así que parece que no es imposible. La pregunta sigue siendo si tiene sentido en absoluto. Voy a argumentar que la mayoría de las veces no lo hace:

1. Reparticionar es un proceso costoso. En un escenario típico, la mayoría de los datos deben ser serializados, mezclados y deserializados. Por otro lado, el número de operaciones que pueden beneficiarse de datos pre-particionados es relativamente pequeño y se limita aún más si la API interna no está diseñada para aprovechar esto propiedad.

   • se une en algunos escenarios, pero requeriría un soporte interno, {[76]]}
   • window functions llama con el particionador correspondiente. Igual que el anterior, limitado a una definición de ventana única. Sin embargo, ya está particionado internamente, por lo que el pre-particionado puede ser redundante,
   • agregaciones simples con GROUP BY - es posible reducir la huella de memoria de los búferes temporales**, pero el costo general es mucho mayor. Más o menos equivalente a groupByKey.mapValues(_.reduce) (comportamiento actual) vs reduceByKey (pre-partición). Poco probable que sea útil en la práctica.
   • compresión de datos con SqlContext.cacheTable. Dado que parece que está utilizando la codificación de longitud de ejecución, la aplicación de OrderedRDDFunctions.repartitionAndSortWithinPartitions podría mejorar la relación de compresión.

2. El rendimiento depende en gran medida de la distribución de las teclas. Si está sesgado, resultará en una utilización de recursos subóptima. En el peor de los casos será imposible terminar el trabajo en absoluto.

3. Todo un punto de usar una alta la API declarativa de nivel es aislarse de los detalles de implementación de bajo nivel. Como ya se mencionó por @dwysakowicz y @RomiKuntsman una optimización es un trabajo del Catalyst Optimizer . Es una bestia bastante sofisticada y realmente dudo que pueda mejorar fácilmente sin sumergirse mucho más en sus componentes internos.

Conceptos relacionados

Particionamiento con fuentes JDBC :

Soporte de fuentes de datos JDBC predicates argumento. Se puede utilizar de la siguiente manera:

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

Crea una sola partición JDBC por predicado. Tenga en cuenta que si los conjuntos creados utilizando predicados individuales no son disjuntos, verá duplicados en la tabla resultante.

partitionBy método en DataFrameWriter:

Spark DataFrameWriter proporciona el método partitionBy que se puede usar para "particionar" los datos al escribir. Separa los datos al escribir utilizando el conjunto proporcionado de columnas

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

Esto habilita predicado empuje hacia abajo en leer para consultas basadas en clave:

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

, Pero no es equivalente a DataFrame.repartition. En particular, agregaciones como:

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

Todavía requerirá 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

bucketBy método en DataFrameWriter (Spark >= 2.0):

bucketBy tiene aplicaciones similares a partitionBy pero solo está disponible para tablas (saveAsTable). La información de bucketing se puede usar para optimizar joins:

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

* By partition layout Me refiero solo a un dato distribución. partitioned RDD ya no tiene un particionador. ** Asumiendo que no hay proyección temprana. Si la agregación cubre solo un pequeño subconjunto de columnas, probablemente no haya ganancia alguna.