将 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"))

将自动推断列的类型。