Dataflow + Datastore = DatastoreException：I/Oエラー

Question

com.google.cloud.datastoreを使用してDataFlowからデータストアに書き込もうとしています。私はこのエラーを取得してい

public void processElement(ProcessContext c) { 
    LocalDatastoreHelper HELPER = LocalDatastoreHelper.create(1.0); 
    Datastore datastore = HELPER.options().toBuilder().namespace("ghijklmnop").build().service(); 
    Key taskKey = datastore.newKeyFactory() 
     .ancestors(PathElement.of("TaskList", "default")) 
     .kind("Task") 
     .newKey("sampleTask"); 
    Entity task = Entity.builder(taskKey) 
     .set("category", "Personal") 
     .set("done", false) 
     .set("priority", 4) 
     .set("description", "Learn Cloud Datastore") 
     .build(); 
    datastore.put(task); 
} 

：

私のコードは、この（[1]の例に触発された）のように見えます

exception: "java.lang.RuntimeException: com.google.cloud.dataflow.sdk.util.UserCodeException: com.google.cloud.datastore.DatastoreException: I/O error 
at com.google.cloud.dataflow.sdk.runners.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:162) 
at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnContext.sideOutputWindowedValue(DoFnRunnerBase.java:314) 
at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnProcessContext.sideOutput(DoFnRunnerBase.java:470) 
at com.google.cloud.dataflow.sdk.transforms.Partition$PartitionDoFn.processElement(Partition.java:172) 

私はDatastoreIOシンクを使用しようとしました、ストリーミングランナーでは現在サポートされていないようです。

このエラーを回避するにはどうすればよいですか？またはDataFlowからDataStoreに書き込む最良の方法は何ですか？ @Sam McVeetyのアドバイスに従い

[1] https://github.com/GoogleCloudPlatform/java-docs-samples/blob/master/datastore/src/main/java/com/google/datastore/snippets/Concepts.java

Answer 1

、私は、データフローの外で私のデータストアのコードを分離することを試みました。そして、私は実際に同じエラーを持っています！

しかし、これはまた、私はデータフローログに表示されませんでした、例外の原因を見ることができ：

Caused by: java.net.ConnectException: Connection refused 

手がかりは、私が使っていたこのインポートラインである：com.google.cloud.datastore.testing.LocalDatastoreHelper。

これは基本的にDatastore APIをローカルに偽装するためのテストのヘルパーです。おっとっと。

これは、私はいくつかのローカルデバッグ後に今持っているコードです：

public void processElement(ProcessContext c) { 
    final Datastore datastore = DatastoreOptions.defaultInstance().service(); 
    final KeyFactory keyFactory = datastore.newKeyFactory().kind("Task"); 

    Key key = datastore.allocateId(keyFactory.newKey()); 
    Entity task = Entity.builder(key) 
     .set("description", StringValue.builder(":D").excludeFromIndexes(true).build()) 
     .set("created", DateTime.now()) 
     .set("done", false) 
     .build(); 
    datastore.put(task); 
} 

主な違いは次のとおりです。

LocalDatastoreHelper.create(1.0).options().toBuilder().namespace("ghijklmnop").build().service() 

は

DatastoreOptions.defaultInstance().service();