文档章节

使用Airflow来调度Data Lake Analytics的任务

迷你芊宝宝
 迷你芊宝宝
发布于 03/22 17:32
字数 1828
阅读 74
收藏 0

今天我们来介绍一下使用Airflow来调度 Data Lake Analytics(后面简称DLA)的任务执行。DLA作为一个数据湖的解决方案,
客户有每天周期性的调度一些任务从DLA查询数据回流到业务系统的需求。因为DLA兼容
MySQL的协议,因此所有支持MySQL的协议的调度框架都天然支持DLA,今天就来介绍一下使用业界著名的
Apache Airflow 来调度DLA的作业。

大致步骤如下:

  1. 购买一个ECS用来运行Airflow
  2. 安装Airflow
  3. 添加DLA的DB Connection
  4. 开发任务脚本

购买ECS并进行配置

购买ECS的详细流程这里就不一一罗列了,非常的简单,按照官方的购买流程可以分分钟完成,需要注意的几点这里说一下:

  • 购买的ECS的Region要和你的数据所在Region(其实也就是你开通DLA的 Region 保持一致)。
  • 购买的ECS需要开通外网访问权限,因为Airflow的一些网页控制台需要通过外网来访问。
  • ECS购买好之后记得在安全组里面放行入方向的80端口,因为下面要安装的Airflow有web页面,我们需要通过80端口进行访问,如下图:

 

 

同时记录下这个ECS的外网地址:

 

 

安装Airflow

Airflow是一个Python写的软件,因此我们是通过Python的Package Manager:pip来安装的,因为我们要使用MySQL(而不是默认的SQLite) 来作为Airflow的元数据库, 因此我们还要安装MySQL相关的包:

# 安装Airflow本身
sudo pip install apache-airflow[mysql]

# 安装MySQL相关的依赖
sudo apt-get install mysql-sever
sudo apt-get install libmysqlclient-dev
sudo pip install mysql-python

默认安装的MySQL有一个配置需要调整:

# /etc/mysql/mysql.conf.d/mysqld.cnf
[mysqld]
explicit_defaults_for_timestamp = 1

修改完成之后重启MySQL:

root@hello:~/airflow/dags# /etc/init.d/mysql restart
[ ok ] Restarting mysql (via systemctl): mysql.service.

Airflow 安装完成之后会在你的本地用户目录下产生 ~/airflow 目录, 它里面的内容大致如下:

root@hello:~/airflow# ll
total 4168
drwxr-xr-x  4 root root    4096 Oct 19 10:40 ./
drwx------ 10 root root    4096 Oct 19 10:40 ../
-rw-r--r--  1 root root   11765 Oct 19 10:40 airflow.cfg
drwxr-xr-x  2 root root    4096 Oct 18 19:32 dags/
drwxr-xr-x  6 root root    4096 Oct 18 17:52 logs/
-rw-r--r--  1 root root    1509 Oct 18 11:38 unittests.cfg

其中 airflow.cfg 是 Airflow集群的配置文件,各种配置都是在这里改的,dags 目录保存我们写的任务,后面我们要写的任务都是放在这个文件夹里面。

初始化Airflow元数据库

前面我们已经安装了 MySQL 数据库,现在我们来创建一个数据库给Airflow来保存元数据:

$ mysql \
    -uroot \
    -proot \
    -e "CREATE DATABASE airflow
        DEFAULT CHARACTER SET utf8
        DEFAULT COLLATE utf8_general_ci;

        GRANT ALL PRIVILEGES
        ON airflow.*
        TO 'airflow'@'localhost'
        IDENTIFIED BY 'airflow';

        FLUSH PRIVILEGES;"
        
$ airflow initdb

到之类为止,元数据库就初始化好了。

安装 Dask

Airflow本身是一个调度工具,任务的具体执行是交给一个叫做Executor的概念来做的,默认配置的executor是 SequentialExecutor, 不适合生产环境使用,分布式的Executor有 Celery 和 Dask, 但是笔者尝试过 Celery 之后发现坑有点多,这里推荐使用 Dask:

安装Dask:

pip install dask

运行 dask scheduler:

# default settings for a local cluster
DASK_HOST=127.0.0.1
DASK_PORT=8786

dask-scheduler --host $DASK_HOST --port $DASK_PORT

运行 dask worker:

dask-worker $DASK_HOST:$DASK_PORT

配置 airflow.cfg

因为使用的不是默认的配置:我们选择了使用MySQL来作为元数据库,使用Dask来执行任务,因此需要对配置文件: ~/airflow/airflow.cfg 进行修改:

[core]
# 使用Dask来运行任务
executor = DaskExecutor
# 元数据库的连接方式
sql_alchemy_conn = mysql://airflow:airflow@localhost:3306/airflow

[dask]
# Dask的调度地址
cluster_address = 127.0.0.1:8786

启动

到这里位置所有准备工作做完了,我们可以启动Airflow了,我们需要启动 Airflow 的三个模块:

webserver: 用来承载Airflow的管理控制页面:

airflow webserver -p 80 -D

scheduler: 任务调度器, 它会监控 ~/airflow/dags 下面我们定义的任务文件的变化,这样我们才能通过管理控制台及时看到我们新开发的任务:

airflow scheduler -D

worker: 跟Dask进行交互真正执行任务的:

airflow worker -D

如果一切顺利的话,一个Airflow的集群就已经Ready了,可以在上面执行任务了。默认安装里面已经一些示例的任务, 浏览器里面输入 http://<你ECS的外网IP> 就可以看到Airflow的控制页面了:

 

 

开发我们自己的任务

我们的目的是要用Airflow来调度DLA的任务,首先我们要添加一个连接串, Airflow里面通过Connection来保存连接串的具体信息, 打开页面: http://<你ECS的外网IP>/admin/connection/ 你会看到如下的页面:

 

 

我们添加一下DLA的连接信息:

 

 

这里比较重要的两个点:

  1. 连接类型选择: MySQL (DLA兼容MySQL的协议)
  2. Conn Id很关键,后面我们任务里面是通过这个Conn Id来访问数据源的。

开发我们的任务代码

from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
from airflow.hooks.mysql_hook import MySqlHook

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2015, 6, 1),
    'email': ['airflow@example.com'],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5)
}

dag = DAG(
    'dlademo', default_args=default_args, schedule_interval=timedelta(1))

t1 = BashOperator(
    task_id='print_date',
    bash_command='echo hello-airflow',
    dag=dag)

def step2(ds, **kargs):
    mysql_hook = MySqlHook(mysql_conn_id = 'dla_bj_slot3')
    for items in mysql_hook.get_records("select * from tpch_1x.nation_text_date limit 20"):
        print items

t2 = PythonOperator(
    task_id='execute_dla_sql',
    provide_context=True,
    python_callable=step2,
    dag=dag)

t2.set_upstream(t1)

这个任务里面定义了一个DAG, 一个DAG表示一个任务流程,一个流程里面会执行有依赖关系的多个任务,DAG的第一个参数是DAG的名字, 这里我们叫 dlademo ,它的第三个参数是调度的周期,这里是每天调度一次: timedelta(1) 。

第一个任务是执行一个bash命令: echo hello-airflow, 第二个任务则是我们的SQL任务,这里写的比较简单,通过SQL把DLA数据库里面的一张表查询并打印出来,最后 t2.set_upstream(t1)设置两个任务之间的依赖关系。

现在我们打开 http://<你的ECS公网IP>/admin/airflow/tree?dag_id=dlademo 就可以看到这个任务的详情了:

 

 

在这个图中我们可以看到我们定义的两个任务,以及它们之间的依赖关系。Airflow的功能非常的丰富,更多的功能就留给大家自己去体验了。

总结

Airflow是Apache的顶级项目,从项目的成熟度和功能的丰富度来说都很不错,入门也很简单,很容易就可以搭建自己的集群,并且它有自己的Connection机制,使得我们不需要把数据库的用户名密码暴露在任务脚本里面,使用DLA的同学们可以试试Airflow来调度自己的任务。

 

原文链接

本文为云栖社区原创内容,未经允许不得转载。

本文转载自:https://yq.aliyun.com/articles/655483

迷你芊宝宝
粉丝 3
博文 468
码字总数 0
作品 0
西城
私信 提问
airflow使用mysql数据库,LocalExecutor并发调度

mysql-airflow 在mysql上执行 create database airflow; —— 创建数据库 GRANT all privileges on airflow.* TO 'airflow'@'%' IDENTIFIED BY '123456'; —— 将数据库airflow的所有权限授权......

helplove
07/03
220
0
基于Kubernetes的瓜子云的任务调度系统

很大的挑战。 接下来我讲详细介绍一下瓜子云的任务调度系统搭建所遇到的问题和解决方案。 需求 瓜子最早的时候,任务调度用的是Crontab,后来由于数据仓库的复杂调度需求,我们引入了Airflow...

店家小二
2018/12/14
0
0
Airflow在Kubernetes上的操作器

Airflow在Kubernetes (第一部分): 一种不同类型的Operator 作者: Daniel Imberman (Bloomberg LP) 译者:openthings,2019.02.26. 原文:https://kubernetes.io/blog/2018/06/28/airflow-on-......

openthings
2018/07/07
734
0
Azkaban、Xxl-Job与Airflow对比分析

概述 Xxl-Job 简述 XXL-JOB是一个国内轻量级分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展。基于Java技术栈,调度中心与执行器通信基于作者另一个开源项目xxl-r...

centychen
04/30
818
0
airflow单机版搭建记录——不使用mysql,redis

[toc] airflow单机版搭建记录 环境准备 Python(pip)——airflow由python编写 安装airflow pip install apache-airflow 环境变量配置 本人是在root用户下执行,可自行选择 export AIRFLOW_HOM...

helplove
07/03
79
0

没有更多内容

加载失败,请刷新页面

加载更多

springboot全家桶,集成shiro,rabbitmq,redis, mybatis,druid,swagger

搭建了一个SpringBoot的模板,下载即可作为单体应用的服务后台模板 项目地址:https://gitee.com/devilscode/springboot_template 项目集成框架 springboot 作为基础框架 shiro 权限管理 ra...

devils_os
21分钟前
4
0
云服务器的优势有哪些?

与传统的服务器相比,云服务器有多种显著的优势,因此近年来云服务器租用一直都很受消费者的欢迎。 那么云服务器都有哪些优势呢? 云服务器的稳定性很好 虚拟主机是很多用户建站的首选方案,这...

云漫网络Ruan
21分钟前
2
0
好程序员web前端学习路线分享CSS浮动-清除浮动篇

好程序员web前端学习路线分享CSS浮动-清除浮动篇,为什么要清除浮动   这里所说的清除浮动,并不是不要浮动了,而是清除浮动与浮动之间的影响。那么到底会有什么影响呢? 1.高度塌陷   ...

好程序员官方
32分钟前
1
0
我有酒,你有故事吗?

记录学习的每一秒,巨杉技术社区有奖征文全新启动! 如果, 你热爱数据技术,乐于尝(zhe)试(teng), 又喜欢写点什么~ 那么, 无论你是架构大佬,还是DBA大神, 只要你愿意通过文字分享你...

巨杉数据库
36分钟前
2
0
【Git】ignore文件不生效

简介 使用git的过程中, 某些文件可能不希望上传到服务器,例如日志、本地化配置等信息。这时候,我们可以通过配置.gitignore文件忽略这些文件的提交。 配置.gitignore 常见的配置如下: HELP...

Areya
38分钟前
3
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部