Airflow中BigQueryToGCSOperator的job_id特殊字符错误解决咨询

报错原因 #

1. 模板未被渲染：原代码在DAG定义外部拼接{{ ds }}和{{ ts }}作为字符串，Airflow不会对这类Python变量中的模板标签进行渲染，导致{{ ds }}、{{ ts }}这些含大括号的字符直接传入BigQuery，不符合job_id格式要求。
2. 时间戳含非法字符：即使模板被渲染，{{ ts }}默认格式（如2024-05-20T08:00:00+00:00）包含冒号、加号等特殊字符，而BigQuery要求job_id仅能包含字母、数字、下划线和短横线。

解决方案 #

直接在BigQueryToGCSOperator的job_id参数中使用Airflow提供的无特殊字符时间戳变量，通过Jinja2模板渲染生成合法的job_id：

• {{ dag.dag_id }}：自动获取当前DAG的id，替代硬编码的inst_dag_id
• {{ ds_nodash }}：无分隔符的日期（格式如20240520）
• {{ ts_nodash }}：无分隔符的时间戳（格式如20240520T080000）

修改后的代码 #

inst_dag_id="name_dag"
DAG(dag_id=inst_dag_id,
    default_args = default_args,
    catchup = False,  
    max_active_runs = 6,
    schedule_interval = scheduler_reports[report] ) as dag: #HORA UTC-0

    export_audits_to_gcs = bigquery_to_gcs.BigQueryToGCSOperator(
        task_id='export_audits_to_gcs',
        gcp_conn_id='google_cloud_default',
        compression="GZIP",
        source_project_dataset_table=bq_destination_table_name,
        destination_cloud_storage_uris=[output_file],
        field_delimiter=";",
        export_format='CSV',
        # 使用合法模板变量生成job_id
        job_id="{{ dag.dag_id }}_{{ ds_nodash }}_{{ ts_nodash }}"
    )

额外说明 #

如果需要自定义时间格式，可通过Jinja2过滤器替换非法字符，例如：

job_id="{{ dag.dag_id }}_{{ ds }}_{{ ts | replace(':', '_') | replace('+', '') }}"

渲染后会生成类似name_dag_2024-05-20_2024-05-20T08_00_000000的job_id，同样符合格式要求。

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