【问题标题】:Merge two tables in Scala/Spark在 Scala/Spark 中合并两个表
【发布时间】:2017-08-03 04:17:29
【问题描述】:

我有两个制表符分隔的数据文件,如下所示:

文件 1:

number  type    data_present
 1       a        yes
 2       b        no

文件 2:

type    group   number  recorded
 d       aa      10       true
 c       cc      20       false

我想合并这两个文件,使输出文件如下所示:

number  type    data_present    group   recorded
  1      a         yes           NULL    NULL
  2      b         no            NULL    NULL
  10     d         NULL           aa     true
  20     cc        NULL           cc     false

如您所见,对于其他文件中不存在的列,我将用 NULL 填充这些位置。

关于如何在 Scala/Spark 中执行此操作的任何想法?

【问题讨论】:

  • 这是一种交叉连接或联合,但准备数据是乏味的部分。你走了多远?你能发布一些火花代码吗?

标签: scala apache-spark


【解决方案1】:

为您的数据集创建两个文件:

$ cat file1.csv 
number  type    data_present
 1       a        yes
 2       b        no

$ cat file2.csv
type    group   number  recorded
 d       aa      10       true
 c       cc      20       false

将它们转换为 CSV:

$ sed -e 's/^[ \t]*//' file1.csv | tr -s ' ' | tr ' ' ',' > f1.csv
$ sed -e 's/^[ ]*//' file2.csv | tr -s ' ' | tr ' ' ',' > f2.csv

使用spark-csv模块将CSV文件加载为数据框:

$ spark-shell --packages com.databricks:spark-csv_2.10:1.1.0

import org.apache.spark.sql.SQLContext
val sqlContext = new SQLContext(sc)
val df1 = sqlContext.load("com.databricks.spark.csv", Map("path" -> "f1.csv", "header" -> "true"))
val df2 = sqlContext.load("com.databricks.spark.csv", Map("path" -> "f2.csv", "header" -> "true"))

现在执行连接:

scala> df1.join(df2, df1("number") <=> df2("number") && df1("type") <=> df2("type"), "outer").show()

+------+----+------------+----+-----+------+--------+
|number|type|data_present|type|group|number|recorded|
+------+----+------------+----+-----+------+--------+
|     1|   a|         yes|null| null|  null|    null|
|     2|   b|          no|null| null|  null|    null|
|  null|null|        null|   d|   aa|    10|    true|
|  null|null|        null|   c|   cc|    20|   false|
+------+----+------------+----+-----+------+--------+

更多详情请转至hereherehere

【讨论】:

【解决方案2】:

这将为您提供所需的输出:

val output = file1.join(file2, Seq("number","type"), "outer")

【讨论】:

  • 面对类似的问题,试过这个但得到:发现:Seq[String] required: org.apache.spark.sql.Column stackoverflow.com/questions/38149483/…
  • @user3407267 我想你的问题已经被 zero323 回答了,你还面临这个问题吗??
【解决方案3】:

简单地将所有列转换为字符串,而不是在两个 DF 上进行联合。

【讨论】:

    猜你喜欢
    • 2015-10-18
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 2021-02-21
    • 2018-05-07
    • 2018-12-31
    • 2018-05-18
    • 1970-01-01
    相关资源
    最近更新 更多