scala - writeAsCSV() 和 writeAsText() 是意外的

Question

我通过 Scala API 使用 apache flink，在某个时候我获得了一个 DataSet[(Int, Int, Int)]。使用 writeAsCSV() 和 writeAsText() 方法的结果是意外的。它创建一个目录。该目录具有方法调用的第一个参数的位置和名称(例如 filePath)。在该目录中出现两个名为“1”和“2”的文件。在这些文件中我可以看到数据集数据。他们似乎将数据集内容划分为这两个文件。尝试重新创建此行为以显示更简洁的代码片段，但我做不到。也就是说，我目睹了在预期位置创建了一个具有预期名称的文件，但没有创建目录。 val mas = ma_ groupBy(0,1) sum(2) mas.writeAsCsv("c:\flink\mas.csv")

结果会创建一个名为“mas.csv”的目录，并在其中创建两个文件“1”和“2”。什么时候会发生这样的事情呢？使用flink 9.1本地模式，Windows 7，scala 2.10，eclipse3.0.3

Answer 1

这是预期的行为。如果你想得到单个输出文件，你需要将sink的并行度设置为1。

dataset = dataset.writeAsCsv("filename").setParallelism(1);

对于 DataStream API，您需要插入额外的 rebalane() 来打破运算符链。否则，整个链将以 dop=1 执行，否则 setParallelism() 可能会被忽略。

datastream = datastream.rebalance().writeAsCsv("filename").setParallelism(1);