Spark Dataframe/RDD 无法通过计算另一列的内容来创建新列答案

【问题标题】：Spark Dataframe/RDD cannot create new column by counting the contents of another columnSpark Dataframe/RDD 无法通过计算另一列的内容来创建新列
【发布时间】：2017-12-11 02:34:44
【问题描述】：

我有一个包含以下列（每个结构的示例）的 Spark RDD（或 Dataframe - 转换为任何一个都不是问题）：

res248: org.apache.spark.rdd.RDD[(String, Array[String])] = MapPartitionsRDD[1004] at map at <console>:246
org.apache.spark.sql.DataFrame = [id: string, list: array<string>]

我想扩展这个 RDD/DF 以增加一个包含列表数组大小的列。所以输出应该是这样的（例子）：

org.apache.spark.sql.DataFrame = [id: string, list: array<string>, length_of_list: int]

我尝试发送rdd.map(x=> (x._1,x._2,count(x._2)))，但收到一条错误消息：

<console>:246: error: overloaded method value count with alternatives:
  (columnName: String)org.apache.spark.sql.TypedColumn[Any,Long] <and>
  (e: org.apache.spark.sql.Column)org.apache.spark.sql.Column

尝试使用带有函数withColumn("new_column",count($"list")) 或其任何变体的DF 添加新列。它仍然不起作用。我收到一条关于聚合的错误消息。

您知道无需收集 RDD 即可实现此目的的方法吗？

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

您可以使用简单的UDF 创建新列以应用于列list，如下所示：

val df = Seq(
  ("a", Array("x1", "x2", "x3")),
  ("b", Array("y1", "y2", "y3", "y4"))
).toDF(
  "id", "list"
)
// df: org.apache.spark.sql.DataFrame = [id: string, list: array<string>]

val listSize = (l: Seq[String]) => l.size
// listSize: Seq[String] => Int = <function1>

def listSizeUDF = udf(listSize)
// listSizeUDF: org.apache.spark.sql.expressions.UserDefinedFunction

val df2 = df.withColumn("length_of_list", listSizeUDF($"list"))

df2.show
+---+----------------+--------------+
| id|            list|length_of_list|
+---+----------------+--------------+
|  a|    [x1, x2, x3]|             3|
|  b|[y1, y2, y3, y4]|             4|
+---+----------------+--------------+

[更新]

正如@Ramesh Maharjan 所指出的，Spark 中有一个内置的size 函数，我不知何故忘记了。我将保留旧答案作为使用 UDF 的简单用例。

【讨论】：

【解决方案2】：

有一个内置函数size，它返回数组或映射的长度。

import org.apache.spark.sql.functions._
df.withColumn("length_of_list", size($"list"))

【讨论】：