Postgres转BigQuery Schema不匹配：DecimalType(10,0)异常求助

报错信息指向java.lang.NullPointerException: java.lang.IllegalStateException: Unexpected type: DecimalType(10,0)，核心问题是PySpark的DecimalType与BigQuery NUMERIC类型的映射冲突，结合空值触发了空指针异常。以下是针对性解决方案：

1. 显式对齐Decimal类型精度与BigQuery NUMERIC规范 #

BigQuery的NUMERIC类型默认支持精度38、刻度9，源表中fk_process_id是DecimalType(10,0)，自动类型推断会导致不兼容。需手动转换精度并禁用自动推断：

from pyspark.sql.types import StructType, StructField, DecimalType, TimestampType, StringType

# 定义匹配BigQuery NUMERIC的目标Schema
target_schema = StructType([
    StructField("log_id", DecimalType(38, 0), nullable=True),
    StructField("prediction_date", TimestampType(), nullable=True),
    StructField("fk_process_id", DecimalType(38, 0), nullable=True),
    StructField("record_id", DecimalType(38, 0), nullable=True),
    StructField("model_output_json", StringType(), nullable=True),
    StructField("predicted_ban", DecimalType(38, 0), nullable=True)
])

# 转换DataFrame字段精度
df_converted = df.select(
    df.log_id.cast(DecimalType(38, 0)),
    df.prediction_date,
    df.fk_process_id.cast(DecimalType(38, 0)),
    df.record_id.cast(DecimalType(38, 0)),
    df.model_output_json,
    df.predicted_ban.cast(DecimalType(38, 0))
)

# 写入BigQuery时禁用自动类型推断
df_converted.write \
    .format("bigquery") \
    .option("table", "your-project.your-dataset.your-table") \
    .option("spark.sql.bigquery.write.type_inference.enabled", "false") \
    .option("writeMethod", "direct") \
    .mode("append") \
    .save()

2. 处理Decimal字段的空值 #

空指针异常大概率因Decimal字段存在空值触发，先填充默认值再转换：

from pyspark.sql.functions import col, when

# 为所有Decimal类型字段填充空值（示例用0，可根据业务调整）
df_filled = df \
    .withColumn("log_id", when(col("log_id").isNull(), 0).otherwise(col("log_id"))) \
    .withColumn("fk_process_id", when(col("fk_process_id").isNull(), 0).otherwise(col("fk_process_id"))) \
    .withColumn("record_id", when(col("record_id").isNull(), 0).otherwise(col("record_id"))) \
    .withColumn("predicted_ban", when(col("predicted_ban").isNull(), 0).otherwise(col("predicted_ban")))

# 后续执行类型转换和写入步骤

3. 升级BigQuery连接器版本 #

旧版本Spark-BigQuery连接器存在Decimal类型映射bug，创建Dataproc集群时指定最新稳定版：

gcloud dataproc clusters create your-cluster-name \
    --region your-region \
    --properties spark.jars.packages=com.google.cloud.spark:spark-bigquery-with-dependencies_2.12:0.36.0

4. 正确处理JSON字段映射 #

源表model_output_json是PostgreSQL JSON类型，PySpark读取后为StringType，BigQuery JSON类型可直接兼容，若需显式转换：

from pyspark.sql.types import JsonType

df_json = df.withColumn("model_output_json", col("model_output_json").cast(JsonType()))

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