java - Spark 斯卡拉 : Convert DataFrame OR Dataset to single comma separated string

Question

下面是将打印一列 DataSet[Row] 的 spark scala 代码:

import org.apache.spark.sql.{Dataset, Row, SparkSession}
val spark: SparkSession = SparkSession.builder()
        .appName("Spark DataValidation")
        .config("SPARK_MAJOR_VERSION", "2").enableHiveSupport()
        .getOrCreate()

val kafkaPath:String="hdfs:///landing/APPLICATION/*"
val targetPath:String="hdfs://datacompare/3"
val pk:String = "APPLICATION_ID" 
val pkValues = spark
        .read
        .json(kafkaPath)
        .select("message.data.*")
        .select(pk)
        .distinct() 
pkValues.show()

关于代码的输出:

+--------------+
|APPLICATION_ID|
+--------------+
|           388|
|           447|
|           346|
|           861|
|           361|
|           557|
|           482|
|           518|
|           432|
|           422|
|           533|
|           733|
|           472|
|           457|
|           387|
|           394|
|           786|
|           458|
+--------------+

问题:

如何将此数据框转换为逗号分隔的字符串变量？

预期输出:

val   data:String= "388,447,346,861,361,557,482,518,432,422,533,733,472,457,387,394,786,458"

请建议如何将 DataFrame[Row] 或 Dataset 转换为一个 String 。

Answer 1

我认为这不是个好主意，因为 dataFrame 是分布式对象并且可能非常庞大。 Collect会将所有的数据带到驱动程序中，所以你应该小心执行这种操作。

以下是您可以使用 dataFrame 执行的操作(两个选项):

df.select("APPLICATION_ID").rdd.map(r => r(0)).collect.mkString(",")
df.select("APPLICATION_ID").collect.mkString(",")

只有 3 行的测试数据帧的结果:

String = 388,447,346

编辑:使用 DataSet，您可以直接执行以下操作:

ds.collect.mkString(",")