代码之家 › 专栏 › 技术社区 › daydayup

Java Spark广播并连接两个RDD

broadcast rdd join apache-spark

daydayup · 技术社区 · 6 年前

JavaPairRDD<String, MySchema> RDD1 ,和一个较小的 JavaPairRDD<String, Double> RDD2 . 我想加入这两个RDD,我知道最好的方法是 RDD2

// I ignored the parsing part, just simplified it as loading from the files. 
JavaPairRDD<String, MySchema> RDD1 = sc.textFile ("path_to_small_dataset");
JavaPairRDD<String, Double> RDD2 = sc.textFile("path_to_large_dataset"); 

// Broadcast RDD2
Set<Tuple2<String, Double>> set2 = new HashSet<>();
set2.addAll(RDD2.collect());

// now I have set2 and RDD1, how can I join them?

1 回复 | 直到 6 年前

mahmoud mehdi 6 年前

假设您有两个要加入的RDD,第一个RDD足够小,可以放入每个工作者的内存中( ),而第二个根本不需要洗牌( ).

在加入之前,您必须确保将大型RDD[T]转换为RDD[(key,T)]。键表示联接操作期间使用的列。

这段代码应该在Scala中做到这一点(但基本原理在Java中是相同的)

val smallLookup = sc.broadcast(smallRDD.collect.toMap)
largeRDD.flatMap { case(key, value) =>
  smallLookup.value.get(key).map { otherValue =>
  (key, (value, otherValue))
 }
}

我希望有帮助

推荐文章

alborzdev · 如何在PySpark中加入两个RDD?

2 年前

Vlad · 从未解析的文本字符串在spark中创建数据框

6 年前

Vinod · 跨职业和电影类型的电影收视率百分比分析

6 年前

juamd · 顺序(k,<元组>)RDD

6 年前

ka_boom · 在pyspark中链接多个groupBy

6 年前

udit · 使用scala比较两个不同的RDD

6 年前

Jerry George · 优化Pyspark代码以快速运行

6 年前

Bryce Ramgovind · PySpark-将映射函数添加为列

6 年前

Jerry George · 将和RDD转换为Spark数据帧(Pyspark)。这起作用了。但出现了新的错误

6 年前

Jerry George · 在PySpark中将StringType转换为ArrayType

6 年前