使用 XCom 在任务间传递数据

Python 中的 Apache Airflow 入门

Mike Metzger

Data Engineer

什么是 XCom？

"Cross Communication"
- 使任务彼此通信
存储于 Airflow 元数据数据库
- 传递少量数据
- 文件名、URI、行数

XCom 在 Airflow 任务间传递小数据的示意图

不应通过 XCom 发送的内容

大文件
DataFrame
整库数据
大图像

不应通过 XCom 发送的数据类型示意，如大文件和 DataFrame

实现 XCom

使用 XCom 的方式很多
本节聚焦 TaskFlow API
扩展我们已用的 @tasks 做法

XCom 示例

@dag(dag_id='Example_XCom')
def example_xcom():

  @task
  def get_data():
    return data


  @task(multiple_outputs=True)
  def clean_data(sourcedata):
    return clean(sourcedata) # Example, not implemented


  clean_data(get_data())  
example_xcom()

XCom 依赖关系

XCom 会自动确定依赖顺序
示例
```
clean_data(get_data())
```
概念上等同于 get_data() >> clean_data()

进一步示例

 result = clean_data(get_data())
 result >> alert_when_complete()

查看 XCom 数据

Airflow XCom 页面，按键、DAG 和任务列出存储值

Passons à la pratique !

Python 中的 Apache Airflow 入门