用另一个 RDD 替换一个 RDD 的值答案

【问题标题】：Replacing the values of an RDD with another用另一个 RDD 替换一个 RDD 的值
【发布时间】：2016-08-23 17:24:24
【问题描述】：

我有两个数据集，如下所示。每个数据集的每一行都有“,”分隔的数字。

数据集 1

1,2,0,8,0

2,0,9,0,3

数据集 2

7,5,4,6,3

4,9,2,1,8

我必须用数据集 2 中的相应值替换第一个数据集的零。

所以结果应该是这样的

1,2,4,8,3

2,9,9,1,3

我用下面的代码替换了这些值。

val rdd1 = sc.textFile(dataset1).flatMap(l => l.split(","))
val rdd2 = sc.textFile(dataset2).flatMap(l => l.split(","))
val result = rdd1.zip(rdd2).map( x => if(x._1 == "0") x._2 else x._1)

我得到的输出格式为 RDD[String]。但我需要 RDD[Array[String]] 格式的输出，因为这种格式更适合我的进一步转换。

【问题讨论】：

你在寻找类似val result = rdd1.zip(rdd2).map(x => if(x._1 == "0") Array(x._2) else Array(x._1))的东西吗？
@AlexisC。不，rdd1 和 rdd2 的类型是 RDD[Array[String]]。所以代码中的 x._1 指的是数组
嗯，从您的 sn-p 中不清楚。拆分后您是平面映射，导致RDD[String] 和x._1 == "0"；那么x._1 是如何引用数组的呢？除非你想要一个带有 2 个数组的 RDD[Array[String]]（每行一个）？
抱歉不是很清楚。我想要一个带有 2 个数组（每行一个）的 RDD[Array[String]] 类型的结果 RDD。但是，当我使用 flatMap 时，我编写的代码以 RDD[String] 格式给出了结果。

标签： scala apache-spark rdd

【解决方案1】：

如果你想要一个RDD[Array[String]]，其中数组的每个元素对应一行，拆分后不要平面映射值，只需映射它们即可。

scala> val rdd1 = sc.parallelize(List("1,2,0,8,0", "2,0,9,0,3")).map(l => l.split(","))
rdd1: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[1] at map at <console>:27

scala> val rdd2 = sc.parallelize(List("7,5,4,6,3", "4,9,2,1,8")).map(l => l.split(","))
rdd2: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[3] at map at <console>:27

scala> val result = rdd1.zip(rdd2).map{case(arr1, arr2) => arr1.zip(arr2).map{case(v1, v2) => if(v1 == "0") v2 else v1}}
result: org.apache.spark.rdd.RDD[Array[String]] = MapPartitionsRDD[5] at map at <console>:31

scala> result.collect
res0: Array[Array[String]] = Array(Array(1, 2, 4, 8, 3), Array(2, 9, 9, 1, 3))

或者可能不那么冗长：

val result = rdd1.zip(rdd2).map(t => t._1.zip(t._2).map(x => if(x._1 == "0") x._2 else x._1))

【讨论】：

我有另一个 RDD，它具有上述结果的阈值，格式为 Array(Array(6, 100), Array(5, 100), Array(7, 100), Array(0 , 100), 数组 (-1, 100))。如何检查结果RDD的每个Array中的值是否在这些阈值之间？