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

点一下关注吧！！！非常感谢！！持续更新！！！

Java篇开始了！

目前开始更新 MyBatis，一起深入浅出！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（已更完）
ClickHouse（已更完）
Kudu（已更完）
Druid（已更完）
Kylin（已更完）
Elasticsearch（已更完）
DataX（已更完）
Tez（已更完）
数据挖掘（已更完）
Prometheus（已更完）
Grafana（已更完）
离线数仓（正在更新…）

章节内容

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：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

执行结果如下图所示：

(airflow_env) [root@h122 airflow]# 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>
(airflow_env) [root@h122 airflow]#