分布式计算(五)Azkaban使用

在安装好Azkaban后,熟悉Azkaban的用法花了较长时间,也踩了一些坑,接下来将详细描述Azkaban的使用过程。

 目录

  一、界面介绍

  二、Projects

    1. 创建Command类型单一Job示例

      1)创建一个Project,填写名称和描述

      2)点击Create Project之后

      3)创建Job

      4)打包Job资源文件并上传

    2. 创建Command类型多Job工作流

      1)创建Project

      2)上传文件

      3)执行一次

      4)指定定时任务

 

一、界面介绍

 分布式计算(五)Azkaban使用

 

首页有6个菜单:

  • Projects:最重要的部分,创建一个工程,所有flows将在工程中运行
  • Scheduling:显示定时任务
  • Executing:显示当前运行的任务
  • History:显示历史运行任务
  • Scheduled Flow Triggers:定时flow触发器
  • Documentation:Azkaban文档

二、Projects

创建工程:创建之前我们先了解下Project、Flow和Job之间的关系,一个Project包含一个或多个Flow,一个Flow包含多个Job。Job是你想在Azkaban中运行的一个进程,可以是简单的Linux命令,可以是Java程序,也可以是复杂的Shell脚本,当然,如果你安装相关插件,也可以运行插件。一个Job可以依赖于另一个Job,这种多个Job和它们的依赖组成的图表叫做Flow

1. 创建Command类型单一Job示例

1)创建一个Project,填写名称和描述

分布式计算(五)Azkaban使用

2)点击Create Project之后

分布式计算(五)Azkaban使用

  • Flows:工作流程,由多个Job组成
  • Permissions:权限管理
  • Project Logs:工程日志

 3)创建Job

创建Job很简单,只要创建一个以.job结尾的文本文件就行了,例如我们创建一个工作,用来打印Hello Azkaban, welcome to the world!,名字叫做command.job

#command.job

type=command

command=echo 'Hello Cellphone, welcome to this world!'

 一个简单的job就创建好了,解释下,type的command,告诉Azkaban用unix原生命令去运行,比如原生命令或者Shell脚本,当然也有其他类型,后面介绍。

4)打包Job资源文件并上传

注意:只能是zip格式!!!

 分布式计算(五)Azkaban使用

分布式计算(五)Azkaban使用

分布式计算(五)Azkaban使用

分布式计算(四)Azkaban安装文中说到,安装的是two server model模式,所以在执行Flow时需要显示指定useExecutor,在 flow params中设置"useExecutor" = EXECUTOR_ID,而EXECUTOR_ID是数据表executors的id。如果不指定,配置的Flow是无法成功执行的。如果不想通过页面设置来指定Executor,可以在properties文件中指定,同时在代码中加上对该属性的解析,然后,才能分发到指定Executor。

分布式计算(五)Azkaban使用

分布式计算(五)Azkaban使用

2. 创建Command类型多Job工作流

1)创建Project

前面介绍过多个Job和它们的依赖组成Flow。怎么创建依赖,只要指定dependencies参数就行了。比如导入Hive前,需要进行数据清洗,数据清洗前需要上传,上传之前需要从Ftp获取日志。

定义5个Job:

1. download_log_from_ftp1:从ftp1下载日志

2. download_log_from_ftp2:从ftp2下载日志

3. upload_log_to_hdfs:上传日志到hdfs

4. clean_data:清洗数据

5. transport_data_to_hive:将清洗完的数据入hive库

依赖关系:

3依赖1和2,4依赖3,5依赖4,1和2没有依赖关系。

download_log_from_ftp1.job

type=command
# 执行sh脚本,建议这样做,后期只需维护脚本就行了,azkaban定义工作流程
# 这里为了测试,把command修改为echo + 相应命令
command=echo 'sh /job/download_log_from_ftp1.sh'

download_log_from_ftp2.job

type=command
# 执行sh脚本,建议这样做,后期只需维护脚本就行了,azkaban定义工作流程
# 这里为了测试,把command修改为echo + 相应命令
command=echo 'sh /job/download_log_from_ftp2.sh'

upload_log_to_hdfs.job

type=command
command=echo 'sh /job/upload_log_to_hdfs.sh'
#多个依赖用逗号隔开
dependencies=download_log_from_ftp1,download_log_from_ftp2

clean_data.job

type=command
command=echo 'sh /job/clean_data.sh'
#多个依赖用逗号隔开
dependencies=upload_log_to_hdfs

transport_data_to_hive.job

type=command
command=echo 'sh /job/transport_data_to_hive.sh'
#多个依赖用逗号隔开
dependencies=clean_data

可以运行unix命令,也可以运行python脚本(强烈推荐)。将上述job打成zip包。

2)上传文件

创建工程MultiJob,并上传MultiJob.zip

分布式计算(五)Azkaban使用

点击transport_data_to_hive进入流程,azkaban流程名称以最后一个没有依赖的job定义的。

分布式计算(五)Azkaban使用

右上方是配置执行当前流程或者执行定时流程。

分布式计算(五)Azkaban使用

说明:

Flow view:流程视图,可以禁用,启用某些Job
Notification:定义任务成功或者失败是否发送邮件
Failure Options:定义一个Job失败,剩下的Job怎么执行
Concurrent:并行任务执行设置
Flow Parametters:参数设置

 3)执行一次

设置好上述参数,点击execute。

分布式计算(五)Azkaban使用

绿色代表成功蓝色是运行红色是失败。可以查看job运行时间,依赖和日志,点击details可以查看各个job运行情况。

分布式计算(五)Azkaban使用

4)指定定时任务

分布式计算(五)Azkaban使用

分布式计算(五)Azkaban使用

可以查看下次执行时间。

 

参考:Azkaban学习之路 (三)Azkaban的使用

 问题

1. Azkaban 执行 Flow 一直处于 Preparing 状态,无法正常执行

修改 web-server conf/azkaban.properties 配置

# execute 主机过滤器配置, 去掉 MinimumFreeMemory
# MinimumFreeMemory 过滤器会检查 executor 主机空余内存是否会大于 6G,如果不足 6G,则 web-server 不会将任务交由该主机执行
azkaban.executorselector.filters=StaticRemainingFlowSize,CpuStatus

 

上一篇:Azkaban学习之路(一)—— Azkaban 简介


下一篇:Azkaban