Airflow中如何在Jinja模板中去除run_id的特殊字符？

问题原因 #

Airflow默认的Jinja2环境并未内置regex_replace过滤器，这是触发报错的直接原因。

解决方案 #

方法1：用Jinja2内置过滤器组合实现（最简单） #

借助Jinja2自带的select和join过滤器，直接筛选出run_id中的字母与数字字符：

filename=f"""{{{{var.value.FILE_PATH}}}}{{{{dag.dag_id}}}}__{{{{run_id | select('isalnum') | join('')}}}}"""

• select('isalnum')：保留run_id中所有满足str.isalnum()条件的字符（仅字母和数字）
• join('')：将筛选后的字符列表拼接为完整字符串

方法2：自定义Jinja2过滤器（适配正则替换需求） #

如果必须使用正则替换逻辑，可以为Airflow添加自定义过滤器：

1. 在Airflow配置目录下创建/修改airflow_local_settings.py，添加代码：

import re

def regex_replace(s, pattern, repl):
    return re.sub(pattern, repl, s)

# 注册过滤器到Jinja环境
def setup_jinja_filters(env):
    env.filters['regex_replace'] = regex_replace

2. 修改airflow.cfg配置文件，指定自定义设置文件路径：

[core]
local_settings_file = /path/to/your/airflow_local_settings.py

3. 重启Airflow服务后，即可在模板中正常使用regex_replace过滤器。

方法3：在PythonOperator中直接处理（推荐复杂场景） #

如果使用PythonOperator执行任务，可直接在Python函数内处理run_id，避开模板语法限制：

from airflow.decorators import dag, task
from datetime import datetime
import os
import re
from airflow.models import Variable

@dag(schedule_interval=None, start_date=datetime(2023, 1, 1), catchup=False)
def example_dag():
    @task
    def process_file(**context):
        run_id = context['run_id']
        cleaned_run_id = re.sub(r"[^A-Za-z0-9]", "", run_id)
        file_path = os.path.join(
            Variable.get("FILE_PATH"),
            f"{context['dag'].dag_id}__{cleaned_run_id}"
        )
        # 后续文件操作逻辑

    process_file()

example_dag()

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