Spark将输出写入foreach批处理中的一个文件

Posted 2023-03-23

【中文标题】Spark将输出写入foreach批处理中的一个文件

【英文标题】：Spark write the output to one file in foreach batch

【发布时间】：2021-10-07 16:19:07

【问题描述】：

是否有任何方法可以控制 spark 将输出写入一个文件，即使在 foreach 批处理中？

我有一个 Spark 作业，需要从 kafka 读取数据，然后将三个键（appName、moduleName、serviceName）保存到 S3，您可以将它们视为主键，所以我想在连续读取时将它们附加到一个文件中来自kafka的数据。

有什么想法吗？

Dataset<Row> rowDataset = dataset.selectExpr("CAST(value AS STRING)")
                .select(from_csv(col("value"), SpanSchema4Csv.spanStructType, new Map.Map1<>("escape", "\"")).as("data"))
                .select(col("data.TAGS_APPNAME").as("applicationname"), col("data.TAGS_MODULENAME").as("modulename"),
                        col("data.LOCALENDPOINT_SERVICENAME").as("microservicename"))
                ;
DataStreamWriter<Row> rowDataStreamWriter = rowDataset.writeStream();
DataStreamWriter<Row> option = rowDataStreamWriter.foreachBatch((row, time) -> 
            row.persist();
            row.coalesce(1).write().mode(OutputMode.Append().toString()).format("json")
                    .save("file:///D:/sparkData/CalculateFilterFromKafka/result");
            row.unpersist();

【问题讨论】：

即使合并为一个它也不会写入一个文件？同样，persist 和 unpersist 在这里也没用。

@RobertReynolds 使用 coalesce (1) 它可以批量写入一个文件，但我想要的是在一个文件中多批写入，顺便说一句，我只是使用官方网站上的持久化和非持久化示例

【参考方案1】：

也许试试这个：

        DataStreamWriter<Row> option = rowDataStreamWriter.foreachBatch((row, time) -> 
row.coalesce(1).write().mode(SaveMode.Append()).format("json").save("file:///D:/sparkData/CalculateFilterFromKafka/result");

包含 Apped 值的枚举是 SaveMode，您不必对其调用 toString()。

https://spark.apache.org/docs/latest/api/java/org/apache/spark/sql/SaveMode.html#Append