所以我在Apache Beam中实现这个作业,最终在数据流中运行它。所以我用直接运行器进行了测试,但是当我将其更改为数据流运行器时,它崩溃了:
java.lang.reflect.InvocationTargetException
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 org.codehaus.mojo.exec.ExecJavaMojo$1.run (ExecJavaMojo.java:293)
at java.lang.Thread.run (Thread.java:748)
Caused by: java.lang.RuntimeException: Failed to construct instance from factory method DataflowRunner#fromOptions(interface org.apache.beam.sdk.options.PipelineOptions)
at org.apache.beam.sdk.util.InstanceBuilder.buildFromMethod (InstanceBuilder.java:233)
at org.apache.beam.sdk.util.InstanceBuilder.build (InstanceBuilder.java:162)
at org.apache.beam.sdk.PipelineRunner.fromOptions (PipelineRunner.java:52)
at org.apache.beam.sdk.Pipeline.create (Pipeline.java:142)
at com.lf.myApacheBeam.MemoryTestProject.MemoryTest.main (MemoryTest.java:171)
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 org.codehaus.mojo.exec.ExecJavaMojo$1.run (ExecJavaMojo.java:293)
at java.lang.Thread.run (Thread.java:748)
Caused by: java.lang.reflect.InvocationTargetException
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 org.apache.beam.sdk.util.InstanceBuilder.buildFromMethod (InstanceBuilder.java:222)
at org.apache.beam.sdk.util.InstanceBuilder.build (InstanceBuilder.java:162)
at org.apache.beam.sdk.PipelineRunner.fromOptions (PipelineRunner.java:52)
at org.apache.beam.sdk.Pipeline.create (Pipeline.java:142)
at com.lf.myApacheBeam.MemoryTestProject.MemoryTest.main (MemoryTest.java:171)
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 org.codehaus.mojo.exec.ExecJavaMojo$1.run (ExecJavaMojo.java:293)
at java.lang.Thread.run (Thread.java:748)
Caused by: java.lang.NoSuchMethodError: org.apache.beam.runners.dataflow.options.DataflowPipelineOptions.setUserAgent(Ljava/lang/String;)V
at org.apache.beam.runners.dataflow.DataflowRunner.fromOptions (DataflowRunner.java:304)
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 org.apache.beam.sdk.util.InstanceBuilder.buildFromMethod (InstanceBuilder.java:222)
at org.apache.beam.sdk.util.InstanceBuilder.build (InstanceBuilder.java:162)
at org.apache.beam.sdk.PipelineRunner.fromOptions (PipelineRunner.java:52)
at org.apache.beam.sdk.Pipeline.create (Pipeline.java:142)
at com.lf.myApacheBeam.MemoryTestProject.MemoryTest.main (MemoryTest.java:171)
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 org.codehaus.mojo.exec.ExecJavaMojo$1.run (ExecJavaMojo.java:293)
at java.lang.Thread.run (Thread.java:748)
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 7.287 s
[INFO] Finished at: 2018-01-30T14:32:51-06:00
[INFO] Final Memory: 38M/376M
[INFO] ------------------------------------------------------------------------
[ERROR] Failed to execute goal org.codehaus.mojo:exec-maven-plugin:1.4.0:java (default-cli) on project word-count-beam: An exception occured while executing the Java class. null: InvocationTargetException: Failed to construct instance from factory method DataflowRunner#fromOptions(interface org.apache.beam.sdk.options.PipelineOptions): org.apache.beam.runners.dataflow.options.DataflowPipelineOptions.setUserAgent(Ljava/lang/String;)V -> [Help 1]
[ERROR]
[ERROR] To see the full stack trace of the errors, re-run Maven with the -e switch.
[ERROR] Re-run Maven using the -X switch to enable full debug logging.
[ERROR]
[ERROR] For more information about the errors and possible solutions, please read the following articles:
[ERROR] [Help 1] http://cwiki.apache.org/confluence/display/MAVEN/MojoExecutionException
我使用以下命令运行我的代码:
mvn compile exec:java -Dexec.mainClass=com.lf.myApacheBeam.MemoryTestProject.MemoryTest -Dexec.args="--runner=DataflowRunner"
如果我使用直接更改数据流,那么我的代码就可以工作并运行。
它崩溃的行是:
Pipeline pipeline = Pipeline.create(options);
我的"选项"声明为:
MemoryTestExtractOptions options = PipelineOptionsFactory.fromArgs(args)
.withValidation()
.as(MemoryTestExtractOptions.class);
options.setStreaming(true);
//options.setRunner(DataflowRunner.class);
options.setTempLocation("gs://blahblahblah/temp/");
options.setBigQuerySchema(TestResultToRowConverter.getSchema());
最后,我的界面"MemoryTestExtractOptions"是:
private interface MemoryTestExtractOptions
extends Options, BigQueryTableOptionsForMemoryTest, StreamingOptions, PubsubTopicAndSubscriptionOptions, DataflowPipelineOptions//, DataflowPipelineOptions
{
@Description("BigQuery table to write to, specified as "
+ "<project_id>:<dataset_id>.<table_id>. The dataset must already exist.")
String getOutput();
void setOutput(String value);
}
该错误抱怨"DataflowPipelineOptions.setUserAgent",但我没有调用它,我检查了我的选项,它们都格式良好。我是否缺少数据流的特定内容?谢谢。
请确保您的pom.xml
将org.apache.beam:beam-runners-google-cloud-dataflow-java
工件列为依赖项。如果使用beam-sdks-java-maven-archetypes-examples
原型生成项目,则可以将-Pdataflow-runner
添加到 maven 命令行。
下的 Beam 快速入门说明中有一个示例:https://beam.apache.org/get-started/quickstart-java/
发现问题。我的pom.xml是指定数据流运行器的版本,而不是匹配${beam.version}
我改变了,它就像一个魅力。
感谢您为我指明正确方向的提示