我正在尝试修改bigquery-dataflow示例以处理CSV文件。 https://github.com/GoogleCloudPlatform/bigquery-etl-dataflow-sample
我已经将读取对象更改为解析 CSV 并将其添加到基准对象。运行管道时，我收到以下错误：
(2b01c6a9d56ae128)： java.lang.RuntimeException： com.google.cloud.dataflow.sdk.coders.CoderException： 无法编码 空长在 com.google.cloud.dataflow.sdk.runners.worker.SimpleParDoFn$1.output(SimpleParDoFn.java：160) 在 com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnContext.outputWindowedValue(DoFnRunnerBase.java：288) 在 com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnProcessContext.output(DoFnRunnerBase.java：450) 在 com.google.cloud.dataflow.sdk.transforms.MapElements$1.processElement(MapElements.java：109) 由以下原因引起：com.google.cloud.dataflow.sdk.coders.CoderException： 不能 编码空长于 com.google.cloud.dataflow.sdk.coders.VarLongCoder.getEncodedElementByteSize(VarLongCoder.java：92) 在 com.google.cloud.dataflow.sdk.coders.VarLongCoder.getEncodedElementByteSize(VarLongCoder.java：34) 在 com.google.cloud.dataflow.sdk.coders.StandardCoder.registerByteSizeObserver(StandardCoder.java：185) 在 com.google.cloud.dataflow.sdk.coders.KvCoder.registerByteSizeObserver(KvCoder.java：156) 在 com.google.cloud.dataflow.sdk.coders.KvCoder.registerByteSizeObserver(KvCoder.java：42) 在 com.google.cloud.dataflow.sdk.util.WindowedValue$FullWindowedValueCoder.registerByteSizeObserver(WindowedValue.java：641) 在 com.google.cloud.dataflow.sdk.util.WindowedValue$FullWindowedValueCoder.registerByteSizeObserver(WindowedValue.java：552) 在 com.google.cloud.dataflow.sdk.runners.worker.MapTaskExecutorFactory$ElementByteSizeObservableCoder.registerByteSizeObserver(MapTaskExecutorFactory.java：351) 在 com.google.cloud.dataflow.sdk.util.common.worker.OutputObjectAndByteCounter.update(OutputObjectAndByteCounter.java：125) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowOutputCounter.update(DataflowOutputCounter.java：61) 在 com.google.cloud.dataflow.sdk.util.common.worker.OutputReceiver.process(OutputReceiver.java：46) 在 com.google.cloud.dataflow.sdk.runners.worker.SimpleParDoFn$1.output(SimpleParDoFn.java：158) 在 com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnContext.outputWindowedValue(DoFnRunnerBase.java：288) 在 com.google.cloud.dataflow.sdk.util.DoFnRunnerBase$DoFnProcessContext.output(DoFnRunnerBase.java：450) 在 com.google.cloud.dataflow.sdk.transforms.MapElements$1.processElement(MapElements.java：109) 在 com.google.cloud.dataflow.sdk.util.SimpleDoFnRunner.invokeProcessElement(SimpleDoFnRunner.java：49) 在 com.google.cloud.dataflow.sdk.util.DoFnRunnerBase.processElement(DoFnRunnerBase.java：139) 在 com.google.cloud.dataflow.sdk.runners.worker.SimpleParDoFn.processElement(SimpleParDoFn.java：188) 在 com.google.cloud.dataflow.sdk.runners.worker.ForwardingParDoFn.processElement(ForwardingParDoFn.java：42) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerLoggingParDoFn.processElement(DataflowWorkerLoggingParDoFn.java：47) 在 com.google.cloud.dataflow.sdk.util.common.worker.ParDoOperation.process(ParDoOperation.java：55) 在 com.google.cloud.dataflow.sdk.util.common.worker.OutputReceiver.process(OutputReceiver.java：52) 在 com.google.cloud.dataflow.sdk.util.common.worker.ReadOperation.runReadLoop(ReadOperation.java：221) 在 com.google.cloud.dataflow.sdk.util.common.worker.ReadOperation.start(ReadOperation.java：182) 在 com.google.cloud.dataflow.sdk.util.common.worker.MapTaskExecutor.execute(MapTaskExecutor.java：69) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.executeWork(DataflowWorker.java：285) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.doWork(DataflowWorker.java：221) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowWorker.getAndPerformWork(DataflowWorker.java：171) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerHarness$WorkerThread.doWork(DataflowWorkerHarness.java：192) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerHarness$WorkerThread.call(DataflowWorkerHarness.java：172) 在 com.google.cloud.dataflow.sdk.runners.worker.DataflowWorkerHarness$WorkerThread.call(DataflowWorkerHarness.java：159) at java.util.concurrent.FutureTask.run(FutureTask.java：266) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java：1142) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java：617) at java.lang.Thread.run(Thread.java：745)
错误与我在这个问题(我的)上看到的错误相同。
您可能需要仔细检查是否正在使用 KV，但您不小心将 NULL 值分配给 K(如果 CSV 文件对要用作 K 的字段具有空值，则很容易发生这种情况)