小编Noa*_* Be的帖子

是否可以将 tqdm 栏从

[Step 1]: 100%|\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88\xe2\x96\x88  | 109/109 [00:03<00:00, 32.46it/s]\n

类似的东西

[Step 1]: 100%[==========================>  ] 109/109 [00:03<00:00, 32.46it/s]\n

我目前正在使用 Spark 结构化流制作原始日志数据聚合器。

输入流由文本文件目录组成：

// == Input == //

val logsDF = spark.readStream
  .format("text")
  .option("maxFilesPerTrigger", 1)
  .load("input/*")

然后解析日志...

// == Parsing == //

val logsDF2 = ...

...并汇总

// == Aggregation == //

val windowedCounts = logsDF2
  .withWatermark("window_start", "15 minutes")
  .groupBy(
    col("window"),
    col("node")
  ).count()

当我使用“控制台”接收器时，一切工作正常：结果在控制台中按浴批量更新：

// == Output == //

val query = windowedCounts.writeStream
  .format("console")
  .outputMode("complete")
  .start()
  .awaitTermination()

现在我想将结果保存在一个唯一的文件中（json、parquet、csv ..）

// == Output == //

val query = windowedCounts.writeStream
  .format("csv")
  .option("checkpointLocation", "checkpoint/")
  .start("output/")
  .awaitTermination()

但它向我输出 400 个空 csv ...我如何才能像在控制台中那样获得结果？

非常感谢 …