Spark分区
1分区:
- 自定义分区在对RDD数据进行分区时,默认使用的是
HashPartitioner
,该partitioner对key进行哈希,然后mod 上分区数目,mod的结果相同的就会被分到同一个partition中如果嫌HashPartitioner
功能单一,可以自定义partitioner。
2.自定义分区
-
实现自定义分区分为三个步骤:
1.继承org.apache.spark.partitioner 2.重写numPartitons和getPartition方法 3.在分区使用
3.代码示例:
- TestPartitionerMain.scala
package cn.wc.rdd
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object TestPartitionerMain {
def main(args: Array[String]): Unit = {
val sparkConf: SparkConf = new SparkConf().setAppName("wcSpark01").setMaster("local[2]")
val sc = new SparkContext(sparkConf)
sc.setLogLevel("warn")
// 造点数据
val data: RDD[String] = sc.parallelize(List("hadoop", "hdfs", "hive", "spark", "flume", "kafka", "sqoop", "python", "java"))
// 获取每一个元素的长度,封装成一个元组
val wordLengthRDD: RDD[(String, Int)] = data.map(x => (x, x.length))
// 对RDD进行自定义分区 这里可以自定义分区数量
val result:RDD[(String,Int)] = wordLengthRDD.partitionBy(new NewPartitioner(3))
result.saveAsTextFile("./data")
sc.stop()
}
}
-
NewPartitioner.scala
package cn.wc.rdd import org.apache.spark.Partitioner class NewPartitioner(num:Int) extends Partitioner { // 用于指定RDD分区数 override def numPartitions: Int = { num } // 指定分区策略 数据后期按照自定义规则指定到分区中 override def getPartition(key: Any): Int = { // key 是单词 val partiton:Int = Math.abs(key.toString.hashCode % num) partiton } }
? 执行完会在当前目录data下生成3个分区文件
-
自定义分区:
package cn.wc.rdd import org.apache.spark.Partitioner class MyPartitioner(num:Int) extends Partitioner{ // 用于指定RDD分区数 override def numPartitions: Int = { num } // 指定分区策略 数据后期按照自定义规则指定到分区中 override def getPartition(key: Any): Int = { // 按照单位的长度指定进入到对应分区中 val length:Int = key.toString.length length match { case 4 => 0 case 5 => 1 case 6 => 2 case _ => 0 } } }