scala - 将 StringType 列添加到现有 Spark DataFrame，然后应用默认值

Question

Scala 2.10 在这里使用 Spark 1.6.2。我有一个与 this one 类似(但不相同)的问题，然而，接受的答案不是 SSCCE并假设对 Spark 有一定的“前期知识”；因此我无法复制它或理解它。 更重要的是 ，这个问题也仅限于向现有数据帧添加新列，而我需要为数据帧中的所有现有行添加一列和值。

因此，我想向现有 Spark DataFrame 添加一列，然后将该新列的初始(“默认”)值应用于所有行。

val json : String = """{ "x": true, "y": "not true" }"""
val rdd = sparkContext.parallelize(Seq(json))
val jsonDF = sqlContext.read.json(rdd)

jsonDF.show()

当我运行时，我得到以下输出(通过 .show() ):

+----+--------+
|   x|       y|
+----+--------+
|true|not true|
+----+--------+

现在我想向 jsonDF 添加一个新字段, 在它创建之后并且不修改 json字符串，这样生成的 DF 将如下所示:

+----+--------+----+
|   x|       y|   z|
+----+--------+----+
|true|not true| red|
+----+--------+----+

意思是，我想向 DF 添加一个新的“ z”列，类型为 StringType ，然后默认所有行都包含 z -值 "red" .

从另一个问题我拼凑了以下伪代码:

val json : String = """{ "x": true, "y": "not true" }"""
val rdd = sparkContext.parallelize(Seq(json))
val jsonDF = sqlContext.read.json(rdd)

//jsonDF.show()

val newDF = jsonDF.withColumn("z", jsonDF("col") + 1)

newDF.show()

但是当我运行它时，我在 .withColumn(...) 上得到一个编译器错误方法:

org.apache.spark.sql.AnalysisException: Cannot resolve column name "col" among (x, y);
    at org.apache.spark.sql.DataFrame$$anonfun$resolve$1.apply(DataFrame.scala:152)
    at org.apache.spark.sql.DataFrame$$anonfun$resolve$1.apply(DataFrame.scala:152)
    at scala.Option.getOrElse(Option.scala:120)
    at org.apache.spark.sql.DataFrame.resolve(DataFrame.scala:151)
    at org.apache.spark.sql.DataFrame.col(DataFrame.scala:664)
    at org.apache.spark.sql.DataFrame.apply(DataFrame.scala:652)

我也没有看到任何允许我设置的 API 方法 "red"作为默认值。关于我要去哪里的任何想法？

Answer 1

您可以使用 lit功能。首先你必须导入它

import org.apache.spark.sql.functions.lit

并按如下所示使用它

jsonDF.withColumn("z", lit("red"))

将自动推断列的类型。