scala - 在 Spark 中加入两个 HDFS 文件

Question

我想使用 spark shell 连接来自 HDFS 的两个文件。这两个文件都是制表符分隔的，我想加入第二列

尝试过的代码但不给出任何输出

val ny_daily= sc.parallelize(List("hdfs://localhost:8020/user/user/NYstock  /NYSE_daily"))

val ny_daily_split = ny_daily.map(line =>line.split('\t'))

val enKeyValuePair = ny_daily_split.map(line => (line(0).substring(0, 5), line(3).toInt))


val ny_dividend= sc.parallelize(List("hdfs://localhost:8020/user/user/NYstock/NYSE_dividends"))

val ny_dividend_split = ny_dividend.map(line =>line.split('\t'))

val enKeyValuePair1 = ny_dividend_split.map(line => (line(0).substring(0, 4),     line(3).toInt))

enKeyValuePair1.join(enKeyValuePair)

但是我没有得到任何关于如何在特定列上加入文件的信息请建议

Answer 1

I am not getting any information for how to join files on particular column

RDD 在它们的键上连接，因此您在编写时决定要连接的列:

val enKeyValuePair = ny_daily_split.map(line => (line(0).substring(0, 5), line(3).toInt))
...
val enKeyValuePair1 = ny_daily_split.map(line => (line(0).substring(0, 4), line(3).toInt))

您的 RDD 将连接来自 line(0).substring(0, 5) 和 line(0).substring(0, 4) 的值.

您可以找到join 函数(以及许多其他有用的函数)here和 Spark Programming Guide是了解 Spark 工作原理的重要引用。

Tried code But not giving any output

为了看到输出，你必须让 Spark 打印它:

enKeyValuePair1.join(enKeyValuePair).foreach(println)

注意:要从文件中加载数据，您应该使用 sc.textFile():sc.parallelize() 仅用于使来自 Scala 集合的 RDD。

下面的代码应该可以完成这项工作:

val ny_daily_split = sc.textFile("hdfs://localhost:8020/user/user/NYstock/NYSE_daily").map(line =>line.split('\t'))
val ny_dividend_split = sc.textFile("hdfs://localhost:8020/user/user/NYstock/NYSE_dividends").map(line =>line.split('\t'))

val enKeyValuePair = ny_daily_split.map(line => line(0).substring(0, 5) -> line(3).toInt)
val enKeyValuePair1 = ny_dividend_split.map(line => line(0).substring(0, 4) -> line(3).toInt)

enKeyValuePair1.join(enKeyValuePair).foreach(println)

顺便说一句，你提到你想加入第二列，但你实际上正在使用 line(0)，这是故意的吗？

希望这对您有所帮助!