点一下关注吧！！！非常感谢！！一连更新！！！
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
编写脚本
- 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 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]#
关系图如下：
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。 |
