Javaアプリケーションからのデータフロージョブの起動

Question

パラメータに基づいてバッチデータフローパイプラインを起動しようとするアプリケーションを作成しています。これについては、私はPipelineOptionsFactory.create().as(...)を使用し、その後にオプションを設定するための設定を行います。

しかし、私は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ファイルのパスを決定しようとしているように思えアップロードする。すべての必要なクラスを持つ唯一の瓶（uberjar）があります。そして明らかに、考慮されているパスは正しくありません。

プログラムでデータフロージョブを起動するために使用できる回避策はありますか。

Answer 1

クラスパスの検出とアップロードのロジックはファイルに限定されており、jarが他のjarファイルに埋め込まれているシナリオはサポートされていません。これを解決することができ、いくつかの方法がある：

1. がすべて抽出されたjarファイルを含む1瓶にマルチジャー・ジャーをフラット化。 1つのjarプロパティを保持し、パイプラインを変更するコードを書く必要がないので、これは良いことですが、これを定期的にビルドするとビルドが複雑になります。私はMaven shadeプラグインとあなたのためにこれを行うためのバンドル、またはあなたのビルドシステムに依存する同等の技術を調べます。
2. 各個別のjarを個別に指定する、より伝統的な設定を使用します。 Maven execプラグインを使用して、アプリケーションをビルドして起動するためのシナリオを手助けすることができます。
3. ランタイム中にすべてのjarを抽出し、set the filesToStage property within PipelineOptionsにステージングしたいすべてのリソースを抽出します。
4. Apache Beam/Dataflowに組み込みjarシナリオのサポートを追加しました。あなたがこの貢献をしてみたいと思ったら、私はこの追跡issueを提出しました。

あり、この非常に関連SO questionだけでなく、ユーザが実行するためuberjarを生成するために、彼らのIDEを使用して、あなたはについての質問どのように同様のシナリオを打っていたところ。