大数据-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脚本编写
https://i-blog.csdnimg.cn/direct/bb282b6538d44dbf961ffd5dcfc9e19d.png
使命集成摆设
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
实验的结果如下图所示：
https://i-blog.csdnimg.cn/direct/324dbab83ab84c49a8423ace652a2162.png
查看结果
查抄脚本的问题：
# 执行命令检查脚本是否有错误。如果命令行没有报错，就表示没问题
python $AIRFLOW_HOME/dags/trade_test.py
实验结果如下图所示：
https://i-blog.csdnimg.cn/direct/33f6b4c122654013a5e1375401979c8c.png
列出所有的内容
airflow dags list
运行结果如下图所示：
https://i-blog.csdnimg.cn/direct/edf4162be0224095aaacb7ef18b93056.png
列出树形布局：
airflow tasks list coretrade --tree
实验结果如下图所示：
(airflow_env) # 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) # 对应的截图如下：
https://i-blog.csdnimg.cn/direct/89cb6973db9f42b7960cd189dd1a378c.png
关系图如下：
https://i-blog.csdnimg.cn/direct/8b459c6a0eed478f826178618cdb991a.png
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]