如何寫多個Datastream的單個文件

Question

比方說，我有兩個不同類型的DataStream的：

val stream1: DataStream[(Int, Int, Int)] = ... 
val stream2: DataStream[(Int, Int, Int, Int, Float)] = ... 

我怎麼能寫兩個流成單個文件？

我試過不同的事情，但似乎沒有工作。舉例來說，我不能只是寫直線距離爲

stream1.writeAsText("path/to/file.txt").setParallelism(1) 
stream2.writeAsText("path/to/file.txt").setParallelism(1) 

因爲弗林克將與以下消息抱怨：

java.io.IOException: File or directory already exists. 
Existing files and directories are not overwritten in NO_OVERWRITE mode. 
Use OVERWRITE mode to overwrite existing files and directories. 

在另一方面，我不能覆蓋這樣的：

stream1.writeAsText("path/to/file.txt").setParallelism(1) 
stream2.writeAsText("path/to/file.txt", FileSystem.WriteMode.OVERWRITE).setParallelism(1) 

因爲（據我所知）第二個流將覆蓋第一個流寫入的內容。

最後，我想到了連接流這樣

val connectedStream: ConnectedStream = stream1.connect(stream2) 

但我會得到一個ConnectedStream，它不具有writeAsText方法。 （爲了記錄，我實際上有4個流，我想要寫入一個文件）。

Answer 1

一個非常簡單的解決方案是使用映射器將每個事件映射到String（或另一種常見類型，如byte[]）。然後，您有四個相同類型的流（DataStream[String]），您可以將它們合併爲一個流並將其作爲一個流寫入文件。

這將如下所示：

val s1: DataStream[String] = ??? 
val s2: DataStream[String] = ??? 
val s3: DataStream[String] = ??? 
val s4: DataStream[String] = ??? 

val out: DataStream[String] = s1.union(s2).union(s3).union(s4) 
out.writeAsText("path/to/file")