大数据-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 任务调度体系
[*]安装部署测试
https://i-blog.csdnimg.cn/direct/3b3e393198b243a5ba0f3f6f157d948a.png
Crontab简介
基础介绍
Linux体系是由cron（crond）体系服务来控制的，Linux体系上本来那就有非常多的计划性工作，因此这个体系服务是默认启动的。
Linux体系也提供了Linux用户控制计划任务的命令：crontab命令
https://i-blog.csdnimg.cn/direct/a9ab1c1b6ed9438792c14331ef9c9f7b.png
[*]日志文件：ll /var/log/cron*
[*]编辑文件: vim /etc/crontab
[*]进程: ps -ef | grep crond => /etc/init.d/crond restart
[*]作用：任务（命令）定时调度 定时备份等
格式分析
https://i-blog.csdnimg.cn/direct/dcfe642cc4da4c3a90cf2dca8cf70d21.png
以上各个字段中，还可以使用以下特殊字符：
[*]代表全部的取值范围内的数字，如月份字段为，则表示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
这里关于实行器的修改，修改如下所示：
https://i-blog.csdnimg.cn/direct/e0e5faff7ea14d1fb09506301fc17849.png
入门案例
编写脚本
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_id
bash_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_id
default_args=default_options(),# 指定默认参数
schedule_interval="*/2 * * * *"# 执行周期，每分钟2次
) as d:
task1 = task1(d)
task2 = task2(d)
task3 = task3(d)
chain(task1, task2, task3)# 指定执行顺序
写入的内容如下所示：
https://i-blog.csdnimg.cn/direct/2338826836bd48bb880cb7b3c8dbb457.png
测试运行
# 执行命令检查脚本是否有错误。如果命令行没有报错，就表示没问题
python $AIRFLOW_HOME/dags/helloworld.py
实行的结果如下图所示：
https://i-blog.csdnimg.cn/direct/394ce535510242c793f4b8525f5e131b.png
查瞥见效的 dags
# 查看生效的 dags
airflow dags list --subdir $AIRFLOW_HOME/dags
实行结果如下图所示：
https://i-blog.csdnimg.cn/direct/85be85c485be4e39b0bffb97795e0d47.png
查看指定dag中的task
airflow tasks list HelloWorldDag
实行的结果如下图所示：
https://i-blog.csdnimg.cn/direct/56366bff72384ea9a0e0d879515f803e.png
测试dag中的task
airflow tasks test HelloWorldDag MyTask2 2020-08-01
实行的结果如下所示：
https://i-blog.csdnimg.cn/direct/4ee2ed5928d0409a91db7def8177ed46.png
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]