当我执行以下命令时:
scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4).partitionBy(new HashPartitioner(10)).persist()
rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[10] at partitionBy at :22
scala> rdd.partitions.size
res9: Int = 10
scala> rdd.partitioner.isDefined
res10: Boolean = true
scala> rdd.partitioner.get
res11: org.apache.spark.Partitioner = org.apache.spark.HashPartitioner@a
它说有10个分区,分区完成使用HashPartitioner.但是当我执行以下命令时:
scala> val rdd = sc.parallelize(List((1,2),(3,4),(3,6)),4)
...
scala> rdd.partitions.size
res6: Int = 4
scala> rdd.partitioner.isDefined
res8: Boolean = false
它说有4个分区,并且没有定义分区器.那么,什么是Spark中的默认分区方案？/如何在第二种情况下对数据进行分区？
你必须区分两个不同的东西:
分区为在分区之间分配数据,取决于仅限于PairwiseRDDs(RDD[(T, U)])的密钥值.这将在分区和可在给定分区上找到的密钥集之间创建关系.
将分割输入分区为多个分区,其中数据被简单地划分为包含连续记录的块,以实现分布式计算.确切的逻辑取决于特定的来源,但它是记录的数量或块的大小.
如果parallelize数据使用索引在分区之间均匀分布.如果HadoopInputFormats(喜欢textFile)它取决于mapreduce.input.fileinputformat.split.minsize/ 等属性mapreduce.input.fileinputformat.split.maxsize.
因此,默认分区方案根本就是无,因为分区不适用于所有RDD.对于需要在分区操作PairwiseRDD(aggregateByKey,reduceByKey等)默认方法是使用散列分区.
京公网安备 11010802040832号 | 京ICP备19059560号-6 