Google Dataflow：写入数据存储区而不覆盖现有实体

Question

TLDR：寻找一种在不通过 Dataflow 覆盖现有数据的情况下更新 Datastore 实体的方法

我正在使用 dataflow 2.0.0 (beam) 来更新 Google Datastore 中的实体。我的数据流从数据存储加载实体，更新它们，然后将它们保存回数据存储（覆盖现有实体）。

但是，在更新过程中，我还发现了可能已经存在或不存在的其他实体。为了防止覆盖现有实体，我之前会从 Datastore 加载所有实体并减少它们（按键分组），删除新的重复项。

随着实体数量的增加，我希望避免将所有实体加载到 Dataflow 中（而不是根据最旧的时间戳分批获取它们），但我遇到了旧实体在被覆盖时被覆盖的问题不在当前批次中。

---

我正在使用（在两个地方，一个用于现有实体，一个用于新实体）将实体写入 Dataflow：

collection.apply(DatastoreIO.v1().write().withProjectId("..."))

如果有类似DatastoreIO.v1().writeNew() 的方法，那就太好了，但遗憾的是它不存在。感谢您的帮助。

Answer 1

如果您想编写一个 Datastore 上不存在的新实体，只需使用新密钥创建一个并写入即可。

List<String> keyNames = Arrays.asList("L1", "L2"); // Somewhat you have new keys to store
PTransform<PCollection<Entity>, ?> write =
        DatastoreIO.v1().write().withProjectId(project_id); // This is a typical write operation

p.
    apply("GetInMemory", Create.of(keyNames)).setCoder(StringUtf8Coder.of()). // L1 and L2 are loaded
    apply("Proc1", ParDo.of(new DoFn<String, Entity>(){
        @ProcessElement
        public void processElement(ProcessContext c) {
            Key.Builder key = makeKey("k2", c.element());  // Generate an entity key
            final Entity entity = Entity.newBuilder().
                    setKey(key). // Set the key
                    putProperties("p1", makeValue(new String("test constant value")
                        ).setExcludeFromIndexes(true).build()).
                    build();
            c.output(entity);
        }
    })).
    apply(write); // Write them
p.run();

可以在我的代码库https://github.com/yiu31802/gcp-project/commit/cc224b34中引用整个代码