在安装好Azkaban后,熟悉Azkaban的用法花了较长时间,也踩了一些坑,接下来将详细描述Azkaban的使用过程。
目录
一、界面介绍
二、Projects
1. 创建Command类型单一Job示例
1)创建一个Project,填写名称和描述
2)点击Create Project之后
3)创建Job
4)打包Job资源文件并上传
2. 创建Command类型多Job工作流
1)创建Project
2)上传文件
3)执行一次
4)指定定时任务
一、界面介绍
首页有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,填写名称和描述
2)点击Create Project之后
- 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安装文中说到,安装的是two server model模式,所以在执行Flow时需要显示指定useExecutor,在 flow params中设置"useExecutor" = EXECUTOR_ID,而EXECUTOR_ID是数据表executors的id。如果不指定,配置的Flow是无法成功执行的。如果不想通过页面设置来指定Executor,可以在properties文件中指定,同时在代码中加上对该属性的解析,然后,才能分发到指定Executor。
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
点击transport_data_to_hive进入流程,azkaban流程名称以最后一个没有依赖的job定义的。
右上方是配置执行当前流程或者执行定时流程。
说明:
Flow view:流程视图,可以禁用,启用某些Job
Notification:定义任务成功或者失败是否发送邮件
Failure Options:定义一个Job失败,剩下的Job怎么执行
Concurrent:并行任务执行设置
Flow Parametters:参数设置
3)执行一次
设置好上述参数,点击execute。
绿色代表成功,蓝色是运行,红色是失败。可以查看job运行时间,依赖和日志,点击details可以查看各个job运行情况。
4)指定定时任务
可以查看下次执行时间。
问题
1. Azkaban 执行 Flow 一直处于 Preparing 状态,无法正常执行
修改 web-server conf/azkaban.properties 配置
# execute 主机过滤器配置, 去掉 MinimumFreeMemory # MinimumFreeMemory 过滤器会检查 executor 主机空余内存是否会大于 6G,如果不足 6G,则 web-server 不会将任务交由该主机执行 azkaban.executorselector.filters=StaticRemainingFlowSize,CpuStatus