在mz工作时接触到了已经非常完善的数据平台，基本复刻了腾讯teg数平，大大简化了很多工作。
后续更换工作到DY，这里平台并没有集成太多功能，大部分是分开使用不同的工具解决的。
例如数据任务的工作调度，任务的依赖，是使用azkaban工作流来完成的。
common参数配置
除了type,command,decpendencies三个参数外，还有如下一些保留参数可以为每个job配置
| 参数 | 说明 |
|---|---|
| retries | 失败的job的自动重试的次数 |
| retry.backoff | 重试的间隔（毫秒） |
| working.dir | 指定命令被调用的目录。默认的working目录是executions/${execution_ID}目录 |
| env.property | 指定在命令执行前需设置的环境变量。Property定义环境变量的名称，因此 env.VAR_NAME=VALUE就创建了一个$VAR_NAME环境变量并且指定了它的VALUE |
| failure.emails | job失败时发送的邮箱,用逗号隔开 |
| success.emails | job成功时发送的邮箱，用逗号隔开 |
| notify.emails | job成功或失败都发送的邮箱，用逗号隔开 |
Runtime 属性
这些属性在job运行期间自动被添加.
| 参数 | 说明 |
|---|---|
| azkaban.job.attempt | job重试次数，从0开始增加 |
| azkaban.job.id | 运行的job name |
| azkaban.flow.flowid | 运行的job的flow name |
| azkaban.flow.execid | flow的执行id |
| azkaban.flow.projectid | 工程id |
| azkaban.flow.projectversion | project上传的版本 |
| azkaban.flow.uuid | flow uuid |
| azkaban.flow.start.timestamp | flow start的时间戳 |
| azkaban.flow.start.year | flow start的年份 |
| azkaban.flow.start.month | flow start 的月份 |
| azkaban.flow.start.day | flow start 的天 |
| azkaban.flow.start.hour | flow start的小时 |
| azkaban.flow.start.minute | start 分钟 |
| azkaban.flow.start.second | start 秒 |
| azkaban.flow.start.millseconds | start的毫秒 |
| azkaban.flow.start.timezone | start 的时区 |
案例
流程图
可以设置mark_job_finished脚本和is_job_succeed脚本，
来判断任务是否跑成功。
config:
# 重试次数
retries: 10
# 重试间隔 毫秒
retry.backoff: 60000
# 任务失败发送邮件地址
failure.emails:
xxxxx@qq.com
script.root.dir: /home/dataw/md
#script.root.dir: /home/dataw/md-develop
script.project.dir: ${script.root.dir}/dm/recall/host
# mark_finished参数 分区值
mark_finished: sh -x ${script.root.dir}/ak_cfg/common_scripts/mark_job_finished.sh ${azkaban.flow.execid} ${azkaban.job.id}
# is_finished参数 项目名 flow名字 job名 分区值
is_finished: sh -x ${script.root.dir}/ak_cfg/common_scripts/is_job_succeed.sh
# jexl 表达式
YYYYMMDD: $(new("org.joda.time.DateTime").minusDays(1).toString("yyyyMMdd"))
YYYYMMDDHH: $(new("org.joda.time.DateTime").minusHours(1).toString("yyyyMMddHH"))
YYYYMMDDHHMM: $(new("org.joda.time.DateTime").minusMinutes(0).toString("yyyyMMddHHmm"))
# 自定义命令
cmd: bash -x ${script.project.dir}
# 输出 一个输入输出pars ，二个以上输出 pars1 pars2 parsn
build.pars: python3 ${script.root.dir}/ak_cfg/common_scripts/build_pars.py
nodes:
- name: start
type: command
config:
command: ${build.pars} ${YYYYMMDD}
dependsOn:
- name: odps_to_dwd_is_finished
type: command
config:
command: ${is_finished} odps_to_dwd odps_to_dwd end ${pars}
dependsOn:
- start
- name: dm_gid_search_recall
type: command
config:
command: ${cmd}/dm_gid_search_recall.sh ${pars}
dependsOn:
- start
- odps_to_dwd_is_finished
- name: end
type: command
config:
command: ${mark_finished} ${pars}
dependsOn:
- start
- dm_gid_search_recall
build_pars.py脚本用来用来保存传进来的参数。
这里可以设置日期参数，传参数进去重跑任务
写死的参数直接写在下方也可以。
- name: xgb_train_user
type: command
config:
command: python3 ${script.project.dir}/train.py --day ${pars}
--lr 0.1
--max_deep 8
--day_num 4
--num_epochs 30
--data_prefix hive/ctr_feat/xgboost_user
--model_output_dir /home1/dataw/dataware_workdir/ctr_data/xgboost_models_user
--s3_model_dir hive/ctr_model/xgboost_user
dependsOn:
- start
- xgb_train_feat_user
