4

我正在尝试在 Flink 中编写一个需要两个阶段的计算。

在第一阶段,我从一个文本文件开始,执行一些参数估计,得到一个代表数据统计模型的 Java 对象。

在第二阶段,我想使用这个对象来生成模拟数据。

我不确定该怎么做。我尝试使用 a 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)
    ...
4

2 回答 2

3

在最新版本 (0.9-milestone-1)collect()中,向 Flink 添加了一个方法

public List<T> collect()

它获取驱动程序的DataSet<T>一个。还将触发程序的立即执行(不需要调用)。目前,数据集的大小限制约为 10 MB。List<T>collect()ExecutionEnvironment.execute()

如果您不在驱动程序中评估模型,您还可以将两个程序链接在一起,并通过附加数据接收器将模型发送到一边。这将更有效率,因为数据不会在客户端机器上进行往返。

于 2015-04-30T14:50:46.370 回答
0

如果您使用的是 0.9 之前的 Flink,您可以使用以下代码段将数据集收集到本地集合:

val dataJavaList = new ArrayList[K]
val outputFormat = new LocalCollectionOutputFormat[K](dataJavaList)
dataset.output(outputFormat)
env.execute("collect()")

K您要收集的对象类型在哪里

于 2015-05-01T16:30:29.847 回答