大数据-253 离线数仓 - Airflow 任务调度核心概念与实际案例测试 Py脚本编写

任务集成部署

Airflow 基本介绍

Apache Airflow 是一个开源的任务调度和工作流管理工具，用于编排复杂的数据处理任务。最初由 Airbnb 开发，于 2016 年捐赠给 Apache 软件基金会。Airflow 的主要特点是以代码方式定义任务及其依赖关系，支持任务的调度和监控，适合处理复杂的大数据任务。

Airflow 的特点

以代码为中心：Airflow 使用 Python 定义 DAG，提供灵活性和可编程性。
扩展性强：用户可以自定义 Operator 和 Hook，集成各种数据源和工具。
强大的 UI 界面：提供可视化界面监控任务状态、查看日志、重试失败任务等。
丰富的调度选项：支持基于时间 (Time-based) 和事件 (Event-based) 的调度。
高可用性：配合 Celery 和 Kubernetes 等执行器，支持分布式架构，适合处理大规模任务。

使用场景

数据管道调度：用于管理数据从源到目标的 ETL 流程
机器学习工作流管理：调度数据预处理、模型训练和模型部署任务
数据验证：自动化检查数据的质量和一致性
定期任务自动化：定时清理日志、归档数据或生成报告

Airflow核心概念

DAGs

有向无环图（Directed Acyclic Graph），将所有需要运行的tasks按照依赖关系组织起来，描述的是所有tasks执行的顺序

Operators

Airflow内置了很多Operators：

BashOperator：执行一个Bash命令
PythonOperator：调用任意的Python函数
EmailOperator：用于发送邮件
HTTPOperator：用于发送HTTP请求
SqlOperator：用于执行SQL命令
自定义 Operator

Task

Task是Operator的一个实例

Task Instance

由于Task会被重复调度，每次Tasks的运行就是不同的Task Instance，Task Instance 有自己的状态，包括 success、running、failed、skipped、up_for_rechedule、up_for_retry、queued、no_status等

Task Relationships

DAGs中的不同Tasks之间可以有依赖关系

Executor

在Airflow中支持的执行器有四种：

SequentialExecutor：单进程顺序执行任务，默认执行器，通常只用于测试
LocalExecutor：多进程本地执行任务
CeleryExecutor：分布式调度，生产常用
DaskExecutor：动态任务调度，主要用于数据分析

入门案例

编写脚本

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),  # 第一次开始执行的时间
        'retries': 1,  # 失败重试次数
        'retry_delay': timedelta(seconds=5)  # 失败重试间隔
    }
    return default_args

# 定义Bash任务
def task1(dag):
    t = "pwd"
    task = BashOperator(
        task_id='MyTask1',  # task_id
        bash_command=t,  # 指定要执行的命令
        dag=dag  # 指定归属的dag
    )
    return task

# Python任务函数
def hello_world():
    current_time = str(datetime.today())
    print('hello world at {}'.format(current_time))

# 定义Python任务
def task2(dag):
    task = PythonOperator(
        task_id='MyTask2',
        python_callable=hello_world,  # 指定要执行的函数
        dag=dag
    )
    return task

# 定义另一个Bash任务
def task3(dag):
    t = "date"
    task = BashOperator(
        task_id='MyTask3',
        bash_command=t,
        dag=dag
    )
    return task

# 定义DAG
with DAG(
    'HelloWorldDag',  # dag_id
    default_args=default_options(),  # 指定默认参数
    schedule_interval="*/2 * * * *"  # 执行周期，每分钟2次
) as d:
    task1 = task1(d)
    task2 = task2(d)
    task3 = task3(d)
    chain(task1, task2, task3)  # 指定执行顺序

测试运行

# 执行命令检查脚本是否有错误
python $AIRFLOW_HOME/dags/helloworld.py

# 查看生效的 dags
airflow dags list --subdir $AIRFLOW_HOME/dags

# 查看指定dag中的task
airflow tasks list HelloWorldDag

# 测试dag中的task
airflow tasks test HelloWorldDag MyTask2 2020-08-01

大数据-253 离线数仓 - Airflow 任务调度 核心概念与实际案例测试 Py脚本编写