Dataflow + Datastore = DatastoreException: I/O err

问题:

I'm trying to write to DataStore from DataFlow using com.google.cloud.datastore.

My code looks like this (inspired by the examples in [1]):

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);
}

I'm getting this error:

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)

I have tried to use the DatastoreIO sink, but it looks like it is not currently supported in the streaming runner.

How can I avoid that error ? or What's the best way to write from DataFlow to DataStore ?

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

回答1:

Following @Sam McVeety advice, I tried to isolate my Datastore code outside of Dataflow. And I indeed got the same error !

But this also allowed me to see the cause of the exception, which I didn't see in Dataflow logs:

Caused by: java.net.ConnectException: Connection refused

The clue is in this import line that I was using: com.google.cloud.datastore.testing.LocalDatastoreHelper.

It's a helper for tests that takes care of basically mocking Datastore API locally. Oops.

So this is the code that I've got now after some local debugging:

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);
}

The main difference is:

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

Becomes

DatastoreOptions.defaultInstance().service();