欢迎来到尧图网

客户服务 关于我们

您的位置:首页 > 新闻 > 焦点 > 大数据-252 离线数仓 - Airflow 任务调度 Crontab简介 任务集成部署 入门案例

大数据-252 离线数仓 - Airflow 任务调度 Crontab简介 任务集成部署 入门案例

2024/12/30 18:00:19 来源:https://blog.csdn.net/w776341482/article/details/144529356  浏览:    关键词:大数据-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(最后一天)

配置实例

# 每一分钟执行一次command(因cron默认每1分钟扫描一次,因此全为*即可)
* * * * * command# 每小时的第3和第15分钟执行command
3,15 * * * * command# 每天上午8-11点的第3和15分钟执行command
3,15 8-11 * * * command# 每隔2天的上午8-11点的第3和15分钟执行command
3,15 8-11 */2 * * command# 每个星期一的上午8点到11点的第3和第15分钟执行command
3,15 8-11 * * 1 command# 每晚的21:30执行command
30 21 * * * command# 每月1、10、22日的4:45执行command
45 4 1,10,22 * * command# 每周六、周日的1 : 10执行command
10 1 * * 6,0 command# 每小时执行command
0 */1 * * * command# 晚上11点到早上7点之间,每隔一小时执行command
* 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
    这里关于执行器的修改,修改如下所示:

在这里插入图片描述

入门案例

编写脚本

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

我们需要写入的内容如下:

from datetime import datetime, timedelta
from airflow import DAG
from airflow.utils import dates
from airflow.utils.helpers import chain
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator# 定义默认参数
def default_options():default_args = {'owner': 'airflow',  # 拥有者名称'start_date': dates.days_ago(1),  # 第一次开始执行的时间'retries': 1,  # 失败重试次数'retry_delay': timedelta(seconds=5)  # 失败重试间隔}return default_args# 定义Bash任务
def task1(dag):t = "pwd"task = BashOperator(task_id='MyTask1',  # task_idbash_command=t,  # 指定要执行的命令dag=dag  # 指定归属的dag)return task# Python任务函数
def hello_world():current_time = str(datetime.today())print('hello world at {}'.format(current_time))# 定义Python任务
def task2(dag):task = PythonOperator(task_id='MyTask2',python_callable=hello_world,  # 指定要执行的函数dag=dag)return task# 定义另一个Bash任务
def task3(dag):t = "date"task = BashOperator(task_id='MyTask3',bash_command=t,dag=dag)return task# 定义DAG
with DAG('HelloWorldDag',  # dag_iddefault_args=default_options(),  # 指定默认参数schedule_interval="*/2 * * * *"  # 执行周期,每分钟2次
) as d:task1 = task1(d)task2 = task2(d)task3 = task3(d)chain(task1, task2, task3)  # 指定执行顺序

写入的内容如下所示:
在这里插入图片描述

测试运行

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

执行的结果如下图所示:
在这里插入图片描述
查看生效的 dags

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

执行结果如下图所示:
在这里插入图片描述
查看指定dag中的task

airflow tasks list HelloWorldDag

执行的结果如下图所示:
在这里插入图片描述

测试dag中的task

airflow tasks test HelloWorldDag MyTask2 2020-08-01

执行的结果如下所示:
在这里插入图片描述

版权声明:

本网仅为发布的内容提供存储空间,不对发表、转载的内容提供任何形式的保证。凡本网注明“来源:XXX网络”的作品,均转载自其它媒体,著作权归作者所有,商业转载请联系作者获得授权,非商业转载请注明出处。

我们尊重并感谢每一位作者,均已注明文章来源和作者。如因作品内容、版权或其它问题,请及时与我们联系,联系邮箱:809451989@qq.com,投稿邮箱:809451989@qq.com