使用 foreach 行在数据帧内捕获和写入字符串

Question

在使用 Scala.但由于它部署在集群上，无法捕获任何记录。任何人都可以提供解决方案吗？

假设 TEST_DB.finalresult 有 2 个字段 input1 和 input2：

val finalresult=spark.sql("select * from TEST_DB.finalresult")

finalResult.foreach { row => 
    val param1=row.getAs("input1").asInstanceOf[String]
    val param2=row.getAs("input2").asInstanceOf[String]

    val string = """new values of param1 and param2 are -> """ + param1 + """,""" + param2
    // how to append modified string to csv file continously for each microbatch in hdfs ??
}

Answer 1

在您的代码中，您创建了想要的 string 变量，但它没有保存在任何地方，因此您看不到结果。

您可以在每个 foreach 执行中打开所需的 csv 文件并附加新字符串，但我想提出一个不同的解决方案。

如果可以，请尝试始终使用 Spark 的内置功能，因为它（通常）在处理空输入方面更加优化和更好。您可以通过以下方式达到同样的目的：

import org.apache.spark.sql.functions.{lit, concat, col}

val modifiedFinalResult = finalResult.select(
 concat(
  lit("new values of param1 and param2 are -> "),
  col("input1"),
  lit(","),
  col("input2")
 ).alias("string")
)

在变量 modifiedFinalResult 中，您将拥有一个带有名为 string 的单列的 spark 数据框，它表示与代码中的变量 string 完全相同的输出。之后，您可以将数据框直接保存为单个 csv 文件（使用重新分区功能）：

modifiedFinalResult.repartition(1).write.format("csv").save("path/to/your/csv/output")

PS：也是对未来的一个建议，尽量避免以数据类型命名变量。

更新：通过使用“concat_ws”而不是 concat 和合并到每个字段来修复空行问题。似乎一些为空的值在转换后将整个连接字符串转换为空。尽管如此，这个解决方案现在仍然有效！