如何使用事务性DatastoreIO

Question

我正在使用来自流式数据流管道的DatastoreIO,并在使用相同的密钥编写实体时收到错误.

2016-12-10T22:51:04.385Z: Error:   (af00222cfd901860): Exception: com.google.datastore.v1.client.DatastoreException: A non-transactional commit may not contain multiple mutations affecting the same entity., code=INVALID_ARGUMENT

如果我在密钥中使用随机数,那么事情可行,但我需要更新相同的密钥,那么有没有使用DataStoreIO执行此操作的事务方法？

static class CreateEntityFn extends DoFn<KV<String, Tile>, Entity> {
  private static final long serialVersionUID = 0;

  private final String namespace;
  private final String kind;

  CreateEntityFn(String namespace, String kind) {
    this.namespace = namespace;
    this.kind = kind;
  }

  public Entity makeEntity(String key, Tile tile) {
    Entity.Builder entityBuilder = Entity.newBuilder();
    Key.Builder keyBuilder = makeKey(kind, key );
    if (namespace != null) {
      keyBuilder.getPartitionIdBuilder().setNamespaceId(namespace);
    }
    entityBuilder.setKey(keyBuilder.build());
    entityBuilder.getMutableProperties().put("tile", makeValue(tile.toString()).build()); 
    return entityBuilder.build();   
  }

  @Override
  public void processElement(ProcessContext c) { 
    String key = c.element().getKey();
    // this works    key = key.concat(":" + UUID.randomUUID().toString());
    c.output(makeEntity(key, c.element().getValue()));
  }
}

...

...
 inputData = pipeline
                .apply(PubsubIO.Read.topic(pubsubTopic));
 windowedDataStreaming = inputData
                .apply(Window.<String>into(
                      SlidingWindows.of(Duration.standardMinutes(15))
                                    .every(Duration.standardSeconds(31))));


                             ...
                             ...
                             ...
 //Create a Datastore entity 
 PCollection<Entity> siteTileEntities = tileSiteKeyed
      .apply(ParDo.named("CreateSiteEntities").of(new CreateEntityFn(options.getNamespace(), options.getKind())));       

 // write site tiles to datastore
 siteTileEntities
      .apply(DatastoreIO.v1().write().withProjectId(options.getDataset()));

 // Run the pipeline
 pipeline.run(); 

Answer 1

您的代码段不解释如何tileSiteKeyed创建.据推测它是一个PCollection<KV<String, Tile>,但如果它可能有重复的String键,这将解释这个问题.

通常,a PCollection<KV<K, V>>可以包含具有相同密钥的多个KV对.如果您想确保每个窗口的唯一键,您可以使用a GroupByKey来执行此操作.这将为您PCollection<KV<K, Iterable<V>>>提供每个窗口的唯一键.然后扩充CreateEntityFn以获取Iterable<Tile>并根据您需要进行的更改创建单个突变.