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

Question

我有一个火花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=

<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

尝试使用DF和函数withColumn（“new_column”，count（$“list”））或其任何变体添加新列。它仍然不起作用。我收到一条关于聚合的错误消息。

你知道一种不用收集RDD就能实现这一点的方法吗？




Answer 1

您可以使用简单的自定义项创建新列，以应用于列列表，如下所示：

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的简单用例。

Answer 2

内置函数size返回数组或映射的长度。

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