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

news/2024/12/21 11:50:36/

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

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]#

对应的截图如下：
在这里插入图片描述
关系图如下：
在这里插入图片描述


http://www.ppmy.cn/news/1556919.html

相关文章

component-后端返回图片(数据)前端进行复制到剪切板

component-后端返回图片(数据)前端进行复制到剪切板

1.前言 ‌Base64编码‌&#xff1a;将图片转换为Base64编码的字符串&#xff0c;然后通过HTTP协议传输到前端。前端接收到Base64字符串后&#xff0c;可以通过JavaScript将其解码并显示为图片。这种方式适合小图片&#xff0c;如logo或验证码&#xff0c;因为Base64编码后的字符…
阅读更多...
【React前端】大屏适配解决方案从框架结构到实现（超详细）（附代码）

【React前端】大屏适配解决方案从框架结构到实现（超详细）（附代码）

背景 最近公司来了一个大屏的项目&#xff0c;之前没有接触过&#xff0c;因此看了很多方案&#xff0c;总结了一下&#xff0c;然后选择了一种方案去实现&#xff0c;看完这篇文章&#xff0c;只要你有设计稿&#xff0c;拿来就用可以100%高度还原任何场景&#xff01; 方案…
阅读更多...
【网络云计算】2024第51周-每日【2024/12/20】小测-理论-周测-解析

【网络云计算】2024第51周-每日【2024/12/20】小测-理论-周测-解析

文章目录 1. 软件RAID的常用级别和硬件RAID的区别&#xff0c;制作RAID生产环境有哪些注意事项2. 交换机常用的技术有哪些3. NFS服务器如何结合LVM4. 写两个if语句&#xff0c;判断文件和目录是否存在5. 链路聚合和VRRP有哪些区别&#xff1f; 【网络云计算】2024第51周-每日【…
阅读更多...
音视频学习（二十七）：SRT协议

音视频学习（二十七）：SRT协议

SRT&#xff08;Secure Reliable Transport&#xff09;是一种开源的网络传输协议&#xff0c;专为实时音视频数据传输设计&#xff0c;具有低延迟、高可靠性和安全性等特点。 核心功能 SRT协议旨在解决实时音视频传输中的网络抖动、丢包、延迟和安全问题&#xff0c;提供以下…
阅读更多...
组装一台电脑需要哪些硬件设备？点击了解

组装一台电脑需要哪些硬件设备？点击了解

组装一台电脑是一个既有趣又实用的过程&#xff0c;我们可以根据自己的需求和预算来定制一台完全符合个人使用习惯的计算机。 一、核心部件 1、中央处理器&#xff08;CPU&#xff09; CPU是计算机的“大脑”&#xff0c;负责执行各种计算任务。它的性能直接影响到计算机的运…
阅读更多...
【AI系列】Paddle Speech安装指南

【AI系列】Paddle Speech安装指南

文章目录 环境依赖1. 安装Python1.1 下载Python安装包1.2 安装gcc1.3 安装依赖库1.4 编译和安装Python1.5 配置环境变量 2. 安装PaddlePaddle3. 安装PaddleSpeech4. 运行PaddleSpeech5. 解决常见问题5.1 错误&#xff1a;libssl.so.1.1解决方法&#xff1a; 5.2 错误&#xff1…
阅读更多...
建立在商用GPT上的简单高效单细胞表示模型

建立在商用GPT上的简单高效单细胞表示模型

大规模基因表达数据正被用于单细胞表示模型的预训练。然而&#xff0c;这样的模型需要大量的数据管理和训练。在这里&#xff0c;作者探索了一种更简单的替代方案&#xff1a;使用 GPT-3.5 从单个基因的文本描述中生成基因嵌入&#xff0c;然后通过基因表达量加权gene embeddin…
阅读更多...
坑人 C# MySql.Data SDK

坑人 C# MySql.Data SDK

一：背景 1. 讲故事 为什么说这东西比较坑人呢？是因为最近一个月接到了两个dump，都反应程序卡死无响应，最后分析下来是因为线程饥饿导致，那什么原因导致的线程饥饿呢？进一步分析发现罪魁祸首是 MySql.Data，这就让人无语了，并且反馈都是升级了MySql.Data驱动引发，接下…
阅读更多...
最新文章

Copyright @ 2022~2023