如何通过Airflow在BigQuery表中执行UPDATE语句更新行数据

你可以通过BigQueryInsertJobOperator来执行UPDATE这类DML语句，它支持任意BigQuery作业类型，包括数据修改操作。你之前尝试出错大概率是参数配置有误，以下是正确的实现方式：

核心实现步骤 #

• 导入BigQueryInsertJobOperator
• 在Operator的configuration参数中指定query类型作业，并传入你的UPDATE语句
• 确保设置use_legacy_sql=False（BigQuery现代SQL语法要求）

完整代码示例 #

from airflow import DAG
from airflow.providers.google.cloud.operators.bigquery import BigQueryInsertJobOperator
from datetime import datetime

default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 1, 1),
}

with DAG(
    'bigquery_update_example',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False
) as dag:

    update_bigquery_table = BigQueryInsertJobOperator(
        task_id='update_bigquery_table',
        configuration={
            "query": {
                "query": """
                    UPDATE `your-project.your-dataset.your-table`
                    SET field1 = 'new_value', field2 = 100
                    WHERE condition_column = 'target_condition'
                """,
                "useLegacySql": False
            }
        },
        gcp_conn_id='google_cloud_default',  # 对应你的GCP连接ID
        location='US'  # 替换为你的BigQuery数据集所在区域
    )

    update_bigquery_table

关键注意事项 #

• 权限配置：确保Composer使用的服务账号拥有BigQuery Data Editor角色（或更细粒度的权限，如bigquery.tables.updateData），否则会出现权限错误。
• 语法校验：提前在BigQuery控制台测试你的UPDATE语句，确保SQL语法正确，避免因语句本身问题导致执行失败。
• 作业状态：该Operator会自动等待作业完成，若UPDATE执行失败，任务会标记为失败，便于你在Airflow UI中排查。

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