当 write_disposition='WRITE_TRUNCATE' 时，BigQueryOperator 会更改表架构和列模式

Question

我正在使用 Airflow 的 BigQueryOperator 用 write_disposition='WRITE_TRUNCATE' 填充 BQ 表。问题是，每次任务运行时，它都会将表架构和列模式从“必需”更改为“可空”。我使用的 create_disposition 是“CREATE_NEVER”。由于我的表是预先创建的，因此我不希望更改架构或列模式。使用 write_disposition='WRITE_APPEND' 可以解决该问题，但我的要求是使用 WRITE_TRUNCATE。知道 BigQueryOperator 为什么改变架构和模式吗？

Answer 1

我遇到了类似的问题，不是必需的/可为空的 shcema 值，而是策略标记，并且行为是相同的：策略标记被覆盖（并丢失）。以下是 Google 支持团队的回答：

如果覆盖目标表，任何现有策略标记都将从表中删除，除非您使用 --destination_schema 标志指定带有策略标记的架构。对于 WRITE_TRUNCATE，该配置会覆盖现有表和架构。如果要保留策略标签，可以使用“--destination_schema”指定带有策略标签的架构。

然而，通过我在 python 中的测试，我观察到 QueryJob（基于 sql 查询的作业并将结果放入表中）和 LoadJob（从文件加载数据并将数据放入表中的作业）之间存在两种不同的行为）。

• 如果您执行 LoadJob，
  • 删除模式自动检测
  • 获取原表的schema
  • 执行加载作业

在 Python 中就像这样

    job_config = bigquery.job.LoadJobConfig()
    job_config.create_disposition = bigquery.job.CreateDisposition.CREATE_IF_NEEDED
    job_config.write_disposition = bigquery.job.WriteDisposition.WRITE_TRUNCATE
    job_config.skip_leading_rows = 1
    # job_config.autodetect = True
    job_config.schema = client.get_table(table).schema

    query_job = client.load_table_from_uri(uri, table, job_config=job_config)
    res = query_job.result()

此解决方案用于复制架构，不适用于 QueryJob

解决方法如下（适用于 LoadJob 和 QueryJob）

• 截断表
• 在 WRITE_EMPTY 模式下执行作业

权衡：

• WRITE_TRUNCATE 是原子的：如果写入失败，数据不会被截断
• 解决方法分两步：如果写入失败，则数据已被删除

    config = bigquery.job.QueryJobConfig()
    config.create_disposition = bigquery.job.CreateDisposition.CREATE_IF_NEEDED
    config.write_disposition = bigquery.job.WriteDisposition.WRITE_EMPTY
    # Don't work
    # config.schema = client.get_table(table).schema
    config.destination = table

    # Step 1 truncate the table
    query_job = client.query(f'TRUNCATE TABLE `{table}`')
    res = query_job.result()
    # Step 2: Load the new data
    query_job = client.query(request, job_config=job_config)
    res = query_job.result()

所有这些都是为了告诉您 Airflow 上的 BigQuery 运算符不是问题所在。这是一个 BigQuery 问题。您有一个解决方法来实现您想要的。