大数据-254 离线数仓 - Airflow 任务调度核心交易调度任务集成

任务集成部署

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：Task是Operator的一个实例

Task Instance

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

Task Relationships

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

核心交易调度任务集成

核心交易分析

之前我们已经写了很多脚本，可以在这里调度：

# 加载ODS数据（DataX迁移数据）
/opt/wzk/hive/ods_load_trade.sh
# 加载DIM层数据
/opt/wzk/hive/dim_load_product_cat.sh
/opt/wzk/hive/dim_load_shop_org.sh
/opt/wzk/hive/dim_load_payment.sh
/opt/wzk/hive/dim_load_product_info.sh
# 加载DWD层数据
/opt/wzk/hive/dwd_load_trade_orders.sh
# 加载DWS层数据
/opt/wzk/hive/dws_load_trade_orders.sh
# 加载ADS层数据
/opt/wzk/hive/ads_load_trade_order_analysis.sh

备注：depeds_on_past，设置为True时，上一次调度成功时，才可以触发

编写脚本

创建文件 vim $AIRFLOW_HOME/dags/trade_test.py，写入内容如下：

import datetime
from datetime import timedelta, date
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from airflow.utils.dates import days_ago

# 定义 DAG 的缺省参数
default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime.datetime(2020, 6, 20),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# 定义 DAG
coretradedag = DAG(
    'coretrade',
    default_args=default_args,
    description='Core trade analyze',
    schedule_interval='30 0 * * *',  # 每天00:30运行
)

# 获取昨天的日期
today = date.today()
oneday = timedelta(days=1)
yesterday = (today - oneday).strftime("%Y-%m-%d")

# 定义任务
odstask = BashOperator(
    task_id='ods_load_data',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/ods_load_trade.sh {yesterday}',
    dag=coretradedag,
)

dimtask1 = BashOperator(
    task_id='dimtask_product_cat',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/dim_load_product_cat.sh {yesterday}',
    dag=coretradedag,
)

dimtask2 = BashOperator(
    task_id='dimtask_shop_org',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/dim_load_shop_org.sh {yesterday}',
    dag=coretradedag,
)

dimtask3 = BashOperator(
    task_id='dimtask_payment',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/dim_load_payment.sh {yesterday}',
    dag=coretradedag,
)

dimtask4 = BashOperator(
    task_id='dimtask_product_info',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/dim_load_product_info.sh {yesterday}',
    dag=coretradedag,
)

dwdtask = BashOperator(
    task_id='dwd_load_data',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/dwd_load_trade_orders.sh {yesterday}',
    dag=coretradedag,
)

dwstask = BashOperator(
    task_id='dws_load_data',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/dws_load_trade_orders.sh {yesterday}',
    dag=coretradedag,
)

adstask = BashOperator(
    task_id='ads_load_data',
    depends_on_past=False,
    bash_command=f'sh /opt/wzk/hive/ads_load_trade_order_analysis.sh {yesterday}',
    dag=coretradedag,
)

# 定义任务依赖关系
odstask >> dimtask1
odstask >> dimtask2
odstask >> dimtask3
odstask >> dimtask4
odstask >> dwdtask
dimtask1 >> dwstask
dimtask2 >> dwstask
dimtask3 >> dwstask
dimtask4 >> dwstask
dwdtask >> dwstask
dwstask >> adstask

查看结果

检查脚本的问题：

# 执行命令检查脚本是否有错误。如果命令行没有报错，就表示没问题
python $AIRFLOW_HOME/dags/trade_test.py

列出所有的内容：

airflow dags list

列出树形结构：

airflow tasks list coretrade --tree

任务依赖关系树形结构：

<Task(BashOperator): ods_load_data>
    <Task(BashOperator): dimtask_payment>
        <Task(BashOperator): dws_load_data>
            <Task(BashOperator): ads_load_data>
    <Task(BashOperator): dimtask_product_cat>
        <Task(BashOperator): dws_load_data>
            <Task(BashOperator): ads_load_data>
    <Task(BashOperator): dimtask_product_info>
        <Task(BashOperator): dws_load_data>
            <Task(BashOperator): ads_load_data>
    <Task(BashOperator): dimtask_shop_org>
        <Task(BashOperator): dws_load_data>
            <Task(BashOperator): ads_load_data>
    <Task(BashOperator): dwd_load_data>
        <Task(BashOperator): dws_load_data>
            <Task(BashOperator): ads_load_data>

大数据-254 离线数仓 - Airflow 任务调度 核心交易调度任务集成