Spark Java读取BigQuery中JSON列解析失败问题求助

可能的问题点及对应解决方法 #

1. JSON字符串存在转义或外层引号 #

BigQuery中存储的sender列大概率是转义后的嵌套JSON字符串（比如双引号被转义为\"，或整个JSON被外层引号包裹），Spark的from_json无法直接识别这种格式，导致解析失败返回空结构。

解决方法：先对sender列做格式清理，去掉外层引号和多余转义符：

// 去掉外层引号 + 清理转义字符
Dataset<Row> cleanedDf = df.withColumn("cleaned_sender", 
        functions.regexp_replace(
            functions.substring(df.col("sender"), 2, df.col("sender").length()-2), 
            "\\\\", ""
        )
);

2. Schema字段类型或nullable设置不匹配 #

• UUID字段类型冲突：示例中的UUID是数值，但如果BigQuery中该字段实际存储为带引号的字符串（比如"32341983"），用IntegerType解析会直接失败，导致整个UserInfo结构为空。建议先改为StringType测试：

  userInfo.add("UUID", DataTypes.StringType, true);
  

• nullable设置过严：你将UserInfo设为不可为空（false），只要解析过程中任何字段不匹配，就会返回空结构。建议先将所有字段的nullable改为true，便于排查问题：

  StructType userInfo = new StructType()
          .add("CorporateEmailAddress", DataTypes.StringType, true)
          .add("UUID", DataTypes.StringType, true)
          .add("FirstName", DataTypes.StringType, true)
          .add("FirmNumber", DataTypes.IntegerType, true)
          .add("PersonalEmailAddress", DataTypes.StringType, true)
          .add("LastName", DataTypes.StringType, true)
          .add("AccountName", DataTypes.StringType, true)
          .add("AccountNumber", DataTypes.IntegerType, true);
  
  StructType schema = new StructType().add("UserInfo", userInfo, true);
  

3. 先验证原始JSON格式 #

在解析前先查看sender列的原始内容，确认是否为标准JSON：

df.select("sender").show(false);

如果输出的内容带有额外的转义或引号，就需要针对性做清理。

修正后的完整示例代码 #

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;

public class BigQueryJsonParse {
    public static void main(String[] args) {
        StructType userInfo = new StructType()
                .add("CorporateEmailAddress", DataTypes.StringType, true)
                .add("UUID", DataTypes.StringType, true)
                .add("FirstName", DataTypes.StringType, true)
                .add("FirmNumber", DataTypes.IntegerType, true)
                .add("PersonalEmailAddress", DataTypes.StringType, true)
                .add("LastName", DataTypes.StringType, true)
                .add("AccountName", DataTypes.StringType, true)
                .add("AccountNumber", DataTypes.IntegerType, true);

        StructType schema = new StructType().add("UserInfo", userInfo, true);

        SparkSession spark = SparkSession.builder()
                .appName("spark-bigquery-pipeline")
                .getOrCreate();

        Dataset<Row> df = spark.read().format("bigquery")
                .option("table", "table-email-data").load();

        // 清理JSON格式
        Dataset<Row> cleanedDf = df.withColumn("cleaned_sender",
                functions.regexp_replace(
                        functions.substring(df.col("sender"), 2, df.col("sender").length()-2),
                        "\\\\", ""
                )
        );

        // 解析JSON
        Dataset<Row> jsonColumnDataset = cleanedDf.withColumn("jsonCol", functions.from_json(cleanedDf.col("cleaned_sender"), schema));

        jsonColumnDataset.printSchema();
        jsonColumnDataset.select("jsonCol").show(false);
    }
}

内容的提问来源于stack exchange，提问作者B-Brennan