我正在尝试在 Flink 中编写一个需要两个阶段的计算。
在第一阶段，我从一个文本文件开始，执行一些参数估计，结果得到一个表示数据统计模型的Java对象。
在第二阶段，我想使用此对象为模拟生成数据。
我不确定该怎么做。我尝试使用LocalCollectionOutputFormat，它在本地工作，但是当我在集群上部署作业时，我得到了一个NullPointerException - 这并不奇怪。
Flink 这样做的方式是什么？
这是我的代码：
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
GlobalConfiguration.includeConfiguration(configuration);
// Phase 1: read file and estimate model
DataSource<Tuple4<String, String, String, String>> source = env
.readCsvFile(args[0])
.types(String.class, String.class, String.class, String.class);
List<Tuple4<Bayes, Bayes, Bayes, Bayes>> bayesResult = new ArrayList<>();
// Processing here...
....output(new LocalCollectionOutputFormat<>(bayesResult));
env.execute("Bayes");
DataSet<BTP> btp = env
.createInput(new BayesInputFormat(bayesResult.get(0)))
// Phase 2: BayesInputFormat generates data for further calculations
// ....
这是我得到的例外：
Error: The program execution failed: java.lang.NullPointerException
at org.apache.flink.api.java.io.LocalCollectionOutputFormat.close(LocalCollectionOutputFormat.java:86)
at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:176)
at org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
at java.lang.Thread.run(Thread.java:745)
org.apache.flink.client.program.ProgramInvocationException: The program execution failed: java.lang.NullPointerException
at org.apache.flink.api.java.io.LocalCollectionOutputFormat.close(LocalCollectionOutputFormat.java:86)
at org.apache.flink.runtime.operators.DataSinkTask.invoke(DataSinkTask.java:176)
at org.apache.flink.runtime.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:257)
at java.lang.Thread.run(Thread.java:745)
at org.apache.flink.client.program.Client.run(Client.java:328)
at org.apache.flink.client.program.Client.run(Client.java:294)
at org.apache.flink.client.program.Client.run(Client.java:288)
at org.apache.flink.client.program.ContextEnvironment.execute(ContextEnvironment.java:55)
at it.list.flink.test.Test01.main(Test01.java:62)
...
在最新版本 （0.9-milestone-1） 中，Flink 中添加了一个collect()方法
public List<T> collect()
获取DataSet<T>作为驱动程序List<T>。 collect()也会触发程序的立即执行（不需要调用ExecutionEnvironment.execute()）。目前，数据集的大小限制约为 10 MB。
如果不评估驱动程序中的模型，还可以将两个程序链接在一起，并通过附加数据接收器将模型发出到一侧。这将更有效，因为数据不会在客户端计算机上进行往返。
如果你使用的是 0.9 之前的 Flink，你可以使用以下代码片段将数据集收集到本地集合中：
val dataJavaList = new ArrayList[K]
val outputFormat = new LocalCollectionOutputFormat[K](dataJavaList)
dataset.output(outputFormat)
env.execute("collect()")
其中K是要收集的对象类型