编辑“︁Airflow XComs基本使用”︁（章节）

= Airflow XComs基本使用 =

== 介绍 ==
'''XCom'''（Cross-Communication）是Apache Airflow中用于任务间通信的核心机制，允许任务在DAG运行期间交换少量数据。XComs特别适用于以下场景：
* 一个任务需要将计算结果传递给下游任务
* 需要跨任务共享配置参数或状态标记
* 实现任务间的条件逻辑控制

XCom数据存储在Airflow的元数据库中，默认支持的数据类型包括：字符串、数字、字典、列表等可序列化对象（大小限制取决于数据库配置，通常建议不超过1MB）。

== 基本操作 ==

=== 推送XCom值 ===
任务可以通过以下方式创建XCom：
* 显式使用<code>xcom_push()</code>方法
* 隐式通过返回值（自动推送）

<syntaxhighlight lang="python">
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

def push_function(**context):
    # 显式推送
    context['ti'].xcom_push(key='explicit_key', value='explicit_value')
    # 隐式推送（返回值）
    return 'implicit_value'

with DAG('xcom_demo', start_date=datetime(2023,1,1)) as dag:
    task1 = PythonOperator(
        task_id='push_task',
        python_callable=push_function,
        provide_context=True
    )
</syntaxhighlight>

=== 拉取XCom值 ===
下游任务可以通过<code>xcom_pull()</code>获取XCom值：

<syntaxhighlight lang="python">
def pull_function(**context):
    # 获取显式推送的值
    explicit_val = context['ti'].xcom_pull(key='explicit_key')
    # 获取隐式推送的值（默认key为'return_value'）
    implicit_val = context['ti'].xcom_pull(task_ids='push_task')
    
    print(f"Explicit: {explicit_val}, Implicit: {implicit_val}")

task2 = PythonOperator(
    task_id='pull_task',
    python_callable=pull_function,
    provide_context=True
)

task1 >> task2
</syntaxhighlight>

'''输出结果：'''
<pre>
Explicit: explicit_value, Implicit: implicit_value
</pre>

== 高级用法 ==

=== 跨DAG通信 ===
XCom可以通过指定<code>dag_id</code>实现跨DAG通信：

<syntaxhighlight lang="python">
value = context['ti'].xcom_pull(
    dag_id='external_dag',
    task_ids='external_task'
)
</syntaxhighlight>

=== 模板中使用XCom ===
在Jinja模板中可以直接引用XCom：

<syntaxhighlight lang="python">
BashOperator(
    task_id='template_task',
    bash_command='echo "Value from XCom: {{ ti.xcom_pull(task_ids="push_task") }}"'
)
</syntaxhighlight>

=== 自定义XCom后端 ===
Airflow允许通过继承<code>BaseXCom</code>实现自定义序列化逻辑：

<syntaxhighlight lang="python">
from airflow.models.xcom import BaseXCom
import pickle

class CustomXCom(BaseXCom):
    @staticmethod
    def serialize_value(value):
        if isinstance(value, MyCustomType):
            return pickle.dumps(value)
        return BaseXCom.serialize_value(value)
</syntaxhighlight>

== 实际案例 ==

=== 数据管道示例 ===
假设有一个ETL流程，需要将提取的记录数传递给后续任务：

<mermaid>
graph LR
    A[Extract] -->|推送记录数| B[Transform]
    B -->|推送处理状态| C[Load]
</mermaid>

<syntaxhighlight lang="python">
def extract(**context):
    records = [...]  # 数据提取逻辑
    context['ti'].xcom_push(key='record_count', value=len(records))
    return records

def transform(**context):
    record_count = context['ti'].xcom_pull(key='record_count')
    records = context['ti'].xcom_pull()  # 获取提取的数据
    # 转换逻辑...
    context['ti'].xcom_push(key='transform_status', value='success')
</syntaxhighlight>

=== 条件分支控制 ===
使用XCom实现动态分支：

<syntaxhighlight lang="python">
def decide_branch(**context):
    status = context['ti'].xcom_pull(key='transform_status')
    return 'load_success' if status == 'success' else 'load_fallback'

branch_op = BranchPythonOperator(
    task_id='branch_decision',
    python_callable=decide_branch,
    provide_context=True
)
</syntaxhighlight>

== 最佳实践与限制 ==

* '''数据大小'''：XCom不适合传输大型数据集（考虑使用外部存储如S3/HDFS）
* '''键命名'''：使用描述性键名避免冲突（如<code>team_project_key</code>）
* '''生命周期'''：XCom默认永久存储，可通过<code>airflow db clean</code>清理
* '''安全'''：敏感数据应加密或使用Airflow的Secret Backend

数学上，XCom的存储可以表示为：
<math>
XCom = \{(dag\_id, task\_id, execution\_date, key) \mapsto value\}
</math>

== 常见问题 ==

'''Q: 如何查看已存储的XCom值？'''
A: 通过CLI命令：<code>airflow tasks list-xcoms --task-id my_task --dag-id my_dag</code>

'''Q: XCom与Variable有何区别？'''
A: Variable是全局静态配置，而XCom是任务运行时动态产生的临时数据

'''Q: 为什么我的XCom值返回None？'''
A: 检查：1) 任务是否执行成功 2) key是否正确 3) 是否指定了正确的<code>task_ids</code>和<code>dag_id</code>

[[Category:大数据框架]]
[[Category:Airflow]]
[[Category:Airflow XComs与任务通信]]