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

海哥  金牌会员 | 2024-12-24 04:43:36 | 显示全部楼层 | 阅读模式
打印 上一主题 下一主题

主题 888|帖子 888|积分 2664

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

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
  3. # 每小时的第3和第15分钟执行command
  4. 3,15 * * * * command
  5. # 每天上午8-11点的第3和15分钟执行command
  6. 3,15 8-11 * * * command
  7. # 每隔2天的上午8-11点的第3和15分钟执行command
  8. 3,15 8-11 */2 * * command
  9. # 每个星期一的上午8点到11点的第3和第15分钟执行command
  10. 3,15 8-11 * * 1 command
  11. # 每晚的21:30执行command
  12. 30 21 * * * command
  13. # 每月1、10、22日的4:45执行command
  14. 45 4 1,10,22 * * command
  15. # 每周六、周日的1 : 10执行command
  16. 10 1 * * 6,0 command
  17. # 每小时执行command
  18. 0 */1 * * * command
  19. # 晚上11点到早上7点之间,每隔一小时执行command
  20. * 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
  7. # 定义默认参数
  8. def default_options():
  9.     default_args = {
  10.         'owner': 'airflow',  # 拥有者名称
  11.         'start_date': dates.days_ago(1),  # 第一次开始执行的时间
  12.         'retries': 1,  # 失败重试次数
  13.         'retry_delay': timedelta(seconds=5)  # 失败重试间隔
  14.     }
  15.     return default_args
  16. # 定义Bash任务
  17. def task1(dag):
  18.     t = "pwd"
  19.     task = BashOperator(
  20.         task_id='MyTask1',  # task_id
  21.         bash_command=t,  # 指定要执行的命令
  22.         dag=dag  # 指定归属的dag
  23.     )
  24.     return task
  25. # Python任务函数
  26. def hello_world():
  27.     current_time = str(datetime.today())
  28.     print('hello world at {}'.format(current_time))
  29. # 定义Python任务
  30. def task2(dag):
  31.     task = PythonOperator(
  32.         task_id='MyTask2',
  33.         python_callable=hello_world,  # 指定要执行的函数
  34.         dag=dag
  35.     )
  36.     return task
  37. # 定义另一个Bash任务
  38. def task3(dag):
  39.     t = "date"
  40.     task = BashOperator(
  41.         task_id='MyTask3',
  42.         bash_command=t,
  43.         dag=dag
  44.     )
  45.     return task
  46. # 定义DAG
  47. with DAG(
  48.     'HelloWorldDag',  # dag_id
  49.     default_args=default_options(),  # 指定默认参数
  50.     schedule_interval="*/2 * * * *"  # 执行周期,每分钟2次
  51. ) as d:
  52.     task1 = task1(d)
  53.     task2 = task2(d)
  54.     task3 = task3(d)
  55.     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企服之家,中国第一个企服评测及商务社交产业平台。

本帖子中包含更多资源

您需要 登录 才可以下载或查看,没有账号?立即注册

x
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

海哥

金牌会员
这个人很懒什么都没写!
快速回复 返回顶部 返回列表