spark– 如何定义DataFrame的分区？

在Spark> = 1.6中，可以使用按列分区查询和缓存。参见：SPARK-11410和SPARK-4849使用重分区方法：

 

val df = sc.parallelize(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]

与RDDs不同，Spark Dataset(包括Dataset [Row] a.k.a DataFrame)现在不能使用自定义分区器。你通常可以通过创建一个人工分区列来解决这个问题，但它不会给你相同的灵活性。

Spark< 1.6.0： 您可以做的一件事是在创建DataFrame之前预分区输入数据

import org.apache.spark.sql.types._
import org.apache.spark.sql.Rowimport 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 =newHashPartitioner(5) 

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

val df = sqlContext.createDataFrame(partitioned, schema)

由于从RDD创建DataFrame只需要一个简单的映射阶段现有的分区布局应该保留*：

 

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

以同样的方式，您可以重新分区现有的DataFrame：

 

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

所以看起来这不是不可能的。问题仍然存在，如果它是有意义的。我会说，大多数时候它不：

>重新分区是一个昂贵的过程。在典型情况下，大多数数据必须序列化，混排和反序列化。另一方面，可以从预分割数据中受益的操作数量相对较少，并且如果内部API未被设计为利用该属性，则进一步受限。

>在某些情况下联接，但它需要内部支持，
>窗口函数调用与匹配分区器。同上，限于一个窗口定义。它已经在内部进行了分区，因此预分区可能是多余的，
>使用GROUP BY的简单聚合 – 可以减少临时缓冲区**的内存占用，但总体成本要高得多。或多或少相当于groupByKey.mapValues(_。reduce)(当前行为)vs reduceByKey(预分区)。不太可能在实践中有用。
>使用SqlContext.cacheTable进行数据压缩。由于它看起来像是使用运行长度编码，应用OrderedRDDFunctions.repartitionAndSortWithinPartitions可以提高压缩率。

>性能高度依赖于密钥的分布。如果它是倾斜的，它将导致次优资源利用。在最坏的情况下，根本不可能完成这项工作。
>使用高级声明性API的一个重点是将自己与低级实现细节隔离开来。正如@dwysakowicz和@RomiKuntsman已经提到的，优化是Catalyst Optimizer的工作。它是一个非常复杂的野兽，我真的怀疑你可以轻松地改进，没有深入到它的内部。

使用JDBC源分区：

JDBC数据源支持predicates argument.它可以如下使用：

 

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

它为每个谓词创建一个JDBC分区。请记住，如果使用单个谓词创建的集合不是不相交的，则会在结果表中看到重复的集合。

DataFrameWriter中的partitionBy方法：

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

但它不等同于DataFrame.repartition。特别是聚合：

 

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

仍将需要TunnstenExchange：

 

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

*分区布局我的意思是只有一个数据分布。分区RDD不再是分区器。
**假设没有早期预测。如果聚合仅覆盖列的小子集，则可能没有任何增益。