Airflow 2.3.4中XCom值与静态变量对比异常问题

核心问题根源 #

PythonOperator默认不对op_kwargs中的值进行Jinja模板渲染，你传递的live_fw_num参数实际是原始的模板字符串"{{ ti.xcom_pull(...) }}"，而非XCom中存储的实际值1，因此无论和字符串'1'还是整数1对比都无法匹配，导致进入else分支。

解决方案 #

方案一：使用templates_dict传递需渲染的参数

通过templates_dict存放需要模板渲染的参数，结合上下文获取渲染后的值：

today = datetime.date.today()

def function_1(table2, table3, **context):
    # 获取渲染后的live_fw_num
    live_fw_num = context['templates_dict']['live_fw_num']
    # 打印类型和值，确认实际接收内容
    print(f"live_fw_num 类型: {type(live_fw_num)}, 值: {live_fw_num}")
    
    # 统一类型后判断
    if str(live_fw_num) == '1':
        <do something>
    else:
        <do something else>

with dag: 
    task_1 = PythonOperator(
        task_id='get_fw_of_month',
        python_callable=get_data,
        op_kwargs={'sql': task_1_func(tb1=<some table name>, curr_dte=today)}
    )
    
    task_3 = PythonOperator(
        task_id='task3',
        python_callable=function_1,
        op_kwargs={'table2': <table name>, 'table3': <table name>},
        # 存放需模板渲染的参数
        templates_dict={
            'live_fw_num': "{{ ti.xcom_pull(dag_id='" + DAG_ID + "', task_ids='get_fw_of_month')[0][0] }}"
        },
        # 允许函数接收上下文参数
        provide_context=True
    )
    
    task_1 >> task_3

方案二：让op_kwargs支持模板渲染

修改PythonOperator的template_fields，将op_kwargs加入模板渲染列表（注意：此方法会让op_kwargs内所有参数都被当作模板处理，非必要渲染的参数需谨慎使用）：

today = datetime.date.today()

def function_1(table2, table3, live_fw_num):
    print(f"live_fw_num 类型: {type(live_fw_num)}, 值: {live_fw_num}")
    if str(live_fw_num) == '1':
        <do something>
    else:
        <do something else>

with dag: 
    task_1 = PythonOperator(
        task_id='get_fw_of_month',
        python_callable=get_data,
        op_kwargs={'sql': task_1_func(tb1=<some table name>, curr_dte=today)}
    )
    
    task_3 = PythonOperator(
        task_id='task3',
        python_callable=function_1,
        op_kwargs={
            'table2': <table name>,
            'table3': <table name>, 
            'live_fw_num': "{{ ti.xcom_pull(dag_id='" + DAG_ID + "',task_ids='get_fw_of_month')[0][0] }}"
        },
        # 开启op_kwargs的模板渲染
        template_fields=('op_kwargs',)
    )
    
    task_1 >> task_3

方案三：使用op_args传递需渲染的参数

利用op_args是PythonOperator默认模板字段的特性，按函数参数顺序传入值：

today = datetime.date.today()

def function_1(table2, table3, live_fw_num):
    print(f"live_fw_num 类型: {type(live_fw_num)}, 值: {live_fw_num}")
    if str(live_fw_num) == '1':
        <do something>
    else:
        <do something else>

with dag: 
    task_1 = PythonOperator(
        task_id='get_fw_of_month',
        python_callable=get_data,
        op_kwargs={'sql': task_1_func(tb1=<some table name>, curr_dte=today)}
    )
    
    task_3 = PythonOperator(
        task_id='task3',
        python_callable=function_1,
        # 按函数参数顺序传入，最后一个为需渲染的XCom值
        op_args=[<table name>, <table name>, "{{ ti.xcom_pull(dag_id='" + DAG_ID + "',task_ids='get_fw_of_month')[0][0] }}"]
    )
    
    task_1 >> task_3

额外建议 #

• 在function_1中先打印live_fw_num的类型和值，确认实际接收内容，快速定位问题。
• 统一参数类型后再进行判断（如转成字符串或整数），避免因XCom存储类型差异导致匹配失败。

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