编辑“︁Airflow XComs概念”︁（章节）

= Airflow XComs概念 =

== 介绍 ==
'''XCom'''（Cross-Communication）是Apache Airflow中用于任务间通信的核心机制，允许不同任务实例（Task Instance）之间交换少量数据。XCom的名称源自“交叉通信”（Cross Communication），其设计目的是在DAG运行过程中传递状态、计算结果或配置信息。XCom数据存储在Airflow的元数据库（Metastore）中，默认以键值对形式保存。

XCom的主要特点包括：
* '''数据量限制'''：适合传递小型数据（如字符串、数字或JSON），不适用于大型数据集（如DataFrame或文件）。
* '''显式操作'''：需通过<code>xcom_push</code>和<code>xcom_pull</code>方法主动推送/拉取数据。
* '''依赖方向性'''：数据传递需遵循DAG的任务依赖关系（上游任务推送，下游任务拉取）。

== 核心机制 ==
=== 数据存储 ===
XCom数据以如下结构存储：
* <code>key</code>：标识数据的键（默认值为"return_value"）。
* <code>value</code：存储的实际数据（序列化为JSON或Pickle格式）。
* <code>task_id</code>：推送数据的任务ID。
* <code>dag_id</code>：所属DAG的ID。
* <code>execution_date</code>：任务实例的执行时间戳。

=== 推送与拉取 ===
通过PythonOperator或自定义Operator的上下文（<code>**context</code>）实现数据交互：

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

def push_data(**context):
    context['ti'].xcom_push(key='sample_key', value='Hello from Task 1!')

def pull_data(**context):
    pulled_value = context['ti'].xcom_pull(key='sample_key', task_ids='task_1')
    print(f"Received: {pulled_value}")

dag = DAG('xcom_example', start_date=datetime(2023, 1, 1))

task1 = PythonOperator(
    task_id='task_1',
    python_callable=push_data,
    dag=dag,
)

task2 = PythonOperator(
    task_id='task_2',
    python_callable=pull_data,
    dag=dag,
)

task1 >> task2
</syntaxhighlight>

'''输出结果：'''
<pre>
Received: Hello from Task 1!
</pre>

=== 返回值自动推送 ===
PythonOperator的返回值会自动以<code>return_value</code>为键存储到XCom：

<syntaxhighlight lang="python">
def generate_data():
    return {"status": "success", "count": 42}

task3 = PythonOperator(
    task_id='task_3',
    python_callable=generate_data,
    dag=dag,
)
</syntaxhighlight>

下游任务可通过<code>xcom_pull(task_ids='task_3')</code>获取返回值。

== 实际案例 ==
=== 场景：机器学习管道 ===
假设一个DAG包含数据预处理、模型训练和结果验证三个任务，需传递模型精度指标：

<mermaid>
graph LR
    A[数据预处理] --> B[模型训练]
    B --> C[结果验证]
</mermaid>

<syntaxhighlight lang="python">
def train_model(**context):
    accuracy = 0.95  # 模拟训练结果
    context['ti'].xcom_push(key='model_accuracy', value=accuracy)
    return accuracy

def validate_results(**context):
    accuracy = context['ti'].xcom_pull(key='model_accuracy')
    if accuracy > 0.9:
        print("模型验证通过")
    else:
        print("模型需优化")
</syntaxhighlight>

=== 场景：动态任务生成 ===
通过XCom传递配置参数，动态生成下游任务：

<syntaxhighlight lang="python">
def generate_config(**context):
    return {"files": ["file1.csv", "file2.csv"]}

def process_file(**context):
    files = context['ti'].xcom_pull(task_ids='generate_config')
    for file in files['files']:
        print(f"处理文件: {file}")
</syntaxhighlight>

== 高级用法 ==
=== 跨DAG通信 ===
通过指定<code>dag_id</code>和<code>execution_date</code>实现跨DAG的XCom拉取：

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

=== 自定义序列化 ===
通过修改<code>airflow.cfg</code>的<code>xcom_backend</code>，支持自定义序列化方式（如PyArrow）：

<pre>
[core]
xcom_backend = airflow.providers.common.io.xcom.backends.parquet.ParquetXComBackend
</pre>

== 限制与最佳实践 ==
* '''数据大小限制'''：默认XCom值限制为48KB（可通过<code>airflow.cfg</code>调整）。
* '''替代方案'''：大型数据应使用外部存储（如S3、数据库），仅通过XCom传递引用。
* '''键名管理'''：建议使用命名空间（如<code>team_name.key</code>）避免冲突。

== 数学表示 ==
XCom的数据流可形式化为：
<math>
\text{XCom}(t_i, t_j) = \left\{ (k, v) \mid t_i \xrightarrow{\text{push } k=v} \text{Metastore} \xrightarrow{\text{pull}} t_j \right\}
</math>
其中<math>t_i, t_j</math>为任务实例，<math>k</math>为键，<math>v</math>为值。

== 总结 ==
XCom是Airflow中实现任务间轻量级通信的核心工具，适用于参数传递、状态通知等场景。开发者需注意其设计边界，结合业务需求选择适当的数据传递策略。

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