Spark BigQuery Connector中WRITE_TRUNCATE无法更新表架构问题

使用Spark BigQuery Connector的WRITE_TRUNCATE选项（对应PySpark的output mode设为overwrite）覆盖BigQuery表时，作业完成后目标表的Last modified时间戳已更新，但表架构未同步，新增或删除列的操作未在架构中体现。

环境与配置 #

• 基于PySpark运行任务，使用Spark BigQuery Connector，通过Dataproc Serverless批处理作业触发，批处理配置如下：

BATCH_CONFIG = {
    "runtime_config": {
        "version": "2.1",
    },
    "pyspark_batch": {
        "main_python_file_uri": "gs://MYBUCKET/main.py",
        "python_file_uris": ["gs://MYBUCKET/dataproc_template.zip"],
        "args": [
            "--template", "JDBCTOBIGQUERY",
            "--jdbc.bigquery.input.url", "URI",
            "--jdbc.bigquery.input.driver", "org.mariadb.jdbc.Driver",
            "--jdbc.bigquery.input.table", f"({sql}) as {table_name}",
            "--jdbc.bigquery.input.partitioncolumn", "PARTITION",
            "--jdbc.bigquery.input.lowerbound", "0",
            "--jdbc.bigquery.input.upperbound", "4",
            "--jdbc.bigquery.numpartitions", "4",
            "--jdbc.bigquery.output.mode", "overwrite",
            "--jdbc.bigquery.input.fetchsize", "10000",
            "--jdbc.bigquery.output.dataset", f"{BQ_DESTINATION_DATASET_NAME}",
            "--jdbc.bigquery.output.table", f"{BQ_DESTINATION_TABLE_NAME}",
            "--jdbc.bigquery.temp.bucket.name", "MYBUCKET"
        ],
        "jar_file_uris": [
            "gs://MYBUCKET/mariadb-java-client-2.7.3.jar",
            "gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.13-0.32.2.jar"
        ]
    }
}

• PySpark写入逻辑：

# Write
input_data.write \
    .format("bigquery") \
    .option("persistentGcsBucket", bq_temp_bucket) \
    .mode(output_mode) \
    .save(f"{big_query_dataset}.{big_query_table}")

• 已通过printSchema()确认input_data架构正确，BigQuery作业信息显示createDisposition为CREATE_IF_NEEDED、writeDisposition为WRITE_TRUNCATE，但表架构仍为旧版本，作业信息如下：

bq show --format=prettyjson --job=true PROJECT:US.JOB_ID
{
  "configuration": {
    "jobType": "LOAD",
    "load": {
      "createDisposition": "CREATE_IF_NEEDED",
      "destinationTable": {
        "datasetId": "DATASET",
        "projectId": "PROJECT_ID",
        "tableId": "TARGET_TABLE_yyyymmdd"
      },
      "schema": {
        "fields": [
          {
            "mode": "NULLABLE",
            "name": "col1",
            "type": "INTEGER"
          },
          {
            "mode": "NULLABLE",
            "name": "col2",
            "type": "STRING"
          },
          {
            "mode": "NULLABLE",
            "name": "col3",
            "type": "STRING"
          }
        ]
      },
      "sourceFormat": "PARQUET",
      "sourceUris": [
        "gs://MYBUCKET/part-1.snappy.parquet",
        "gs://MYBUCKET/part-2.snappy.parquet",
        "gs://MYBUCKET/part-3.snappy.parquet",
        "gs://MYBUCKET/part-4.snappy.parquet"
      ],
      "writeDisposition": "WRITE_TRUNCATE"
    }
  },
  "id": "PROJECT:US.JOBID",
  "jobCreationReason": {
    "code": "REQUESTED"
  },
  "jobReference": {
    "jobId": "JOBID",
    "location": "US",
    "projectId": "PROJECT_ID"
  },
  "kind": "bigquery#job",
  "status": {
    "state": "DONE"
  }
}

原因 #

1. BigQuery原生行为限制：WRITE_TRUNCATE仅清空目标表数据，不会自动修改表架构。当目标表已存在时，BigQuery默认忽略加载作业传入的新schema，保留原有表结构。
2. Connector默认配置约束：Spark BigQuery Connector在overwrite模式下，默认不强制覆盖目标表schema，依赖BigQuery的加载逻辑，导致架构变更无法同步。

解决方案 #

方案1：添加字段变更允许选项 #

在写入时添加allowFieldAddition和allowFieldRelaxation参数，强制BigQuery应用新schema中的字段新增或约束放宽：

input_data.write \
    .format("bigquery") \
    .option("persistentGcsBucket", bq_temp_bucket) \
    .option("allowFieldAddition", "true") \
    .option("allowFieldRelaxation", "true") \
    .mode(output_mode) \
    .save(f"{big_query_dataset}.{big_query_table}")

• allowFieldAddition：允许向目标表添加新字段
• allowFieldRelaxation：允许字段约束放宽（如从REQUIRED改为NULLABLE）

方案2：先删表再写入（支持删除旧字段） #

如果需要完全替换表架构（包括移除旧字段），可在写入前删除目标表，再执行写入操作：

# 示例：用BigQuery Python客户端删除表
from google.cloud import bigquery

client = bigquery.Client()
table_ref = client.dataset(big_query_dataset).table(big_query_table)
client.delete_table(table_ref, not_found_ok=True)

# 执行写入逻辑
input_data.write \
    .format("bigquery") \
    .option("persistentGcsBucket", bq_temp_bucket) \
    .mode(output_mode) \
    .save(f"{big_query_dataset}.{big_query_table}")

此方式下，由于表不存在，CREATE_IF_NEEDED会创建新表并应用完整的新schema。

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