在 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

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

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 工作原理的绝佳参考。

尝试过代码但没有给出任何输出

为了看到输出，你必须让 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)，这是故意的吗？

希望这会有所帮助！