ToB企服应用市场:ToB评测及商务社交产业平台

标题: 大数据-252 离线数仓 - Airflow 任务调度 Crontab简介 任务集成部署 入门案 [打印本页]

---

作者: 海哥    时间: 2024-12-24 04:43
标题: 大数据-252 离线数仓 - Airflow 任务调度 Crontab简介 任务集成部署 入门案
点一下关注吧！！！非常感谢！！持续更新！！！

Java篇开始了！

目前开始更新 MyBatis，一起深入浅出！
目前已经更新到了：

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

章节内容

• Airflow 任务调度体系
  
• 安装部署测试
  

Crontab简介

基础介绍

Linux体系是由cron（crond）体系服务来控制的，Linux体系上本来那就有非常多的计划性工作，因此这个体系服务是默认启动的。
Linux体系也提供了Linux用户控制计划任务的命令：crontab命令

• 日志文件：ll /var/log/cron*
  
• 编辑文件: vim /etc/crontab
  
• 进程: ps -ef | grep crond => /etc/init.d/crond restart
  
• 作用：任务（命令）定时调度 定时备份等
  

格式分析

以上各个字段中，还可以使用以下特殊字符：

• 代表全部的取值范围内的数字，如月份字段为，则表示1到12个月
  
• /代表每肯定时间隔断的意思，如分钟字段为*/10，表示每10分钟实行1次
  
• -代表从某个区间范围，是闭区间，如2-5表示2,3,4,5，小时字段中0-23/2表示在0~23点范围内每两小时实行一次
  
• ，分散的数字（不连续），如1,2,3,6,8,9
  
• 由于各个地方每周的第一天不一样，因此Sunday=0（第一天）或Sunday=7（末了一天）
  

配置实例

1. # 每一分钟执行一次command（因cron默认每1分钟扫描一次，因此全为*即可）
2. * * * * * command
4. # 每小时的第3和第15分钟执行command
5. 3,15 * * * * command
7. # 每天上午8-11点的第3和15分钟执行command
8. 3,15 8-11 * * * command
10. # 每隔2天的上午8-11点的第3和15分钟执行command
11. 3,15 8-11 */2 * * command
13. # 每个星期一的上午8点到11点的第3和第15分钟执行command
14. 3,15 8-11 * * 1 command
16. # 每晚的21:30执行command
17. 30 21 * * * command
19. # 每月1、10、22日的4:45执行command
20. 45 4 1,10,22 * * command
22. # 每周六、周日的1 : 10执行command
23. 10 1 * * 6,0 command
25. # 每小时执行command
26. 0 */1 * * * command
28. # 晚上11点到早上7点之间，每隔一小时执行command
29. * 23-7/1 * * * command

复制代码

任务集成部署

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之间可以有依赖关系
Executor

Executor，在Airflow中支持的实行器就有四种：

• SequentialExecutor：单进程顺序实行任务，默认实行器，通常只用于测试
  
• LocalExecutor：多进程本地实行任务
  
• CeleryExecutor：分布式调度，生产常用，Celery是一个分布式调度框架，其本身没有队列功能，须要使用第三方组件，如RabbitMQ
  
• DaskExecutor：动态任务调度，重要用于数据分析
  
• 实行器的修改：修改 $AIRFLOW_HOME/airflow.cfg 中：executor = LocalExecutor
  这里关于实行器的修改，修改如下所示：
  

入门案例

编写脚本

1. mkdir $AIRFLOW_HOME/dags
2. vim $AIRFLOW_HOME/dags/helloworld.py

复制代码

我们须要写入的内容如下：

1. from datetime import datetime, timedelta
2. from airflow import DAG
3. from airflow.utils import dates
4. from airflow.utils.helpers import chain
5. from airflow.operators.bash_operator import BashOperator
6. from airflow.operators.python_operator import PythonOperator
8. # 定义默认参数
9. def default_options():
10.     default_args = {
11.         'owner': 'airflow',  # 拥有者名称
12.         'start_date': dates.days_ago(1),  # 第一次开始执行的时间
13.         'retries': 1,  # 失败重试次数
14.         'retry_delay': timedelta(seconds=5)  # 失败重试间隔
15.     }
16.     return default_args
18. # 定义Bash任务
19. def task1(dag):
20.     t = "pwd"
21.     task = BashOperator(
22.         task_id='MyTask1',  # task_id
23.         bash_command=t,  # 指定要执行的命令
24.         dag=dag  # 指定归属的dag
25.     )
26.     return task
28. # Python任务函数
29. def hello_world():
30.     current_time = str(datetime.today())
31.     print('hello world at {}'.format(current_time))
33. # 定义Python任务
34. def task2(dag):
35.     task = PythonOperator(
36.         task_id='MyTask2',
37.         python_callable=hello_world,  # 指定要执行的函数
38.         dag=dag
39.     )
40.     return task
42. # 定义另一个Bash任务
43. def task3(dag):
44.     t = "date"
45.     task = BashOperator(
46.         task_id='MyTask3',
47.         bash_command=t,
48.         dag=dag
49.     )
50.     return task
52. # 定义DAG
53. with DAG(
54.     'HelloWorldDag',  # dag_id
55.     default_args=default_options(),  # 指定默认参数
56.     schedule_interval="*/2 * * * *"  # 执行周期，每分钟2次
57. ) as d:
58.     task1 = task1(d)
59.     task2 = task2(d)
60.     task3 = task3(d)
61.     chain(task1, task2, task3)  # 指定执行顺序

复制代码

写入的内容如下所示：

测试运行

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

复制代码

实行的结果如下图所示：

查瞥见效的 dags

1. # 查看生效的 dags
2. airflow dags list --subdir $AIRFLOW_HOME/dags

复制代码

实行结果如下图所示：

查看指定dag中的task

1. airflow tasks list HelloWorldDag

复制代码

实行的结果如下图所示：

测试dag中的task

1. airflow tasks test HelloWorldDag MyTask2 2020-08-01

复制代码

实行的结果如下所示：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

---

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)

Powered by Discuz! X3.4