理解RDD的Partition

比如读一份本地文件成为RDD,应该分成多少个区? 如何分区? 就是分区的数量以及分区的方法

 private val value: RDD[String] = sc.textFile("D:/tmp")

也可以在读取时候指定分区数量

private val value: RDD[String] = sc.textFile("D:/tmp",2)

看一下textFile方法说明

从HDFS，本地文件系统（在所有节点上都可用）或任何Hadoop支持的文件系统URI中读取文本文件，并将其作为字符串的RDD返回。
参数：
path –支持的文件系统上文本文件的路径
minPartitions –建议的最小RDD分区数

在上述中,我们指定了分区数量是2,那么这2个分区时如何划分的呢?依据什么样的分区函数进行划分?

查看源码可以知道其调用了 hadoopFile方法

 def textFile(path: String,minPartitions: Int = defaultMinPartitions): RDD[String] = withScope {assertNotStopped()hadoopFile(path, classOf[TextInputFormat], classOf[LongWritable], classOf[Text],minPartitions).map(pair => pair._2.toString).setName(path)}

看一下 hadoopFile方法的源码,返回值RDD[(K, V)],就是说返回的是一个kv对型的RDD对象.如何实现呢>关键是new了一个Hadoop RDD.

def hadoopFile[K, V](path: String,inputFormatClass: Class[_ <: InputFormat[K, V]],keyClass: Class[K],valueClass: Class[V],minPartitions: Int = defaultMinPartitions): RDD[(K, V)] = withScope {assertNotStopped()// This is a hack to enforce loading hdfs-site.xml.// See SPARK-11227 for details.FileSystem.getLocal(hadoopConfiguration)// A Hadoop configuration can be about 10 KB, which is pretty big, so broadcast it.val confBroadcast = broadcast(new SerializableConfiguration(hadoopConfiguration))val setInputPathsFunc = (jobConf: JobConf) => FileInputFormat.setInputPaths(jobConf, path)new HadoopRDD(this,confBroadcast,Some(setInputPathsFunc),inputFormatClass,keyClass,valueClass,minPartitions).setName(path)}

看一下hadoopFile方法的帮助信息,

使用任意InputFormat获取Hadoop文件的RDD
参数：
path –输入数据文件的目录，该路径可以是逗号分隔的路径，作为输入列表
inputFormatClass –要读取的数据的存储格式
keyClass –与inputFormatClass参数关联的键的类
valueClass –与inputFormatClass参数关联的值的类
minPartitions –建议的最小RDD分区数
返回：
RDD of tuples of key and corresponding value
注意：
由于Hadoop的RecordReader类为每个记录重复使用相同的Writable对象，因此，直接缓存返回的RDD或将其直接传递给聚合或混搭操作将创建对同一对象的许多引用。 如果计划直接缓存，排序或聚合Hadoop可写对象，则应首先使用map函数复制它们。

那么Hadoop RDD. 又是什么呢?看下帮助信息

:: DeveloperApi ::一种RDD，它提供了核心功能，可以使用较旧的MapReduce API（org.apache.hadoop.mapred）读取Hadoop中存储的数据（例如，HDFS中的文件，HBase或S3中的源）。
参数：
sc –与RDD关联的SparkContext。
broadcastedConf –常规Hadoop配置或其子类。如果包含的变量引用JobConf的实例，则该JobConf将用于Hadoop作业。否则，将使用随附的配置在每个从站上创建一个新的JobConf。
initLocalJobConfFuncOpt –可选的闭包，用于初始化HadoopRDD创建的任何JobConf。
inputFormatClass –要读取的数据的存储格式。
keyClass –与inputFormatClass关联的键的类。
valueClass –与inputFormatClass关联的值的类。
minPartitions –生成的最少HadoopRDD分区数（Hadoop拆分）。
注意：
不建议直接实例化此类，请使用org.apache.spark.SparkContext.hadoopRDD（）

在HadoopRDD中有个方法,如下
可以看出,Array的数量为inputSplits.size,而这个数量又是通过inputFormat.getSplits(jobConf, minPartitions)方法获取的. 所以RDD的分区数量实际上和MR一样,是依据输入文件的大小确定的,假设在MR中,分片大小为128M,文件我5G,那么mapTask数量为5000/128 ,在这里,就是分区数量为5000/128. 当然,这是默认情况下,如果你自己指定了分区数量,那就是你TextFile方法中传入的那个数量.

override def getPartitions: Array[Partition] = {val jobConf = getJobConf()// add the credentials here as this can be called before SparkContext initializedSparkHadoopUtil.get.addCredentials(jobConf)val inputFormat = getInputFormat(jobConf)val inputSplits = inputFormat.getSplits(jobConf, minPartitions)val array = new Array[Partition](inputSplits.size)for (i <- 0 until inputSplits.size) {array(i) = new HadoopPartition(id, i, inputSplits(i))}array}

分区数量已经知道如何确定,那么如何分区呢?

理解RDD的Partition相关推荐

Spark快速入门系列（三）深入理解RDD
这里写目录标题深入 RDD 案例再谈 RDD RDD 为什么会出现? RDD 的特点什么叫做弹性分布式数据集总结: RDD 的五大属性深入 RDD 目标深入理解 RDD 的内在逻辑, 以及 ...
spark任务shell运行_《Spark快速大数据分析》- 根据简单例子理解RDD
1. RDD简介 RDD,弹性分布式数据集(Resiliennt Distributed Datasets),是Spark中最重要的核心概念,是Spark应用中存储数据的数据结构. RDD 其实就是分 ...
彻底理解kafka中partition和消费者对应关系
1个partition只能被同组的一个consumer消费,同组的consumer则起到均衡效果消费者多于partition topic: test 只有一个partition 创建一个topic- ...
在kafka中，可以这么理解topic，partition，broker
一个topic,代表逻辑上的一个业务数据集,比如按数据库里不同表的数据操作消息区分放入不同topic,订单相关操作消息放入订单topic,用户相关操作消息放入用户topic,对于大型网站来说,后端数据 ...
spark：Task，Partition，RDD、节点数、Executor数、core数目的关系和Application，Driver，Job，Task，Stage理解
梳理一下Spark中关于并发度涉及的几个概念File,Block,Split,Task,Partition,RDD以及节点数.Executor数.core数目的关系. 输入可能以多个文件的形式存储在H ...
Spark中Task，Partition，RDD、节点数、Executor数、core数目（线程池）、mem数
Spark中Task,Partition,RDD.节点数.Executor数.core数目的关系和Application,Driver,Job,Task,Stage理解 from:https://bl ...
理解Spark的核心RDD
与许多专有的大数据处理平台不同,Spark建立在统一抽象的RDD之上,使得它可以以基本一致的方式应对不同的大数据处理场景,包括MapReduce,Streaming,SQL,Machine Learn ...
深入理解Spark 2.1 Core （一）：RDD的原理与源码分析
摘要本文提出了分布式内存抽象的概念--弹性分布式数据集(RDD,Resilient Distributed Datasets),它具备像MapReduce等数据流模型的容错特性,并且允许开发人员在大 ...
Spark基础之：rdd的特性，DAG，Stage的理解
rdd的特性,DAG,Stage的理解 RDD 结构化理解 RDD的数据集与Partitions Partitioner Dependencies与Lineage NarrowDependency与S ...

理解RDD的Partition

理解RDD的Partition相关推荐

最新文章

热门文章