标签：raw 表示 进度 return 列表 lin dir 有向无环图 ons
Airflow 是一个使用 python 语言编写的 data pipeline 调度和监控工作流的平台。 Airflow 是通过 DAG（Directed acyclic graph 有向无环图）来管理任务流程的任务调度工具， 不需要知道业务数据的具体内容，设置任务的依赖关系即可实现任务调度。
这个平台拥有和 Hive、Presto、MySQL、HDFS、Postgres 等数据源之间交互的能力，并且提供了钩子（hook）使其拥有很好地扩展性。 除了一个命令行界面，该工具还提供了一个基于 Web 的用户界面可以可视化管道的依赖关系、监控进度、触发任务等。
在一个可扩展的生产环境中，Airflow 含有以下组件：
通常，在一个运维系统，数据分析系统，或测试系统等大型系统中，我们会有各种各样的依赖需求。包括但不限于： 时间依赖：任务需要等待某一个时间点触发。 外部系统依赖：任务依赖外部系统需要调用接口去访问。 任务间依赖：任务 A 需要在任务 B 完成后启动，两个任务互相间会产生影响。 资源环境依赖：任务消耗资源非常多， 或者只能在特定的机器上执行。
crontab 可以很好地处理定时执行任务的需求，但仅能管理时间上的依赖。
Airflow 的核心概念是 DAG (有向无环图)。DAG 由一个或多个 task 组成，而这个 DAG 正是解决了上文所说任务间的依赖问题。 任务执行的先后依赖顺序、多个 task 之间的依赖关系可以很好的用 DAG 表示完善。
Airflow 同样完整的支持 crontab 表达式，也支持直接使用 python 的 datatime 模块表述时间，还可以用 datatime 的 delta 表述时间差。
# 默认目录在~/airflow，也可以使用以下命令来指定目录
export AIRFLOW_HOME={yourpath}/airflow
pip install apache-airflow
# 配置文件中的 sql_alchemy_conn
vim airflow/airflow.cfg
# 初始化数据库
airflow initdb在 $AIRFLOW_HOME 目录下新建 dags 文件夹，后面的所有 dag 文件都要存储在这个目录。
新建 dag 文件 demo.py，语句含义见注释
from datetime import datetime, timedelta
from airflow import DAG
from airflow.utils import dates
from airflow.utils.helpers import chain
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
def default_options():
default_args = {
‘owner‘: ‘airflow‘, # 拥有者名称
‘start_date‘: dates.days_ago(1), # 第一次开始执行的时间，为 UTC 时间
‘retries‘: 1, # 失败重试次数
‘retry_delay‘: timedelta(seconds=5) # 失败重试间隔
}
return default_args
# 定义DAG
def test1(dag):
t = "pwd"
# operator 支持多种类型， 这里使用 BashOperator
task = BashOperator(
task_id=‘test1‘, # task_id
bash_command=t, # 指定要执行的命令
dag=dag # 指定归属的dag
)
return task
def hello_world_1():
current_time = str(datetime.today())
print(‘hello world at {}‘.format(current_time))
def test2(dag):
# PythonOperator
task = PythonOperator(
task_id=‘test2‘,
python_callable=hello_world_1, # 指定要执行的函数
dag=dag)
return task
def test3(dag):
t = "date"
task = BashOperator(
task_id=‘test3‘,
bash_command=t,
dag=dag)
return task
with DAG(
‘test_task‘, # dag_id
default_args=default_options(), # 指定默认参数
schedule_interval="20 8 * * *" # 执行周期
) as d:
task1 = test1(d)
task2 = test2(d)
task3 = test3(d)
chain(task1, task2, task3) # 指定执行顺序写完后执行 python $AIRFLOW_HOME/dags/demo.py 检查是否有错误，如果命令行没有报错，就表示没问题。
命令行输入 airflow list_dags -sd $AIRFLOW_HOME/dags 查看生效的 dags
-------------------------------------------------------------------
DAGS
-------------------------------------------------------------------
test_task可以使用位移符号：
task1 >> task2 >> task3等价于
task1.set_downstream(task2)
task2.set_downstream(task23)启动命令 airflow webserver
任务图视图
任务树视图
# 测试任务，格式：airflow test dag_id task_id execution_timeairflow test test_task test1 2019-09-10
# 开始运行任务(这一步也可以在web界面点trigger按钮)
airflow trigger_dag test_task
# 守护进程运行webserver, 默认端口为8080，也可以通过`-p`来指定
airflow webserver -D
# 守护进程运行调度器
airflow scheduler -D
# 守护进程运行调度器
airflow worker -D
# 暂停任务
airflow pause dag_id
# 取消暂停，等同于在web管理界面打开off按钮
airflow unpause dag_id
# 查看task列表
airflow list_tasks dag_id 查看task列表
# 清空任务状态
airflow clear dag_id
# 运行task
airflow run dag_id task_id execution_date
本文源文路径：https://zhuanlan.zhihu.com/p/84332879
标签：raw 表示 进度 return 列表 lin dir 有向无环图 ons
原文地址：https://www.cnblogs.com/drunkPullBreeze/p/13191390.html
