java - 从 Java 应用程序启动数据流作业

Question

我正在编写一个应用程序，它尝试根据提供的参数启动批处理数据流管道。为此，我使用 PipelineOptionsFactory.create().as(...) 然后使用 setter 来配置选项。

但是当我使用 Pipeline.create(opts) 创建管道对象对象时，出现以下错误:

04:48:08.711 [pool-4-thread-9] ERROR c.g.c.d.s.r.DataflowPipelineRunner - Unable to convert url (jar:file:/ferric.jar!/) to file.
04:48:08.711 [pool-4-thread-9] WARN  BatchJobManager - unable to start materialization for view
java.lang.RuntimeException: Failed to construct instance from factory method DataflowPipelineRunner#fromOptions(interface com.google.cloud.dataflow.sdk.options.PipelineOptions)
    at com.google.cloud.dataflow.sdk.util.InstanceBuilder.buildFromMethod(InstanceBuilder.java:234)
    at com.google.cloud.dataflow.sdk.util.InstanceBuilder.build(InstanceBuilder.java:163)
    at com.google.cloud.dataflow.sdk.runners.PipelineRunner.fromOptions(PipelineRunner.java:58)
    at com.google.cloud.dataflow.sdk.Pipeline.create(Pipeline.java:135)
    at com.brightcove.rna.dataflow.MaterializationPipeline.<init>(MaterializationPipeline.java:45)
    at com.brightcove.rna.dataflow.MaterializationPipeline.create(MaterializationPipeline.java:92)
    at com.brightcove.rna.ferric.DataflowJobService.createPipeline(DataflowJobService.java:121)
    at javaslang.control.Try.mapTry(Try.java:410)
    at javaslang.control.Try.map(Try.java:380)
    at com.brightcove.rna.ferric.DataflowJobService.create(DataflowJobService.java:102)
    at com.brightcove.rna.ferric.BatchJobScheduler.lambda$null$13(BatchJobScheduler.java:94)
    at javaslang.Value.forEach(Value.java:246)
    at com.brightcove.rna.ferric.BatchJobScheduler.lambda$startMaterializationJobs$14(BatchJobScheduler.java:91)
    at javaslang.control.Try.onSuccess(Try.java:442)
    at com.brightcove.rna.ferric.BatchJobScheduler.startMaterializationJobs(BatchJobScheduler.java:90)
    at com.brightcove.rna.ferric.BatchJobScheduler.run(BatchJobScheduler.java:52)
    at sun.reflect.GeneratedMethodAccessor94.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.springframework.scheduling.support.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:65)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.reflect.InvocationTargetException: null
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at com.google.cloud.dataflow.sdk.util.InstanceBuilder.buildFromMethod(InstanceBuilder.java:223)
    ... 27 common frames omitted
Caused by: java.lang.IllegalArgumentException: Unable to convert url (jar:file:/ferric.jar!/) to file.
    at com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner.detectClassPathResourcesToStage(DataflowPipelineRunner.java:3176)
    at com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner.fromOptions(DataflowPipelineRunner.java:291)
    ... 32 common frames omitted
Caused by: java.lang.IllegalArgumentException: URI is not hierarchical
    at java.io.File.<init>(File.java:418)
    at com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner.detectClassPathResourcesToStage(DataflowPipelineRunner.java:3172)
    ... 33 common frames omitted

管道运行器似乎正在尝试确定包含需要上传的类的 jar 的路径。只有一个 jar (uberjar) 包含所有必需的类。显然，所考虑的路径是不正确的。

我可以使用哪些可能的解决方法以编程方式启动数据流作业。

Answer 1

类路径检测和上传逻辑仅限于文件，不支持 jar 嵌入其他 jar 的场景。您可以通过以下方式解决此问题:

1. 将多 jar jar 压平成一个 jar ，其中包含所有提取的 jar 。这很好，因为您保留了一个 jar 属性并且不需要编写任何代码来更改管道，但是如果您定期构建它，将使您的构建更加复杂。我会调查 Maven shade插件和捆绑为您或依赖于您的构建系统的等效技术执行此操作。
2. 使用更传统的设置，分别指定每个单独的 jar。您可以使用 Maven exec插件来帮助构建和启动您的应用程序。
3. 在运行时和 set the filesToStage property within PipelineOptions 中提取所有 jar包含您想要暂存的所有资源。
4. 向 Apache Beam/Dataflow 添加对嵌入式 jar 方案的支持。我提交了这个跟踪issue如果你想看看执行此贡献。

有这个非常相关SO question以及用户使用他们的 IDE 生成用于执行的 uberjar 并遇到与您询问的内容类似的情况。