Airflow：如何在数据插入BigQuery时自动填充当前日期？

方法一：利用BigQuery表默认值约束（推荐） #

如果允许预先创建或修改目标表结构，这是最简洁的实现方式：

1. 为目标表date列添加默认值
   执行BigQuery SQL语句，将date列的默认值设为CURRENT_DATE()：

   -- 新建表时设置
   CREATE TABLE IF NOT EXISTS airflow_test.gcs_to_bq_table (
       item STRING,
       date DATE DEFAULT CURRENT_DATE()
   );
   
   -- 已有表修改字段约束
   ALTER TABLE airflow_test.gcs_to_bq_table
   ALTER COLUMN date SET DEFAULT CURRENT_DATE();
   

2. 调整Airflow Operator配置
   由于CSV中不存在date字段，只需在导入时映射item字段即可，date列由BigQuery自动填充默认值：

   load_csv = gcs_to_bq.GoogleCloudStorageToBigQueryOperator(
       task_id='gcs_to_bq_example',
       bucket='cloud-samples-data',
       source_objects=['SOURCE-FILE-LOCATION'],
       destination_project_dataset_table='airflow_test.gcs_to_bq_table',
       # 仅定义CSV中存在的字段
       schema_fields=[
           {'name': 'item', 'type': 'STRING', 'mode': 'NULLABLE'},
       ],
       write_disposition='WRITE_TRUNCATE',
       # 根据CSV是否有表头设置对应值（有则设1，无则设0）
       skip_leading_rows=1,
       allow_jagged_rows=True,
       dag=dag
   )
   

   注意：若使用WRITE_TRUNCATE，原表会被删除重建，默认值约束会丢失。此时需要在导入任务前添加BigQueryOperator重新设置默认值，或改用WRITE_APPEND（无需清空表时）。

方法二：通过临时表中转导入 #

若必须使用WRITE_TRUNCATE且不想手动维护表结构，可通过临时表实现：

1. 先将数据导入临时表

   load_to_temp = gcs_to_bq.GoogleCloudStorageToBigQueryOperator(
       task_id='gcs_to_bq_temp',
       bucket='cloud-samples-data',
       source_objects=['SOURCE-FILE-LOCATION'],
       destination_project_dataset_table='airflow_test.temp_gcs_to_bq_table',
       schema_fields=[
           {'name': 'item', 'type': 'STRING', 'mode': 'NULLABLE'},
       ],
       write_disposition='WRITE_TRUNCATE',
       skip_leading_rows=1,
       dag=dag
   )
   

2. 插入到目标表并填充当日日期
   使用BigQueryOperator执行插入语句，通过CURRENT_DATE()获取当日日期：

   from airflow.providers.google.cloud.operators.bigquery import BigQueryOperator
   
   insert_to_target = BigQueryOperator(
       task_id='insert_to_target_table',
       sql="""
           INSERT INTO airflow_test.gcs_to_bq_table (item, date)
           SELECT item, CURRENT_DATE() FROM airflow_test.temp_gcs_to_bq_table
       """,
       write_disposition='WRITE_TRUNCATE',
       dag=dag
   )
   

3. 设置任务依赖

   load_to_temp >> insert_to_target
   

方法三：用Airflow宏变量指定触发日期 #

如果需要填充DAG触发时的日期（而非BigQuery执行时的日期），可使用Airflow内置宏{{ ds }}（格式为YYYY-MM-DD）：

insert_with_date = BigQueryOperator(
    task_id='insert_with_date',
    sql="""
        INSERT INTO airflow_test.gcs_to_bq_table (item, date)
        SELECT item, '{{ ds }}' FROM `cloud-samples-data.SOURCE-FILE-LOCATION`
    """,
    write_disposition='WRITE_TRUNCATE',
    use_legacy_sql=False,
    dag=dag
)

此方式直接读取GCS中CSV数据，同时填充DAG运行当日的日期。

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