DataStream
的writeAsText
或writeAsCsv
方法会根据工作线程的数量写入相应数量的文件。据我所知,这些方法只允许您指定这些文件的路径和一些格式设置。为了调试和测试目的,能够将所有内容打印到一个单独的文件中,而不必更改设置以使用单个工作线程,这将非常有用。
是否有任何非过于复杂的方法可以实现此功能?我怀疑可以通过实现自定义
SinkFunction
来实现,但对此并不确定(此外,这似乎对于看起来相对简单的事情来说也有点麻烦)。DataStream
的writeAsText
或writeAsCsv
方法会根据工作线程的数量写入相应数量的文件。据我所知,这些方法只允许您指定这些文件的路径和一些格式设置。SinkFunction
来实现,但对此并不确定(此外,这似乎对于看起来相对简单的事情来说也有点麻烦)。将并行度设置为1可以实现此目的。这样写操作只会在一台机器上进行。
writeAsText(path).setParallelism(1);
val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
env.setParallelism(1)
val sink: StreamingFileSink[String] = StreamingFileSink
.forRowFormat(new Path(outPath), new SimpleStringEncoder[String]("UTF-8"))
.build()
dataStream.map(_.toString).addSink(sink)
StreamingFileSink
在较新的版本中已被弃用,请改用 org.apache.flink.connector.file.sink.FileSink
。 - Sambhav Khare