scala - 如何将 RDD[String] 与 Array[String] 连接以生成字符串？

Question

如何将RDD[String]和转换Array[String]为字符串？

我收到以下错误，

<console>:34: error: type mismatch;
found   : org.apache.spark.rdd.RDD[String]
required: String

这个想法是从 SchemaRDD 中的列中获取不同的日期，并将日期与常量字符串连接为/home/tmp/date=. 所以我应该连接两者，输出应该是

val path =  /home/tmp/date=20140901,/home/tmp/date=20140902,/home/tmp/date=20140903,/home/tmp/date=20140904,... so on

将键入路径sc.textFiles(path)以读取整个数据集。

在这一步，读取数据时出现转换错误。

score 4 · Accepted Answer

野兔是一种方法。首先，设置示例：

val prefix = "/home/tmp/date="
val dates =  Array("20140901", "20140902", "20140903", "20140904")
val datesRDD = sc.parallelize(dates, 2)

压缩前缀很容易：

val datesWithPrefixRDD = datesRDD.map(s => prefix + s)
datesWithPrefixRDD.foreach(println)

这会产生：

/home/tmp/date=20140901
/home/tmp/date=20140903
/home/tmp/date=20140902
/home/tmp/date=20140904

但是你要求一个字符串。显而易见的第一次尝试有一些逗号问题：

val bad = datesWithPrefixRDD.fold("")((s1, s2) => s1 + ", " + s2)
println(bad)

这会产生：

, , /home/tmp/date=20140901, /home/tmp/date=20140902, , /home/tmp/date=20140903, /home/tmp/date=20140904

问题是 Spark RDD 的 fold() 方法以我提供的空字符串开始连接的方式，一次用于整个 RDD，一次用于每个分区。但是我们可以处理空字符串：

val good = datesWithPrefixRDD.fold("")((s1, s2) =>
  s1 match {
    case "" => s2
    case s => s + ", " + s2
  })
println(good)

然后我们得到：

/home/tmp/date=20140901, /home/tmp/date=20140902, /home/tmp/date=20140903, /home/tmp/date=20140904

编辑：实际上， reduce() 产生了一个更整洁的答案，因为它解决了“额外的逗号”问题：

val alternative = datesWithPrefixRDD.reduce((s1, s2) => s1 + ", " + s2)
println(alternative)

我们再次得到：

/home/tmp/date=20140901, /home/tmp/date=20140902, /home/tmp/date=20140903, /home/tmp/date=20140904

1 回答 1