代码之家 › 专栏 › 技术社区 › nomadic_squirrel

使用空值输出TableRow时发生NullPointerException

apache-beam google-cloud-dataflow java

nomadic_squirrel · 技术社区 · 8 年前

我想建立一个 TableRow 对象最终写入BigQuery表,但我得到一个 NullPointerException 如果我包括 null 行中的值。这是完整的stacktrace:

Exception in thread "main" org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.NullPointerException
    at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:349)
    at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:319)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:210)
    at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:66)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:311)
    at org.apache.beam.sdk.Pipeline.run(Pipeline.java:297)
    at dataflowsandbox.StarterPipeline.runTest(StarterPipeline.java:224)
    at dataflowsandbox.StarterPipeline.main(StarterPipeline.java:83)
Caused by: java.lang.NullPointerException
    at com.google.api.client.util.ArrayMap$Entry.hashCode(ArrayMap.java:419)
    at java.util.AbstractMap.hashCode(AbstractMap.java:530)
    at java.util.Arrays.hashCode(Arrays.java:4146)
    at java.util.Objects.hash(Objects.java:128)
    at org.apache.beam.sdk.util.WindowedValue$ValueInGlobalWindow.hashCode(WindowedValue.java:245)
    at java.util.HashMap.hash(HashMap.java:339)
    at java.util.HashMap.get(HashMap.java:557)
    at org.apache.beam.repackaged.beam_runners_direct_java.com.google.common.collect.AbstractMapBasedMultimap.put(AbstractMapBasedMultimap.java:191)
    at org.apache.beam.repackaged.beam_runners_direct_java.com.google.common.collect.AbstractSetMultimap.put(AbstractSetMultimap.java:130)
    at org.apache.beam.repackaged.beam_runners_direct_java.com.google.common.collect.HashMultimap.put(HashMultimap.java:48)
    at org.apache.beam.runners.direct.ImmutabilityCheckingBundleFactory$ImmutabilityEnforcingBundle.add(ImmutabilityCheckingBundleFactory.java:111)
    at org.apache.beam.runners.direct.ParDoEvaluator$BundleOutputManager.output(ParDoEvaluator.java:242)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.SimpleDoFnRunner.outputWindowedValue(SimpleDoFnRunner.java:219)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.SimpleDoFnRunner.access$700(SimpleDoFnRunner.java:69)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:517)
    at org.apache.beam.repackaged.beam_runners_direct_java.runners.core.SimpleDoFnRunner$DoFnProcessContext.output(SimpleDoFnRunner.java:505)
    at dataflowsandbox.StarterPipeline$6.procesElement(StarterPipeline.java:202)

Process finished with exit code 1

这是触发 空指针异常 :

  Pipeline p = Pipeline.create( options );

  p.apply( "kicker", Create.of( "Kick!" ) )
  .apply( "Read values", ParDo.of( new DoFn<String, TableRow>() {
     @ProcessElement
     public void procesElement( ProcessContext c ) {

        TableRow row = new TableRow();

        row.set( "ev_id",       "2323423423" );
        row.set( "customer_id", "111111"     );
        row.set( "org_id",      null         ); // Without this line, no NPE
        c.output( row );  


     } }) )
     .apply( BigQueryIO.writeTableRows()
        .to( DATA_TABLE_OUT )
        .withCreateDisposition( CREATE_NEVER )
        .withWriteDisposition( WRITE_APPEND ) );

  PipelineResult result = p.run();

我的实际代码稍微复杂一点,但是我应该能够捕获 无效的 价值观,不要把它放在一排,但也许我不明白 TableRows .

3 回复 | 直到 8 年前

Guillem Xercavins 8 年前

例如,您可以提供表架构,而只需省略设置字段的值。

表架构,其中 org_id 是 NULLABLE :

List<TableFieldSchema> fields = new ArrayList<>();
fields.add(new TableFieldSchema().setName("ev_id").setType("STRING"));
fields.add(new TableFieldSchema().setName("customer_id").setType("STRING"));
fields.add(new TableFieldSchema().setName("org_id").setType("STRING").setMode("NULLABLE"));
TableSchema schema = new TableSchema().setFields(fields);

只需不为该字段设置任何值(注释掉该行):

row.set( "ev_id",       "2323423423" );
row.set( "customer_id", "111111"     );
// row.set( "org_id",     None         ); // Without this line, no NPE
c.output( row );

在写入步骤中传递表架构:

.apply( BigQueryIO.writeTableRows()
   .to( DATA_TABLE_OUT )
   .withSchema(schema)
   .withCreateDisposition( CREATE_NEVER )
   .withWriteDisposition( WRITE_APPEND ) );

一个 NULL 值将写入BigQuery:

Sruthi Subramanyam 7 年前

如果使用的是DirectRunner,请使用参数--enforceImmutability=false。这对我有效。此问题已由Dataflow Runner处理,但在使用DirectRunner时,如果向tableRow.set()传递了null,则会遇到NPE。如果通过设置--enforceImmutability=false管道选项关闭DirectRunner的ImmutabilityEnforcement检查,则不再显示错误。

裁判: https://issues.apache.org/jira/browse/BEAM-1714

olegarino 8 年前

放置一个临时值,而不是空字符串或空字符串。据我所知,tablerrows不接受空值。