在 Apache Spark 1.3 中将列附加到数据框答案

【问题标题】：Append a column to Data Frame in Apache Spark 1.3在 Apache Spark 1.3 中将列附加到数据框
【发布时间】：2015-06-11 13:44:08
【问题描述】：

向数据框中添加一列是否可能以及最有效的简洁方法是什么？

更具体地说，列可以用作现有数据框的行 ID。

在一个简化的情况下，从文件读取而不是对其进行标记，我可以想到以下内容（在 Scala 中），但它完成时出现错误（在第 3 行），而且无论如何看起来都不是最好的路线：

var dataDF = sc.textFile("path/file").toDF() 
val rowDF = sc.parallelize(1 to DataDF.count().toInt).toDF("ID") 
dataDF = dataDF.withColumn("ID", rowDF("ID"))

【问题讨论】：

您找到解决方案了吗？
有什么错误？这似乎是 API 中的正确方法。
@Chet, withColumn 应该使用相同的数据框 - 即您可以使用 dataDF = dataDF.withColumn("ID", dataDF("ID").map(...)) 仅使用 this 数据框的列，而不是其他人
@OlegShirokikh 啊。嗯。如果该限制被记录在案，那就太好了。那么有趣的问题。我可以从 API 中看到的唯一机制是使用join。建议增强 API 的人选不错。
@Chet - 对，它是任何数据框的核心功能之一。例如，它们具有一些内置功能来更新 Parquet 文件的模式。显然，在分布式环境中这是非常昂贵的操作，但无论如何 IMO 它应该在那里

标签： scala apache-spark dataframe

【解决方案1】：

我发布问题已经有一段时间了，似乎其他人也想得到答案。以下是我找到的。

因此，最初的任务是将带有行标识符的列（基本上是序列1 to numRows）附加到任何给定的数据帧，以便可以跟踪行的顺序/存在（例如，当您采样时）。这可以通过以下方式实现：

sqlContext.textFile(file).
zipWithIndex().
map(case(d, i)=>i.toString + delimiter + d).
map(_.split(delimiter)).
map(s=>Row.fromSeq(s.toSeq))

关于将任何列附加到任何数据框的一般情况：

在 Spark API 中“最接近”此功能的是 withColumn 和 withColumnRenamed。根据Scala docs，前者通过添加列返回一个新的DataFrame。在我看来，这有点令人困惑和不完整的定义。这两个函数都只能对this数据帧进行操作，即给定两个数据帧df1和df2，列col：

val df = df1.withColumn("newCol", df1("col") + 1) // -- OK
val df = df1.withColumn("newCol", df2("col") + 1) // -- FAIL

因此，除非您能够设法将现有数据框中的列转换为您需要的形状，否则您不能使用 withColumn 或 withColumnRenamed 附加任意列（独立或其他数据框）。

正如上面评论的那样，解决方法可能是使用join - 尽管可能，但这会非常混乱 - 将上面带有zipWithIndex 的唯一键附加到数据框或列可能会起作用。虽然效率是...

很明显，在数据框中添加一列对于分布式环境来说并不是一个简单的功能，而且可能根本就没有非常有效、简洁的方法。但我认为，即使有性能警告，让这个核心功能可用仍然非常重要。

【讨论】：

最近有一些支持 row_number 函数的倡议（以及其他） - SPARK-7712 JIRA 票证没有明确提及，因此您可能需要查看 related pull-request
好答案！如果 Spark 知道我正在加入 sorted 键，则可以优化附加列功能。这将大大提升性能。
如果你想添加一个 id 列，你应该看看可以在 withColumn 内部使用的函数 monotonically_increasing_id()。
monotonically_increasing_id() 存在严重的脆弱性问题，使用时必须非常小心stackoverflow.com/questions/35705038/…

【解决方案2】：

我从上面的答案中得到了帮助。但是，如果我们想更改 DataFrame 并且当前的 API 在 Spark 1.6 中几乎没有什么不同，我发现它是不完整的。 zipWithIndex() 返回 (Row, Long) 的 Tuple，其中包含每一行和相应的索引。我们可以根据需要使用它来创建新的Row。

val rdd = df.rdd.zipWithIndex()
             .map(indexedRow => Row.fromSeq(indexedRow._2.toString +: indexedRow._1.toSeq))
val newstructure = StructType(Seq(StructField("Row number", StringType, true)).++(df.schema.fields))
sqlContext.createDataFrame(rdd, newstructure ).show

我希望这会有所帮助。

【讨论】：

【解决方案3】：

不确定它是否在 spark 1.3 中有效，但在 spark 1.5 中我使用 withColumn：

import sqlContext.implicits._
import org.apache.spark.sql.functions._


df.withColumn("newName",lit("newValue"))

当我需要使用与数据框的现有列无关的值时，我会使用它

这类似于@NehaM 的答案，但更简单

【讨论】：

【解决方案4】：

您可以使用row_number 和Window function 如下所示来获取数据帧中每一行的不同ID。

df.withColumn("ID", row_number() over Window.orderBy("any column name in the dataframe"))

你也可以使用monotonically_increasing_id

df.withColumn("ID", monotonically_increasing_id())

还有一些other ways。

【讨论】：