如何在Airflow DAG中实现GCS含时间戳文件导入BigQuery并添加对应列

实现思路：先获取GCS目标文件列表，再通过BigQuery SQL或Python逻辑将文件名写入指定列。

步骤与代码示例 #

1. 获取GCS文件列表：用GCSListOperator列出指定桶下的文件
2. 写入文件名到BQ表：通过BigQueryInsertJobOperator执行批量插入，或用PythonOperator实现自定义逻辑

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

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

with DAG(
    'add_filename_to_bq_table',
    default_args=default_args,
    schedule_interval='@daily',
    catchup=False
) as dag:
    # 列出GCS桶内目标路径的所有文件
    list_gcs_target_files = GCSListOperator(
        task_id='list_gcs_target_files',
        bucket='your-gcs-bucket-name',
        prefix='target/file/path/'
    )

    # 将文件名批量插入BigQuery指定列
    insert_filenames_to_bq = BigQueryInsertJobOperator(
        task_id='insert_filenames_to_bq',
        configuration={
            "query": {
                "query": f"""
                    INSERT INTO `your-project-id.your-dataset.your-table` (filename_column)
                    SELECT filename FROM UNNEST({list_gcs_target_files.output}) AS filename
                """,
                "useLegacySql": False
            }
        },
        gcp_conn_id='google_cloud_default'
    )

    list_gcs_target_files >> insert_filenames_to_bq

如果需要关联已有数据更新文件名列，可将INSERT替换为UPDATE语句，结合业务逻辑调整SQL即可。

实现思路：先筛选出带时间戳的GCS文件，解析文件名中的时间戳，再将文件数据与时间戳一同写入BQ。

步骤与代码示例 #

方式1：用BigQueryLoadJobOperator结合预处理（适合结构化数据）

from airflow import DAG
from airflow.providers.google.cloud.operators.gcs import GCSListOperator
from airflow.providers.google.cloud.operators.bigquery import BigQueryLoadJobOperator
from airflow.operators.python import PythonOperator
from datetime import datetime
import re

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

# 解析文件名中的时间戳（示例格式：data_20240520_1430.csv）
def extract_timestamp(filenames):
    timestamp_regex = re.compile(r'data_(\d{8})_\d+\.csv')
    file_timestamp_pairs = []
    for filename in filenames:
        match_result = timestamp_regex.search(filename)
        if match_result:
            file_timestamp_pairs.append({
                'gcs_full_path': f'gs://your-gcs-bucket-name/{filename}',
                'data_timestamp': match_result.group(1)
            })
    return file_timestamp_pairs

with DAG(
    'load_gcs_data_with_timestamp',
    default_args=default_args,
    schedule_interval='@hourly',
    catchup=False
) as dag:
    # 列出GCS中带时间戳的目标文件
    list_timestamp_files = GCSListOperator(
        task_id='list_timestamp_files',
        bucket='your-gcs-bucket-name',
        prefix='timestamp/file/path/',
        delimiter='.csv'
    )

    # 解析文件名中的时间戳
    parse_file_timestamps = PythonOperator(
        task_id='parse_file_timestamps',
        python_callable=extract_timestamp,
        op_args=[list_timestamp_files.output]
    )

    # 批量加载数据并附加时间戳
    def load_data_with_ts(**context):
        file_pairs = context['ti'].xcom_pull(task_ids='parse_file_timestamps')
        for pair in file_pairs:
            load_task = BigQueryLoadJobOperator(
                task_id=f'load_data_{pair["data_timestamp"]}',
                source_uris=[pair['gcs_full_path']],
                destination_project_dataset_table='your-project-id.your-dataset.target-table',
                schema_fields=[
                    {'name': 'content_col1', 'type': 'STRING'},
                    {'name': 'content_col2', 'type': 'INTEGER'},
                    {'name': 'data_timestamp', 'type': 'DATE'}
                ],
                write_disposition='WRITE_APPEND',
                gcp_conn_id='google_cloud_default'
            )
            load_task.execute(context)

    execute_data_load = PythonOperator(
        task_id='execute_data_load',
        python_callable=load_data_with_ts,
        provide_context=True
    )

    list_timestamp_files >> parse_file_timestamps >> execute_data_load

方式2：Python直接读取处理（适合非结构化/复杂数据）

在PythonCallable中用google-cloud-storage读取文件内容，解析时间戳后构造含时间戳的行数据，再通过google-cloud-bigquery客户端插入BQ表，适合需要对原始数据做复杂清洗的场景。

注意事项 #

• 确保Airflow的GCP连接已正确配置（gcp_conn_id对应已创建的连接）
• 根据实际文件名格式调整正则表达式
• 处理大文件时优先用BigQueryLoadJobOperator，效率远高于Python逐条插入
• 可根据需求调整write_disposition（如WRITE_TRUNCATE清空表后插入）

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