java获取数据流管道内的数据存储实体步骤

Question

我正在尝试设置一个数据流管道

1. 从数据存储中读取实体
2. 对于找到的每个实体，查找不同的实体以提取属性
3. 并使用该属性值更新原始实体

如何查找给定读取的特定实体？还是有更好的方法

以下是我目前掌握的情况：

static class LookupOtherEntityFn extends DoFn<Entity, Entity> {

    private Entity LookupOtherEntityFn(Entity sourceEntity) {
        final Key someOtherEntityKey = sourceEntity.getPropertiesMap()
                .get("otherEntityKey")
                .getKeyValue();
        final DatastoreV1.Read read = DatastoreIO.v1()
                .read(); // ...... (uses someOtherEntityKey in query)

        // **************************
        // How do I retrieve the entity given a Read?
        // **************************

    }

    @ProcessElement
    public void processElement(ProcessContext c) {
        c.output(createBackupEntity(c.element()));
    }

}

Pipeline p = Pipeline.create(options);

p.apply(DatastoreIO.v1().read()...)
    .apply(ParDo.of(new LookupOtherEntityFn()))
     .apply(ParDo.of(new CreateUpdatedEntityFn())
     .apply(DatastoreIO.v1().write()...);

p.run();

Answer 1

您将无法使用DatastoreV1.Read类从转换内部的云数据存储中读取数据DatastoreV1.Read专门用作Apache Beam管道的输入转换

您可以使用云数据存储Java客户端读取转换实现中的实体。理想情况下，您应该实现转换，以便在工作线程之间只共享一个连接，并且异步进行调用

API文档中的示例：

import com.google.cloud.datastore.Datastore;
import com.google.cloud.datastore.DatastoreOptions;
import com.google.cloud.datastore.Entity;
import com.google.cloud.datastore.Key;
import com.google.cloud.datastore.KeyFactory;

// Authentication is automatic inside Google Compute Engine
// and Google App Engine.
Datastore datastore = DatastoreOptions.getDefaultInstance().getService();
KeyFactory keyFactory = datastore.newKeyFactory().setKind(KIND);
Key key = keyFactory.newKey(keyName);
Entity entity = datastore.get(key);

参考文献：
1.https://cloud.google.com/datastore/docs/reference/libraries#client-libraries-install-java
2.https://googlecloudplatform.github.io/google-cloud-java/0.42.1/index.html
3.https://googlecloudplatform.github.io/google-cloud-java/0.42.1/apidocs/com/google/cloud/datastore/Datastore.html