Spark 的join 什么时候是宽依赖什么时候是窄依赖
1准备数据
spark-shell --master spark://linux01:7077 --executor-meg --total-executor-cores 4
val sc: SparkContext = SparkUtils.getSc// 模拟数据产生val rdd1 = sc.parallelize(Array((1, "张三1"),(1, "张三2"),(2, "李四"),(3, "王五"),(4, "Tom"),(5, "Gerry"),(6, "莉莉")), 2)val rdd2 = sc.parallelize(Array((1, "上海"),(2, "北京1"),(2, "北京2"),(3, "南京"),(4, "纽约"),(6, "深圳"),(7, "香港")), 2)
2查看WEBUI
rdd1.join(rdd2).collect -- 宽依赖
rdd1.partitionBy(new HashPartitioner(3)).join(rdd2.partitionBy(new HashPartitioner(3))).collect -- 窄依赖
rdd1.partitionBy(new HashPartitioner(2)).join(rdd2.partitionBy(new HashPartitioner(3))).collect -- 宽依赖
3源码分析
a.首先是默认的join方法,这里使用了一个默认分区器
/**
* Return an RDD containing all pairs of elements with matching keys in `this` and `other`. Each
* pair of elements will be returned as a (k, (v1, v2)) tuple, where (k, v1) is in `this` and
* (k, v2) is in `other`. Performs a hash join across the cluster.
*/
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))] = self.withScope {
join(other, defaultPartitioner(self, other))
}
b.默认分区器,对于第一个join会返回一个以电脑core总数为分区数量的HashPartitioner.第二个join会返回我们设定的HashPartitioner(分区数目3)
def defaultPartitioner(rdd: RDD[_], others: RDD[_]*): Partitioner = {
val rdds = (Seq(rdd) ++ others)
val hasPartitioner = rdds.filter(_.partitioner.exists(_.numPartitions > 0))val hasMaxPartitioner: Option[RDD[_]] = if (hasPartitioner.nonEmpty) {
Some(hasPartitioner.maxBy(_.partitions.length))
} else {
None
}val defaultNumPartitions = if (rdd.context.conf.contains("spark.default.parallelism")) {
rdd.context.defaultParallelism
} else {
rdds.map(_.partitions.length).max
}// If the existing max partitioner is an eligible one, or its partitions number is larger
// than the default number of partitions, use the existing partitioner.
if (hasMaxPartitioner.nonEmpty && (isEligiblePartitioner(hasMaxPartitioner.get, rdds) ||
defaultNumPartitions < hasMaxPartitioner.get.getNumPartitions)) {
hasMaxPartitioner.get.partitioner.get
} else {
new HashPartitioner(defaultNumPartitions)
}
}
c.走到了实际执行的join方法,里面flatMapValues是一个窄依赖,所以说如果有宽依赖应该在cogroup算子中
/**
* Return an RDD containing all pairs of elements with matching keys in `this` and `other`. Each
* pair of elements will be returned as a (k, (v1, v2)) tuple, where (k, v1) is in `this` and
* (k, v2) is in `other`. Uses the given Partitioner to partition the output RDD.
*/
def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))] = self.withScope {
this.cogroup(other, partitioner).flatMapValues( pair =>
for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)
)
}
d.进入cogroup方法中,核心是CoGroupedRDD,根据两个需要join的rdd和一个分区器。由于第一个join的时候,两个rdd都没有分区器,所以在这一步,两个rdd需要先根据传入的分区器进行一次shuffle,因此第一个join是宽依赖。第二个join此时已经分好区了,不需要再再进行shuffle了。所以第二个是窄依赖
/**
* For each key k in `this` or `other`, return a resulting RDD that contains a tuple with the
* list of values for that key in `this` as well as `other`.
*/
def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner)
: RDD[(K, (Iterable[V], Iterable[W]))] = self.withScope {
if (partitioner.isInstanceOf[HashPartitioner] && keyClass.isArray) {
throw new SparkException("HashPartitioner cannot partition array keys.")
}
val cg = new CoGroupedRDD[K](Seq(self, other), partitioner)
cg.mapValues { case Array(vs, w1s) =>
(vs.asInstanceOf[Iterable[V]], w1s.asInstanceOf[Iterable[W]])
}
}
e.两个都打印出OneToOneDependency,是因为在CoGroupedRDD里面,getDependencies方法里面,如果rdd有partitioner就都会返回OneToOneDependency(rdd)。
override def getDependencies: Seq[Dependency[_]] = {
rdds.map { rdd: RDD[_] =>
if (rdd.partitioner == Some(part)) {
logDebug("Adding one-to-one dependency with " + rdd)
new OneToOneDependency(rdd)
} else {
logDebug("Adding shuffle dependency with " + rdd)
new ShuffleDependency[K, Any, CoGroupCombiner](
rdd.asInstanceOf[RDD[_ <: Product2[K, _]]], part, serializer)
}
}
}
4总结
由上述分析可以知道,如果需要join的两个表,本身已经有分区器,且分区的数目相同,此时,相同的key在同一个分区内。就是窄依赖。反之,如果两个需要join的表中没有分区器或者分区数量不同,在join的时候需要shuffle,那么就是宽依赖 ! 注意在创建RDD的时候没有指定分区器,在后面调用join的时候会调用默认的分区器,所以这个时候需要数据的shuffle ,是宽依赖!
Spark 的join 什么时候是宽依赖什么时候是窄依赖相关推荐
- Spark RDD 宽依赖窄依赖
Spark RDD 宽依赖&窄依赖 1.窄依赖 2.宽依赖: 3.阶段的划分 4.宽依赖和窄依赖的作用: 1.窄依赖 每一个父RDD的Partition最多被子RDD的一个Partition使 ...
- Spark RDD的窄依赖和宽依赖
目录 前言 一.窄依赖 二.宽依赖 三.使用WC演示窄依赖.宽依赖 四.窄依赖中的join 五.总结 前言 RDD的五大特性之一就是RDD之间有依赖关系,描述了RDD如何从源头计算过来的. 这样可以做 ...
- Spark _08窄依赖和宽依赖stage
专业术语: 窄依赖和宽依赖 RDD之间有一系列的依赖关系,依赖关系又分为窄依赖和宽依赖. 窄依赖 父RDD和子RDD partition之间的关系是一对一的.或者父RDD一个partition只对应一 ...
- spark系列11:RDD之间的依赖关系,窄依赖和宽依赖
1.RDD 之间的依赖关系 导读 讨论什么是 RDD 之间的依赖关系 继而讨论 RDD 分区之间的关系 最后确定 RDD 之间的依赖关系分类 完善案例的逻辑关系图 什么是 RDD 之间的依赖关系? 什 ...
- 宽依赖和窄依赖_Spark宽依赖和窄依赖深度剖析
宽依赖和窄依赖深度剖析.png RDD依赖关系与stage划分 Spark中RDD的高效与DAG图有着莫大的关系,在DAG调度中需要对计算过程划分stage,而划分依据就是RDD之间的依赖关系. 1. ...
- 宽依赖和窄依赖_Spark术语解释及宽窄依赖执行原理,代码分析
1. 术语解释 2. 窄依赖和宽依赖: RDD之间有一系列的依赖关系,依赖关系又分为窄依赖和宽依赖. 窄依赖: 父RDD和子RDD partition之间的关系是一对一的.或者父RDD一个partit ...
- 宽依赖和窄依赖深度剖析
宽依赖和窄依赖深度剖析 宽依赖和窄依赖的概念: 窄依赖:一个RDD,对其父RDD只有简单的一对一的依赖关系.即父RDD和子RDD的partition的对应关系是一对一的 宽依赖:英文名称shuffle ...
- spark中stage的划分与宽依赖/窄依赖(转载+自己理解/整理)
[1]宽依赖和窄依赖,这是Spark计算引擎划分Stage的根源所在,遇到宽依赖,则划分为多个stage,针对每个Stage,提交一个TaskSet: 上图:一张网上的图: (个人笔记,rdd中有多个 ...
- 深入解读 Spark 宽依赖和窄依赖(ShuffleDependency NarrowDependency)
目前,网上有关宽窄依赖的博客大多都使用下面这张图作为讲解: 实际上,这幅图所表达的内容并不完善.其中,窄依赖的内容表达的不够全面,而宽依赖的部分容易让人产生误解.本文,我将用新的绘图带大家搞清楚究竟什 ...
最新文章
- 尝试去读SQLMAP源码(一)
- python中文件操作中不同的文件操作有何差异_Python学习基础篇 -6: Python中的文件操作...
- ofdm原理_什么是IQ调制,了解I/Q调制的基本原理及其主要特征 - 射频技术第2部分...
- DataProtection设置问题引起不同ASP.NET Core站点无法共享用户验证Cookie
- Linq list 排序,Dictionary 排序
- type-c接口图片_TypeC接口除了充电还能干吗?这些功能都不知道,简直是在浪费...
- spring.net与OracleODP结合时发生的版本问题
- 计算机培训营,计算机学院举办科创训练营第三期培训
- HCI实验之问卷设计
- 兔子吃萝卜java游戏下载_巴迪兔子吃萝卜
- jsp技术被淘汰了?那还要不要学它?
- QQ群红包裂变吸粉引流 玩法解析实战操作
- 深圳可能是理解中国数字城市建设的窗口
- 推荐常用算法之-基于内容的推荐
- Ubuntu下将rm命令替换为trash命令
- outlook qr码在哪里_原平防伪溯源码_橙程(北京)科技有限公司
- 数据结构与算法|马踏棋盘算法(小甲鱼)C语言代码的算法分析
- Win10怎么安装旧版组件DirectPlay?
- w ndows7与windows10区别,w10系统和w7哪个好用2020
- 加载主类 Hello 时出现 LinkageError java.lang.UnsupportedClassVersionError
热门文章
- Win32中BeginPaint和GetDC的使用
- Oracle Primavera P6 20.12 安装要求
- 【初学不要怕】详解利用Python制作PPT的精彩操作(一学就会)
- go使用epoll或kqueue
- 基于opencv实现对图片中的物体计数
- nc 端口占用_使用NC-SI共享网口方式访问IPMI的BMC管理控制器
- 基础组合计数常用的概念和方法总结
- Thread starvation or clock leap detected
- 小程序全时全域自动化箱况检测+信息识别,中集飞瞳全球领先集装箱管理方案,AI自动化箱信息识别+箱况检测+地点报备,智慧港航智能化
- 运算符的使用数据读取语法