union
官方文档描述:
Return the union of this RDD and another one.
Any identical elements will appear multiple times (use `.distinct()` to eliminate them).
函数原型:
def union(other: JavaRDD[T]): JavaRDD[T]
**
union() 将两个 RDD 简单合并在一起,不改变 partition 里面的数据。RangeDependency 实际上也是 1:1,只是为了访问 union() 后的 RDD 中的 partition 方便,保留了原始 RDD 的 range 边界。
**
实例:
List<Integer> data = Arrays.asList(1,2,4,3,5,6,7);
JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);
JavaRDD<Integer> unionRDD = javaRDD.union(javaRDD);
System.out.println("unionRDD~~~~~~~~~~~~~~~~~~~~~~" + unionRDD.collect());
intersection
官方文档描述:
Return the intersection of this RDD and another one.
The output will not contain any duplicate elements, even if the input RDDs did.
Note that this method performs a shuffle internally.
函数原型:
def intersection(other: JavaRDD[T]): JavaRDD[T]
源码分析:
def intersection(other: RDD[T]): RDD[T] = withScope {
this.map(v => (v, null)).cogroup(other.map(v => (v, null)))
.filter { case (_, (leftGroup, rightGroup)) => leftGroup.nonEmpty && rightGroup.nonEmpty }
.keys
}
**
先使用 map() 将 RDD[T] 转变成 RDD[(T, null)],这里的 T 只要不是 Array 等集合类型即可。接着,进行 a.cogroup(b)(后面会详细介绍cogroup)。之后再使用 filter() 过滤掉 [iter(groupA()), iter(groupB())] 中 groupA 或 groupB 为空的 records,得到 FilteredRDD。最后,使用 keys() 只保留 key 即可,得到 MappedRDD。
**
实例:
List<Integer> data = Arrays.asList(1, 2, 4, 3, 5, 6, 7);
JavaRDD<Integer> javaRDD = javaSparkContext.parallelize(data);
JavaRDD<Integer> intersectionRDD = javaRDD.intersection(javaRDD);
System.out.println(intersectionRDD.collect());
网友评论