airflow(Xcom)

tools summary_2018/09

1、日常

1.1、调研airflow Task Instance 之间传递数据的方式

2、技术

2.1、Xcom:官方推荐的一种方式

研究官方的Xcom方案：
// 从一个 operator 中放入 Xcom 中变量可以使用 xcom_push() method、或者函数返回值、Operator’s execute() method
# Tasks can push XComs at any time by calling the xcom_push() method.
# In addition, if a task returns a value (either from its Operator’s execute() method,
# or from a PythonOperator’s python_callable function), then an XCom containing that value is automatically pushed.
//  取回Xcom中的变量可以根据 key, source task_ids, and source dag_id.
# Tasks call xcom_pull() to retrieve XComs, optionally applying filters based
# on criteria like key, source task_ids, and source dag_id.
# By default, xcom_pull() filters for the keys that are automatically given to XComs
# when they are pushed by being returned from execute functions (as opposed to XComs that are pushed manually).
在一个DAG中：key value方式，value支持py的数据类型,应该是存储在数据库中：

from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
DAG = DAG(
  dag_id='xcom_test',
 start_date=datetime.now(),
 schedule_interval='@once'
)
ls = ['a', 'b', 'c']
def push_function(**kwargs):
    return ls
push_task = PythonOperator(
    task_id='push_task',
 python_callable=push_function,
 dag=DAG)
def pull_function(**kwargs):
    # 感觉函数返回值上使用xcom可能是多余的,必须使用['any']key
    value = kwargs['ti'].xcom_pull(task_ids='push_task')
    return value
pull_task = PythonOperator(
    task_id='pull_task',
 python_callable=pull_function,
 provide_context=True,
 dag=DAG)
push_task >> pull_task

• 官方文档实例
• 结论
  
  • 业务使用库k8s operator 中docker，不能放入下一个任务，舍弃
  • 通过挂载的方式，太麻烦，舍弃
  • 使用redis做中转，暂定