Spark Scala 中 R 的 cbind 功能答案

【问题标题】：R's cbind functionality in Spark ScalaSpark Scala 中 R 的 cbind 功能
【发布时间】：2017-12-11 09:30:07
【问题描述】：

我需要使用 scala 在 spark 中 cbind [就像在 R 中发生的那样] 两个数据帧，它们没有 ID 列。关于它的任何现成函数的任何指针，或者它的一些其他解决方法？

例子：

DF1：

    Name Age
    ABC  10
    BCD  11

DF2：

    Marks
    75
    85

需要结果：

    DF3:
    Name Age Marks
    ABC  10  75
    BCD  11  85

【问题讨论】：

要绑定列，您不需要任何关于列名的先验知识。唯一需要匹配的是行数。您可能正在寻找 merge 或 SQL 等价物“连接”。
是的，但我没有任何要加入的关键列。我实际上是在寻找 Spark 等效于 R 的 cbind()。
你见过这个吗？ forums.databricks.com/questions/8180/…
这是一种解决方法。我希望像这样基本的东西可能已经用 spark 的核心功能实现了！
注释并不适合粘贴大量代码。编辑您的问题，或者如果您找到了解决方案，请将其发布为答案。

标签： r scala apache-spark dataframe cbind

【解决方案1】：

这是一种完美的解决方法：

    df1 = df1.withColumn("id", monotonically_increasing_id())
    df2 = df2.withColumn("id", monotonically_increasing_id())
    df3 = df2.join(df1, "id", "outer").drop("id")

对于 spark 1.6.*，最后一行需要是：

    df3 = df2.join(df1, df1("id") === df2("id"), "outer").drop("id")

【讨论】：

由于 monotonically_increasing_id() 不能保证连续的序列，将 cbind 加倍可能会带来麻烦。
第三行的正确形式（Spark 1.4）是：df3 = df2.join(df1, Seq("id"), "outer").drop("id")跨度>

【解决方案2】：

试试这个功能

val cbind: (DataFrame, DataFrame) => DataFrame =
  (df, df2) => {
    val x =
      spark
        .createDataFrame(
          df.rdd.zipWithUniqueId.map(tu => Row(tu._1.toSeq.:+(tu._2): _*)),
          df.schema.add(StructField("primaryKeyForCbind", LongType, false)))
        .withColumn("orderKeyForCbind", $"primaryKeyForCbind")
        .as("df")
    val y =
      spark
        .createDataFrame(
          df2.rdd.zipWithUniqueId.map(tu => Row(tu._1.toSeq.:+(tu._2): _*)),
          df2.schema.add(StructField("primaryKeyForCbind", LongType, false)))
        .as("df2")
    x.join(y, col("df.primaryKeyForCbind") === col("df2.primaryKeyForCbind"))
      .sort("orderKeyForCbind")
      .drop("primaryKeyForCbind", "orderKeyForCbind")
  }

val df =
  Seq(
    ("curycu", "2018-01-01"),
    ("curycu", "2018-01-07"),
    ("curycu", "2018-01-12"),
    ("curycu", "2018-01-15"),
    ("tester", "2018-01-01"),
    ("tester", "2018-01-11"),
    ("tester", "2018-01-18"))
    .toDF("id", "date")

val df2 = Seq(0, 6, 5, 3, 0, 10, 7).toDF("daygap")

val res = cbind(cbind(cbind(df, df2), df2), df)
res.printSchema
res.show

root
 |-- id: string (nullable = true)
 |-- date: string (nullable = true)
 |-- daygap: integer (nullable = false)
 |-- daygap: integer (nullable = false)
 |-- id: string (nullable = true)
 |-- date: string (nullable = true)

+------+----------+------+------+------+----------+
|    id|      date|daygap|daygap|    id|      date|
+------+----------+------+------+------+----------+
|curycu|2018-01-01|     0|     0|curycu|2018-01-01|
|curycu|2018-01-07|     6|     6|curycu|2018-01-07|
|curycu|2018-01-12|     5|     5|curycu|2018-01-12|
|curycu|2018-01-15|     3|     3|curycu|2018-01-15|
|tester|2018-01-01|     0|     0|tester|2018-01-01|
|tester|2018-01-11|    10|    10|tester|2018-01-11|
|tester|2018-01-18|     7|     7|tester|2018-01-18|
+------+----------+------+------+------+----------+

【讨论】：