码迷,mamicode.com
首页 > 其他好文 > 详细

spark自定义分区器实现

时间:2019-06-23 17:30:57      阅读:138      评论:0      收藏:0      [点我收藏+]

标签:form   override   随机   技术   rdd   创建   vat   一个   foreach   

在spark中,框架默认使用的事hashPartitioner分区器进行对rdd分区,但是实际生产中,往往使用spark自带的分区器会产生数据倾斜等原因,这个时候就需要我们自定义分区,按照我们指定的字段进行分区。具体的流程步骤如下:

1、创建一个自定义的分区类,并继承Partitioner,注意这个partitioner是spark的partitioner

2、重写partitioner中的方法

  override def numPartitions: Int = ???
override def getPartition(key: Any): Int = ???

代码实现:
测试数据集:
cookieid,createtime,pv
cookie1,2015-04-10,1
cookie1,2015-04-11,5
cookie1,2015-04-12,7
cookie1,2015-04-13,3
cookie1,2015-04-14,2
cookie1,2015-04-15,4
cookie1,2015-04-16,4
cookie2,2015-04-10,2
cookie2,2015-04-11,3
cookie2,2015-04-12,5
cookie2,2015-04-13,6
cookie2,2015-04-14,3
cookie2,2015-04-15,9
cookie2,2015-04-16,7

  指定按照第一个字段进行分区

步骤1:
package _core.sourceCodeLearning.partitioner

import org.apache.spark.Partitioner
import scala.collection.mutable.HashMap

/**
  * Author Mr. Guo
  * Create 2019/6/23 - 12:19
  */
class UDFPartitioner(args: Array[String]) extends Partitioner {

  private val partitionMap: HashMap[String, Int] = new HashMap[String, Int]()
  var parId = 0
  for (arg <- args) {
    if (!partitionMap.contains(arg)) {
      partitionMap(arg) = parId
      parId += 1
    }
  }

  override def numPartitions: Int = partitionMap.valuesIterator.length

  override def getPartition(key: Any): Int = {
    val keys: String = key.asInstanceOf[String]
    val sub = keys
    partitionMap(sub)
  }
}

  步骤2:

主类测试:

package _core.sourceCodeLearning.partitioner

import org.apache.spark.{SparkConf, TaskContext}
import org.apache.spark.sql.SparkSession

/**
  * Author Mr. Guo
  * Create 2019/6/23 - 12:21
  */
object UDFPartitionerMain {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName(this.getClass.getSimpleName)
    val ssc = SparkSession
      .builder()
      .config(conf)
      .getOrCreate()
    val sc = ssc.sparkContext
    sc.setLogLevel("WARN")

    val rdd = ssc.sparkContext.textFile("file:///E:\\TestFile\\analyfuncdata.txt")
    val transform = rdd.filter(_.split(",").length == 3).map(x => {
      val arr = x.split(",")
      (arr(0), (arr(1), arr(2)))
    })
    val keys: Array[String] = transform.map(_._1).collect()
    val partiion = transform.partitionBy(new UDFPartitioner(keys))
    partiion.foreachPartition(iter => {
      println(s"**********分区号:${TaskContext.getPartitionId()}***************")
      iter.foreach(r => {
        println(s"分区:${TaskContext.getPartitionId()}###" + r._1 + "\t" + r._2 + "::" + r._2._1)
      })
    })
    ssc.stop()
  }
}

  运行结果:

技术图片

这样就是按照第一个字段进行了分区,当然在分区器的中,对于key是可以根据自己的需求随意的处理,比如添加随机数等等

spark自定义分区器实现

标签:form   override   随机   技术   rdd   创建   vat   一个   foreach   

原文地址:https://www.cnblogs.com/Gxiaobai/p/11073381.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!