Dataflow流水线BigQueryIO写入时出现空指针异常求助

我的Dataflow流水线执行BigQueryIO写入操作时抛出了NullPointerException，但异常涉及的所有值都已确认正确定义。流程是从数据库读取数据、转换结果集后，尝试基于结果集的表行在已有数据集内创建表。已确认传入BigQueryIO.writeTableRows()的所有参数均有效，但写入步骤仍抛出异常。

相关代码 #

// 获取首次查询结果
WriteResult results = pipeline
        .apply("Connect", JdbcIO.<TableRow>read()
                .withDataSourceConfiguration(buildDataSourceConfig(options, URL))
                .withQuery(query)
                .withRowMapper(new JdbcIO.RowMapper<TableRow>() {
                  // 将ResultSet转换为PCollection
                  public TableRow mapRow(ResultSet rs) throws Exception {

                    ResultSetMetaData md = rs.getMetaData();
                    int columnCount = md.getColumnCount();
                    TableRow tr = new TableRow();
                    for (int i = 1; i <= columnCount; i++ ) {
                      String name = md.getColumnName(i);
                      tr.set(name, rs.getString(name));
                    }
                    return tr;
                  }
                }))
        .setCoder(TableRowJsonCoder.of())
        .apply("Write to BQ",
                BigQueryIO.writeTableRows()
                        .withSchema(schema)
                        .to(dataset)
                        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)
                        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED));

异常栈信息 #

2023-01-10T20:33:22.4214526Z WARNING: Unable to infer a schema for type com.google.api.services.bigquery.model.TableRow. Attempting to infer a coder without a schema.
2023-01-10T20:33:22.4216783Z Exception in thread "main" java.lang.NullPointerException
2023-01-10T20:33:22.4218945Z    at org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$Write.validateNoJsonTypeInSchema(BigQueryIO.java:3035)
2023-01-10T20:33:22.4221029Z    at org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$Write.continueExpandTyped(BigQueryIO.java:2949)
2023-01-10T20:33:22.4222727Z    at org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$Write.expandTyped(BigQueryIO.java:2880)
2023-01-10T20:33:22.4226464Z    at org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$Write.expand(BigQueryIO.java:2776)
2023-01-10T20:33:22.4228072Z    at org.apache.beam.sdk.io.gcp.bigquery.BigQueryIO$Write.expand(BigQueryIO.java:1786)
2023-01-10T20:33:22.4234778Z    at org.apache.beam.sdk.Pipeline.applyInternal(Pipeline.java:548)
2023-01-10T20:33:22.4237961Z    at org.apache.beam.sdk.Pipeline.applyTransform(Pipeline.java:499)
2023-01-10T20:33:22.4240010Z    at org.apache.beam.sdk.values.PCollection.apply(PCollection.java:376)
2023-01-10T20:33:22.4242466Z    at edu.mayo.mcc.aide.sqaTransfer.SqaTransfer.buildPipeline(SqaTransfer.java:133)
2023-01-10T20:33:22.4244722Z    at edu.mayo.mcc.aide.sqaTransfer.SqaTransfer.main(SqaTransfer.java:99)
2023-01-10T20:33:22.4246444Z . exit status 1

问题定位与修复方案 #

从异常栈可以看出，NPE发生在validateNoJsonTypeInSchema方法中，核心原因是传入的schema对象为null，或schema内的字段定义存在null值。以下是具体排查和修复步骤：

1. 检查schema初始化逻辑
   确保schema变量通过TableSchema构造器正确构建，所有Field对象都已明确设置名称、类型和模式，无null值。示例：

   TableSchema schema = new TableSchema()
       .setFields(Arrays.asList(
           new Field().setName("user_id").setType("STRING").setMode("REQUIRED"),
           new Field().setName("user_name").setType("STRING").setMode("NULLABLE")
       ));
   

2. 验证dataset参数格式
   确认dataset参数是符合要求的TableDestination格式，例如"项目ID:数据集ID.表名"，避免格式错误间接引发schema校验异常。

3. 处理警告信息
   日志中的“无法推断TableRow的schema”警告虽不直接触发NPE，但需确保转换后的TableRow结构与传入的schema完全匹配，避免后续写入时出现数据不兼容问题。

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