spark自定义分区

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

spark自定义分区

上一篇:将xacro格式文件转换为urdf格式并检查格式是否正确常用方式


下一篇:Dart编程--基础语法代码示例