从 jar 文件调用 Dataflow 进程，出现错误 PipelineOptions 缺少名为“gcpTempLocation”的属性

Question

要求

我们正在尝试使用executable jar file 启动数据流批处理过程

流程遵循

• 使用 SDK 2.2.0 按照指令from this page 创建了一个应用
• 使用maven命令生成jar文件mvn package
• 使用此命令执行 jar 文件 java -jar DataFlow-jobs-0.1.jar --tempLocation=gs://events-dataflow/tmp --gcpTempLocation=gs://events-dataflow/tmp --project=google-project-id --runner=DataflowRunner --BQQuery='select t1.user_id google-project-id.deve.user_info t1'

输出

Exception in thread "main" java.lang.IllegalArgumentException: Class interface org.apache.beam.sdk.options.PipelineOptions missing a property named 'gcpTempLocation'.
at org.apache.beam.sdk.options.PipelineOptionsFactory.parseObjects(PipelineOptionsFactory.java:1579)
at org.apache.beam.sdk.options.PipelineOptionsFactory.access$400(PipelineOptionsFactory.java:104)
at org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.as(PipelineOptionsFactory.java:291)
at org.apache.beam.sdk.options.PipelineOptionsFactory$Builder.create(PipelineOptionsFactory.java:270)
at org.customerlabs.beam.WriteFromBQtoES.main(WriteFromBQtoES.java:98)

代码

pom.xml

  <plugin>
    <groupId>org.apache.maven.plugins</groupId>
    <artifactId>maven-assembly-plugin</artifactId>

    <configuration>
        <appendAssemblyId>false</appendAssemblyId>
        <archive>
            <manifest>
                <mainClass>org.customerlabs.beam.WriteFromBQtoES</mainClass>
            </manifest>
        </archive>
        <descriptorRefs>
            <descriptorRef>jar-with-dependencies</descriptorRef>
        </descriptorRefs>
    </configuration>

    <executions>
        <execution>
            <id>make-executable-jar</id>
            <phase>package</phase>
            <goals>
                <goal>single</goal>
            </goals>
        </execution>
    </executions>
  </plugin>

WriteFromBQtoES.java

public class WriteFromBQtoES {
    private static DateTimeFormatter fmt =
        DateTimeFormat.forPattern("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'");
    private static final Logger LOG = LoggerFactory.getLogger(WriteFromBQtoES.class);
    private static final ObjectMapper mapper = new ObjectMapper();

    public interface Options extends PipelineOptions {
        @Description("Bigquery query to fetch data")
        @Required
        String getBQQuery();
        void setBQQuery(String value);
    }

    public static void main(String[] args) throws IOException{
        PipelineOptionsFactory.register(Options.class);
        Options options = PipelineOptionsFactory.fromArgs(args).withValidation().create().as(Options.class);

        Pipeline p = Pipeline.create(options);
        PCollection<TableRow> tableRows = p.apply(BigQueryIO.read().fromQuery(options.getBQQuery()).usingStandardSql());

        tableRows.apply("WriteToCSV", ParDo.of(new DoFn<TableRow, String>() {
        // process WriteToCSV
        }))
        p.run();
    }
}

public static void main(String[] args) throws IOException{
   PipelineOptionsFactory.register(Options.class);
   Options options = PipelineOptionsFactory.fromArgs(args).withValidation().create().as(Options.class);
   String query = options.getBQQuery();
   Pipeline p = Pipeline.create(options);
   .....
   ..... pipeline operations.....
   .....
}

我不确定我们缺少什么，我们遇到了这个错误。我们在命令行中传递参数 gcpTempLocation。请帮助找出这个问题。在此先感谢

Answer 1

我认为不是你想要的 PipelineOptions：

public interface Options extends DataflowPipelineOptions { ... }

gcpTempLocation 在GcpOptions.java 中定义，并由DataflowPipelineOptions.java 扩展。

Answer 2

我也遇到过同样的问题，只是我使用了 maven shade 插件创建了一个 uber jar，其中包含应用程序所需的所有依赖项。使用 Apache Beam 所需的参数执行 jar 文件会导致相同的错误，其中找不到 -gcpTempLocation。将以下代码块添加到您的 pom.xml 将允许您使用 maven shade 来打包您的 uber jar 文件，并解决缺少参数的问题。

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-shade-plugin</artifactId>
  <version>${maven-shade-plugin.version}</version>
  <executions>
    <!-- Run shade goal on package phase -->
    <execution>
      <phase>package</phase>
      <goals>
        <goal>shade</goal>
      </goals>
      <configuration>
        <transformers>
          <!-- Required to ensure Beam Pipeline options can be passed properly. Without this, pipeline options will not be recognised -->
          <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"></transformer>
          <!-- add Main-Class to manifest file -->
          <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
            <mainClass>NAME-OF-YOUR-MAIN-CLASS</mainClass>
          </transformer>
        </transformers>
      </configuration>
    </execution>
  </executions>
</plugin>

transformer line 将确保 Beam 管道选项可以通过命令行参数传递。将此添加到您的 pom.xml 后，运行 mvn package，它将在 root/target 中生成一个 uber jar 文件。之后，您可以使用以下命令执行您的 jar 文件：

java -jar target/[your-jar-name].jar \
--runner=org.apache.beam.runners.dataflow.DataflowRunner \
--tempLocation=[GCS temp folder path] \
--stagingLocation=[GCS staging folder path]