从 Dataflow api 在 Datastore 中保存长度超过 1500 字节的字符串时出错

Question

当我尝试保存一个很长的字符串时，数据流作业抛出此错误消息：属性“myProperty”的值超过 1500 个字节。代码=INVALID_ARGUMENT。

遵循 Google 的 DatastoreWordCount 示例并保存一个长度超过 1500 字节的字符串时出现错误。

我知道在使用 Datastore API 时，我可以通过将属性保存为 com.google.appengine.api.datastore.Text 来保存长度超过 1500 字节的字符串。但是，DatastoreWordCount 示例或 DatastoreHelper 类文档中没有其他选项可以表明支持 Text 类型。

是否可以使用该 api 保存如此长的字符串，以便将其读取为com.google.appengine.api.datastore.Text？

完整的错误信息如下：

java.lang.RuntimeException: com.google.cloud.dataflow.sdk.util.UserCodeException: java.lang.RuntimeException: com.google.cloud.dataflow.sdk.util.UserCodeException: java.lang.RuntimeException: com.google.cloud.dataflow.sdk.util.UserCodeException: java.lang.RuntimeException: com.google.cloud.dataflow.sdk.util.UserCodeException: com.google.datastore.v1.client.DatastoreException: The value of property "dalekTestExecutions" is longer than 1500 bytes., code=INVALID_ARGUMENT
    at com.google.cloud.dataflow.sdk.runners.worker.SimpleParDoFn$1.output(SimpleParDoFn.java:162)
    at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnContext.outputWindowedValue(DoFnRunnerBase.java:288)
    at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnContext.outputWindowedValue(DoFnRunnerBase.java:284)
    at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnProcessContext$1.outputWindowedValue(DoFnRunnerBase.java:508)
    at com.google.cloud.dataflow.sdk.util.GroupAlsoByWindowsAndCombineDoFn.closeWindow(GroupAlsoByWindowsAndCombineDoFn.java:205)
    at com.google.cloud.dataflow.sdk.util.GroupAlsoByWindowsAndCombineDoFn.processElement(GroupAlsoByWindowsAndCombineDoFn.java:192)
    at com.google.cloud.dataflow.sdk.util.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java:49)
    at com.google.cloud.dataflow.sdk.util.DoFnRunnerBase.processElement(DoFnRunnerBase.java:139)
    at com.google.cloud.dataflow.sdk.runners.worker.SimpleParDoFn.processElement(SimpleParDoFn.java:190)
    at com.google.cloud.dataflow.sdk.runners.worker.ForwardingParDoFn.processElement(ForwardingParDoFn.java:42)
    at com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerLoggingParDoFn.processElement(DataflowWorkerLoggingParDoFn.java:47)
    at com.google.cloud.dataflow.sdk.util.common.worker.ParDoOperation.process(ParDoOperation.java:55)
    at com.google.cloud.dataflow.sdk.util.common.worker.OutputReceiver.process(OutputReceiver.java:52)
    at com.google.cloud.dataflow.sdk.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java:224)
    at com.google.cloud.dataflow.sdk.util.common.worker.ReadOperation.start(ReadOperation.java:185)
    at com.google.cloud.dataflow.sdk.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java:72)
    at com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.executeWork(DataflowWorker.java:287)
    at com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.doWork(DataflowWorker.java:223)
    at com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.getAndPerformWork(DataflowWorker.java:173)
    at com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerHarness$WorkerThread.doWork(DataflowWorkerHarness.java:193)
    at com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerHarness$WorkerThread.call(DataflowWorkerHarness.java:173)
    at com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerHarness$WorkerThread.call(DataflowWorkerHarness.java:160)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

Answer 1

您可以通过从索引中排除值来保存超过 1500 字节的字符串：

Value longString = Value.newBuilder()
    .setStringValue(...)
    .setExcludeFromIndexes(true)
    .build();

如果您需要与 App Engine 的 com.google.appengine.api.datastore.Text 类型兼容，您还需要将含义设置为 15：

Value longString = Value.newBuilder()
    .setStringValue(...)
    .setExcludeFromIndexes(true)
    .setMeaning(15)
    .build();

Answer 2

DataStore 为每个属性创建索引，因此属性的默认限制为 1500 字节。现在，如果您需要存储大 JSON 之类的数据，则可以通过以下方式指定此属性不需要索引：

Entity newEntity =
                Entity.newBuilder(key)
                        .set("time", Timestamp.parseTimestamp("1970-01-01T00:00:00Z"))
                        .set("message", StringValue.newBuilder(JSON).setExcludeFromIndexes(true).build())
                        .build();

这样您将能够保存更大尺寸的数据，而不是默认的 1500 字节限制。

Answer 3

确切地说：

StringValue.newBuilder(yourString).setExcludeFromIndexes(true).build()