【发布时间】:2015-04-30 22:17:31
【问题描述】:
我需要定期运行 Dataflow 管道。 The FAQ for Dataflow 声明如下:
您可以通过 Google App Engine 或 GCE 上的自定义 (CRON) 作业流程自动执行管道。 SDK 的未来版本将支持命令行选项,以更精细地控制作业管理。
我尝试使用以下代码从我的 Java 应用程序运行一个非常简单的管道:
public class MyAnalyticsServlet extends HttpServlet {
@Override
public void doGet(HttpServletRequest req, HttpServletResponse resp) throws IOException {
resp.setContentType("text/plain");
if (req.getRequestURI().equals("/dataflow/test")) {
DataflowPipelineOptions options = PipelineOptionsFactory.create().as(DataflowPipelineOptions.class);
options.setProject("redacted");
options.setRunner(DataflowPipelineRunner.class);
Pipeline p = Pipeline.create(options);
p.apply(TextIO.Read.named("TestInput").from("gs://redacted/test/in.txt"))
.apply(new TestTransform())
.apply(TextIO.Write.named("TestOutput")
.to("gs://redacted/test")
.withNumShards(0));
p.run();
} else {
resp.setStatus(404);
resp.getWriter().println("Not Found");
return;
}
resp.getWriter().println("OK");
}
}
我收到以下错误:
java.lang.IllegalArgumentException: Methods [setRunner(Class), getRunner()] on [com.google.cloud.dataflow.sdk.options.PipelineOptions] do not conform to being bean properties.
at com.google.common.base.Preconditions.checkArgument(Preconditions.java:145)
at com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory.validateClass(PipelineOptionsFactory.java:1059)
...
有什么想法吗?
【问题讨论】:
-
我有一个在 AppEngine 中运行 wordcount 的简单管道,我使用最新的 Dataflow SDK (search.maven.org/…)。您是从 Github 构建还是使用 Maven 的最新版本?
-
您使用的是什么版本的 SDK?
-
其他人能否确认能够从 AppEngine 中启动数据流?
-
This answer 详细介绍了在 Google AppEngine 上运行 Dataflow。
标签: java google-app-engine google-cloud-dataflow