Cloud Dataflow写入BigQuery持续报错：咨询row_errors字段位置

我们有一个运行已久的批处理Cloud Dataflow管道，近期突然开始失败。根据数据量和历史运行记录，该任务通常可在_{10分钟内完成，但最近耗时增至}8小时且无法完成，还抛出大量错误。

管道结构 #

从BigQuery读取数据 -> 在GCS中定位数据 -> 进行处理 -> 将数据写入另一个项目的BigQuery表中

报错信息 #

抛出的典型错误如下：

java.lang.RuntimeException: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: Errors found while processing rows. Please refer to the row_errors field for details. The list may not be complete because of the size limitations. Entity: projects/test123-development/datasets/datasetmy/tables/extract979_cknkc_failure/streams/Cic2NTM3NDA3NS0wMDAwLTJiMDQtODczMC1kNGY1NDdlNjZjNDQ6czI
    at org.apache.beam.sdk.io.gcp.bigquery.RetryManager.await(RetryManager.java:270)
    at org.apache.beam.sdk.io.gcp.bigquery.RetryManager.run(RetryManager.java:250)
    at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn.flushAll(StorageApiWriteUnshardedRecords.java:435)
    at org.apache.beam.sdk.io.gcp.bigquery.StorageApiWriteUnshardedRecords$WriteRecordsDoFn.finishBundle(StorageApiWriteUnshardedRecords.java:495)
Caused by: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: Errors found while processing rows. Please refer to the row_errors field for details. The list may not be complete because of the size limitations. Entity: projects/test123-development/datasets/datasetmy/tables/extract979_cknkc_failure/streams/Cic2NTM3NDA3NS0wMDAwLTJiMDQtODczMC1kNGY1NDdlNjZjNDQ6czI
    at com.google.cloud.bigquery.storage.v1.StreamWriter.requestCallback(StreamWriter.java:574)
    at com.google.cloud.bigquery.storage.v1.StreamWriter.access$1000(StreamWriter.java:49)
    at com.google.cloud.bigquery.storage.v1.StreamWriter$2.run(StreamWriter.java:225)
    at com.google.cloud.bigquery.storage.v1.StreamConnection$1.onResponse(StreamConnection.java:63)
    at com.google.cloud.bigquery.storage.v1.StreamConnection$1.onResponse(StreamConnection.java:54)
    at com.google.api.gax.tracing.TracedResponseObserver.onResponse(TracedResponseObserver.java:90)
    at com.google.api.gax.grpc.ExceptionResponseObserver.onResponseImpl(ExceptionResponseObserver.java:74)
    at com.google.api.gax.rpc.StateCheckingResponseObserver.onResponse(StateCheckingResponseObserver.java:62)
    at com.google.api.gax.grpc.GrpcDirectStreamController$ResponseObserverAdapter.onMessage(GrpcDirectStreamController.java:134)
    at io.grpc.ForwardingClientCallListener.onMessage(ForwardingClientCallListener.java:33)
    at io.grpc.ForwardingClientCallListener.onMessage(ForwardingClientCallListener.java:33)
    at io.grpc.ForwardingClientCallListener.onMessage(ForwardingClientCallListener.java:33)
    at io.grpc.internal.DelayedClientCall$DelayedListener.onMessage(DelayedClientCall.java:447)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInternal(ClientCallImpl.java:661)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1MessagesAvailable.runInContext(ClientCallImpl.java:646)
    at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
    at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
    at java.base/java.lang.Thread.run(Thread.java:833)

疑问与排查情况 #

• 不清楚报错提示中“Please refer to the row_errors field for details”里的row_errors字段在哪里查看，或如何启用该字段？
• 本地运行相同管道时结果正常，无报错。

写入BigQuery的代码实现 #

public static BigQueryIO.Write<TestClass> write() {
    return BigQueryIO.<TestClass>write()
                     .to(tableName)
                     .withFormatFunction(TestClass::toTableRow)
                     .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                     .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
                     .withFailedInsertRetryPolicy(InsertRetryPolicy.alwaysRetry())
                     .withMethod(BigQueryIO.Write.Method.STORAGE_WRITE_API);
}

如需更多信息用于进一步排查，请告知。

内容的提问来源于stack exchange，提问作者SRJ