将 Dataframe 的列值提取为 Apache Spark 中的列表

Extract column values of Dataframe as List in Apache Spark

我想将数据框的字符串列转换为列表。我从Dataframe API中可以找到的是RDD,所以我尝试先将其转换回RDD,然后将toArray函数应用于RDD。在这种情况下,长度和 SQL 工作得很好。然而,我从 RDD 得到的结果在每个元素周围都有方括号,就像这样 [A00001]。我想知道是否有适当的方法将列转换为列表或删除方括号的方法。

如有任何建议,我们将不胜感激。谢谢!

这应该 return 包含单个列表的集合:

dataFrame.select("YOUR_COLUMN_NAME").rdd.map(r => r(0)).collect()

如果没有映射,您只会得到一个 Row 对象,其中包含数据库中的每一列。

请记住,这可能会为您提供 Any 类型的列表。如果你想指定结果类型,你可以使用 .asInstanceOf[YOUR_TYPE] in r => r(0).asInstanceOf[YOUR_TYPE] mapping

P.S。由于自动转换,您可以跳过 .rdd 部分。

我知道给出的答案和要求的答案都是针对 Scala 的,所以我只提供一小段 Python 代码,以防 PySpark 用户好奇。语法类似于给定的答案,但要正确弹出列表,我实际上必须在映射函数中再次引用列名,我不需要 select 语句。

即一个 DataFrame,包含一个名为 "Raw"

的列

要将 "Raw" 中的每一行值合并为一个列表,其中每个条目都是来自 "Raw" 的行值,我只需使用:

MyDataFrame.rdd.map(lambda x: x.Raw).collect()

使用 Spark 2.x 和 Scala 2.11

我想出了 3 种可能的方法来将特定列的值转换为列表。

所有方法的通用代码片段

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder.getOrCreate    
import spark.implicits._ // for .toDF() method

val df = Seq(
    ("first", 2.0),
    ("test", 1.5), 
    ("choose", 8.0)
  ).toDF("id", "val")

方法一

df.select("id").collect().map(_(0)).toList
// res9: List[Any] = List(one, two, three)

现在发生了什么?我们正在使用 collect() 向 Driver 收集数据,并从每条记录中选取元素零。

这不是一个很好的方法,让我们用下一个方法改进它。


方法二

df.select("id").rdd.map(r => r(0)).collect.toList 
//res10: List[Any] = List(one, two, three)

怎么更好?我们在工作人员之间分配了地图转换负载,而不是单个驱动程序。

我知道rdd.map(r => r(0))看来你不优雅。所以,让我们在下一个方法中解决它。


方法 3

df.select("id").map(r => r.getString(0)).collect.toList 
//res11: List[String] = List(one, two, three)

这里我们不是将DataFrame转为RDD。查看 map 由于 DataFrame 中的编码器问题,它不会接受 r => r(0)(或 _(0))作为以前的方法。所以最终使用 r => r.getString(0),它将在下一版本的 Spark 中解决。

Conclusion

所有选项都提供相同的输出,但 2 和 3 有效,最后第 3 个有效且优雅(我认为)。

Databricks notebook

在 Scala 和 Spark 2+ 中,试试这个(假设您的列名称是“s”):

df.select('s').as[String].collect
sqlContext.sql(" select filename from tempTable").rdd.map(r => r(0)).collect.toList.foreach(out_streamfn.println) //remove brackets

完美运行

from pyspark.sql.functions import col

df.select(col("column_name")).collect()

这里的 collect 是将其转换为列表的函数。 在庞大的数据集上使用列表时要小心。它会降低性能。 查资料就好了

这是java答案。

df.select("id").collectAsList();
List<String> whatever_list = df.toJavaRDD().map(new Function<Row, String>() {
    public String call(Row row) {
        return row.getAs("column_name").toString();
    }
}).collect();

logger.info(String.format("list is %s",whatever_list)); //verification

因为没有人在java(真正的编程语言)中给出任何解决方案 稍后可以感谢我

为您提供列表的更新解决方案:

dataFrame.select("YOUR_COLUMN_NAME").map(r => r.getString(0)).collect.toList

下面是 Python-

df.select("col_name").rdd.flatMap(lambda x: x).collect()