Scala - 用另一个 map 更新 RDD

Question

我正在尝试使用来自另一个 Map 的更多信息来更新 RDD....我写了这个但是没有用。

地点:

LocalCurrencies 是一个 Sequence of Currency 类

rdd: RDD[字符串, 字符串]

...
val localCurrencies = Await.result(CurrencyDAO.currencies, 30 seconds)

//update ISO3
rdd.map(r => r.updated("currencyiso3", localCurrencies.find(c => c.CurrencyId ==   
rdd.get("currencyid")).get.ISO3))

//Update exponent
rdd.map(r => r.updated("exponent", localCurrencies.find(c => c.CurrencyId == 
rdd.get("currencyid")).get.Exponent))

有什么建议吗？

谢谢

Answer 1

map 不会修改 RDD，它会创建一个新的 RDD(这同样适用于每个 Spark 转换)。如果您实际上不对这个新的 RDD 做任何事情，Spark 甚至不会费心去创建它。所以你想写

val rdd1 = rdd.map(...).map(...) // better to combine two `map`s into one

并从第一个开始使用 rdd1(如果需要，您仍然可以使用 rdd)。这不一定是唯一的错误，但您仍然需要修复它。