java - Google Cloud Dataflow DatastoreIO 读写依赖问题

Question

我需要什么:依赖项版本的正确组合，以便能够通过 DatastoreIO.v1().read/write 在 Dataflow (v. 1.9.0) 中读取和写入数据存储，以及需要哪些依赖项pom 中引用？

在 pom 中引用 Dataflow 1.9.0 的 mavenrepo 中的数据流特定依赖项:

com.google.cloud.dataflow/google-cloud-dataflow-java-sdk-all/1.9.0
com.google.cloud.datastore/datastore-v1-protos/1.0.1
com.google.cloud.datastore/datastore-v1-proto-client/1.1.0
com.google.protobuf/protobuf-java/3.0.0-beta-1

写入数据存储时(实际上是构建实体时)，我收到以下异常:

// CamelExecutionException (Setup running with Camel-Routes, but for development purposes not in Fuse but as a local CamelRoute in Eclipse)
Caused by: java.lang.NoClassDefFoundError: com/google/protobuf/GeneratedMessageV3
    at java.lang.ClassLoader.defineClass1(Native Method)
    at java.lang.ClassLoader.defineClass(ClassLoader.java:763)
    at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
    at java.net.URLClassLoader.defineClass(URLClassLoader.java:467)
    at java.net.URLClassLoader.access$100(URLClassLoader.java:73)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:368)
    at java.net.URLClassLoader$1.run(URLClassLoader.java:362)
    at java.security.AccessController.doPrivileged(Native Method)
    at java.net.URLClassLoader.findClass(URLClassLoader.java:361)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    at com.google.datastore.v1.Value.toBuilder(Value.java:749)
    at com.google.datastore.v1.Value.newBuilder(Value.java:743)
    at xmlsource.dataflow.test.EntityUtil.getStringValue(EntityUtil.java:404)
    at xmlsource.dataflow.test.EntityUtil.getArticleEntity(EntityUtil.java:152)
    at xmlsource.dataflow.test.parser.ArticleToEntity.processElement(ArticleToEntity.java:21)
    at com.google.cloud.dataflow.sdk.util.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:49)
    at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase.processElement(DoFnRunnerBase.java:139)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateHelper(ParDo.java:1229)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateSingleHelper(ParDo.java:1098)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.access$300(ParDo.java:457)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$1.evaluate(ParDo.java:1084)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$1.evaluate(ParDo.java:1079)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.visitTransform(DirectPipelineRunner.java:858)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:221)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:103)
    at com.google.cloud.dataflow.sdk.Pipeline.traverseTopologically(Pipeline.java:260)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.run(DirectPipelineRunner.java:814)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.run(DirectPipelineRunner.java:526)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.run(DirectPipelineRunner.java:96)
    at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:181)
    at xmlsource.dataflow.test.PipelineParseTest.createAndRun(PipelineParseTest.java:208)
    at xmlsource.dataflow.test.PipelineTester.process(PipelineTester.java:11)
    at org.apache.camel.processor.DelegateSyncProcessor.process(DelegateSyncProcessor.java:63)
    ... 8 more

xmlsource.dataflow.test.EntityUtil.getStringValue(EntityUtil.java:404) 中引用的行:

Value.newBuilder().setStringValue(value).build();

当阅读或多或少相同时:

java.lang.NoClassDefFoundError: com/google/protobuf/GeneratedMessageV3
…

将依赖项更改为(仅不是 protobuf-java 的 beta 版本)时

com.google.cloud.datastore/datastore-v1-protos/1.0.1
com.google.cloud.datastore/datastore-v1-proto-client/1.1.0
com.google.protobuf/protobuf-java/3.0.0

并尝试写入，出现以下异常:

// CamelExecutionException...
Caused by: java.lang.VerifyError: Bad type on operand stack
Exception Details:
  Location:
    com/google/datastore/v1/Value$Builder.mergeGeoPointValue(Lcom/google/type/LatLng;)Lcom/google/datastore/v1/Value$Builder; @76: invokevirtual
  Reason:
    Type 'com/google/type/LatLng' (current frame, stack[1]) is not assignable to 'com/google/protobuf/GeneratedMessage'
  Current Frame:
    bci: @76
    flags: { }
    locals: { 'com/google/datastore/v1/Value$Builder', 'com/google/type/LatLng' }
    stack: { 'com/google/protobuf/SingleFieldBuilder', 'com/google/type/LatLng' }
  Bytecode:
    someBytecode                                    
  Stackmap Table:
    same_frame(@50)
    same_frame(@55)
    same_frame(@62)
    same_frame(@80)
    same_frame(@89)

    at com.google.datastore.v1.Value.toBuilder(Value.java:749)
    at com.google.datastore.v1.Value.newBuilder(Value.java:743)
    at xmlsource.dataflow.test.EntityUtil.getStringValue(EntityUtil.java:404)
    at xmlsource.dataflow.test.EntityUtil.getArticleEntity(EntityUtil.java:152)
    at xmlsource.dataflow.test.parser.ArticleToEntity.processElement(ArticleToEntity.java:21)
    at com.google.cloud.dataflow.sdk.util.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:49)
    at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase.processElement(DoFnRunnerBase.java:139)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateHelper(ParDo.java:1229)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateSingleHelper(ParDo.java:1098)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.access$300(ParDo.java:457)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$1.evaluate(ParDo.java:1084)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$1.evaluate(ParDo.java:1079)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.visitTransform(DirectPipelineRunner.java:858)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:221)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:103)
    at com.google.cloud.dataflow.sdk.Pipeline.traverseTopologically(Pipeline.java:260)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.run(DirectPipelineRunner.java:814)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.run(DirectPipelineRunner.java:526)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.run(DirectPipelineRunner.java:96)
    at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:181)
    at xmlsource.dataflow.test.PipelineParseTest.createAndRun(PipelineParseTest.java:208)
    at xmlsource.dataflow.test.PipelineTester.process(PipelineTester.java:11)
    at org.apache.camel.processor.DelegateSyncProcessor.process(DelegateSyncProcessor.java:63)

此处异常引用了函数 mergeGeoPointValue，而我的代码从未调用任何函数来设置 LatLng 或 GeoPoint 值。我的代码中引用的行再次只是设置了 String-Value

在阅读时，在将 POJO 转换为数据存储实体时，我再次遇到了相同的异常

Value.newBuilder().setStringValue("someString").build()

整个查询:

Query query = Query.newBuilder()
  .addKind(KindExpression.newBuilder()
    .setName("test_article").build())
  .setFilter(Filter.newBuilder()
    .setPropertyFilter(PropertyFilter.newBuilder()
      .setProperty(PropertyReference.newBuilder()
        .setName("somePropertyName"))
        .setOp(PropertyFilter.Operator.EQUAL)
        .setValue(Value.newBuilder()
          .setStringValue("someString").build())                                
      .build())
    .build())
  .build();

将依赖项更改为 (datastore-v1-protos/1.3.0):

com.google.cloud.datastore/datastore-v1-protos/1.3.0
com.google.cloud.datastore/datastore-v1-proto-client/1.1.0
com.google.protobuf/protobuf-java/3.0.0 (or 3.2.0)

通过此设置，我可以通过 .apply(DatastoreIO.v1().write().withProjectId("someProjectId")); 成功写入数据存储；

尝试读取时，查询对象已成功构建，但是...:

// CamelExecutionException
Caused by: java.lang.NoSuchMethodError: com.google.datastore.v1.Query$Builder.clone()Lcom/google/protobuf/GeneratedMessage$Builder;
    at com.google.cloud.dataflow.sdk.io.datastore.DatastoreV1$Read$ReadFn.processElement(DatastoreV1.java:648)
    at com.google.cloud.dataflow.sdk.util.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:49)
    at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase.processElement(DoFnRunnerBase.java:139)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateHelper(ParDo.java:1229)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.evaluateSingleHelper(ParDo.java:1098)
    at com.google.cloud.dataflow.sdk.transforms.ParDo.access$300(ParDo.java:457)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$1.evaluate(ParDo.java:1084)
    at com.google.cloud.dataflow.sdk.transforms.ParDo$1.evaluate(ParDo.java:1079)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.visitTransform(DirectPipelineRunner.java:858)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:221)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformTreeNode.visit(TransformTreeNode.java:217)
    at com.google.cloud.dataflow.sdk.runners.TransformHierarchy.visit(TransformHierarchy.java:103)
    at com.google.cloud.dataflow.sdk.Pipeline.traverseTopologically(Pipeline.java:260)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner$Evaluator.run(DirectPipelineRunner.java:814)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.run(DirectPipelineRunner.java:526)
    at com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner.run(DirectPipelineRunner.java:96)
    at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:181)
    at xmlsource.dataflow.test.PipelineParseTest.createAndRun(PipelineParseTest.java:208)
    at xmlsource.dataflow.test.PipelineTester.process(PipelineTester.java:11)
    at org.apache.camel.processor.DelegateSyncProcessor.process(DelegateSyncProcessor.java:63)
    ... 8 more

我尝试从数据存储区读取的行:

PCollection<Entity> entityCollection = p.apply(
  DatastoreIO.v1().read().withNamespace("test_ns_df")
    .withProjectId("someProjectId")
    .withQuery(query));

编辑:使用 GitHubDataflowExample 中的依赖项(和父 pom)时我再次收到 java.lang.NoClassDefFoundError: com/google/protobuf/GenerateMessageV3在构建查询值时....

所以我从来没有让阅读起作用...有没有人遇到过类似的问题并找到如何解决这个问题？或者我需要以不同的方式建立值(value)观吗​​？使用 DatastoreHelper.makeValue 时也会发生相同的异常...工作项目中引用的依赖项也会有很大帮助!

我认为这将是一个依赖/版本问题，但也许你们中的某个人更了解。我不可能是第一个遇到 java.lang.NoSuchMethodError: com.google.datastore.v1.Query$Builder.clone() 问题的人，就像这个人 NoSuchMethodError in DatastoreWordCount example谁刚刚拉了一个错误的版本，但就我而言，这并没有成功。

提前致谢

Answer 1

发现问题:

由于在与 Camel-Fuse 一起运行的同一项目中存在一个预处理，将文件存储在 Google Storage 中，因此我依赖于 google-storage:

<dependency>
    <groupId>com.google.cloud</groupId>
    <artifactId>google-cloud-storage</artifactId>
    <version>0.6.0</version>
</dependency>

在 pom.xml 中数据流依赖之前提到了此依赖。切换依赖关系的顺序(数据存储之前的数据流)并删除所有其他依赖关系后，DatastoreIO 完美运行!然后，根据您的操作(例如 XMLSource)，需要添加一些运行时依赖项