java - 谷歌数据流 : Write to Datastore without overwriting existing entities

Question

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

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

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

随着实体数量的增加，我想避免将所有实体加载到数据流中(而不是根据最旧的时间戳将它们分批处理)，但我遇到了旧实体在更新时被覆盖的问题不在当前批处理中。

---

我正在使用(在两个位置，一个用于现有实体，一个用于新实体)将实体写入数据流:

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

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

Answer 1

如果您想编写一个 Datastore 上不存在的新实体，您只需使用新 key 创建一个并编写它。

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