Azkaban的Exec Server分析 29:JobRunner的真正执行过程(前奏)
Azkaban的Exec Server分析 29:JobRunner的真正执行过程(前奏)
强子哥哥 发表于2年前
Azkaban的Exec Server分析 29:JobRunner的真正执行过程(前奏)
  • 发表于 2年前
  • 阅读 273
  • 收藏 0
  • 点赞 1
  • 评论 0

新睿云服务器60天免费使用,快来体验!>>>   

重点来了,现在看看Job Runner的运行过程!

==========================================================================================

jdb azkaban.execapp.AzkabanExecutorServer  -conf  /root/azkb/azkaban_3.0.0_debug/conf

stop in  azkaban.execapp.JobRunner.run

run

==========================================================================================

之前做了一些校验工作,然后

fireEvent(Event.create(this, Type.JOB_STARTED, null, false));

让我们来看有哪些Listeners

JobRunner-Hello-17[1] print listeners

 listeners = "[

azkaban.execapp.FlowRunner$JobRunnerEventListener@2a32e7d1, 

azkaban.execapp.event.JobCallbackManager@2b274bce, 

azkaban.execapp.jmx.JmxJobMBeanManager@53976f5c

]"

那么,这3个Listener分别做了什么事情呢?

==========================================================================================

第一个,不好意思,这里是start,所以下面的代码都没有执行!

private class JobRunnerEventListener implements EventListener {

public JobRunnerEventListener() {

}

 

@Override

public synchronized void handleEvent(Event event) {

//看到这里了

JobRunner runner = (JobRunner) event.getRunner();

//看到这里了

if (event.getType() == Type.JOB_STATUS_CHANGED) {

updateFlow();

} else if (event.getType() == Type.JOB_FINISHED) {

ExecutableNode node = runner.getNode();

long seconds = (node.getEndTime() - node.getStartTime()) / 1000;

synchronized (mainSyncObj) {

logger.info("Job " + node.getNestedId() + " finished with status " + node.getStatus() + " in "

+ seconds + " seconds");

 

// Cancellation is handled in the main thread, but if the

// flow is

// paused, the main thread is paused too.

// This unpauses the flow for cancellation.

if (flowPaused && node.getStatus() == Status.FAILED && failureAction == FailureAction.CANCEL_ALL) {

flowPaused = false;

}

 

finishedNodes.add(node);

node.getParentFlow().setUpdateTime(System.currentTimeMillis());

interrupt();

fireEventListeners(event);

}

}

//看到这里了

}

==========================================================================================

第二个呢?

@Override

public void handleEvent(Event event) {

// 看到这里了

if (!isInitialized) {

return;

}

 

if (event.getRunner() instanceof JobRunner) {

// 看到这里了

try {

if (event.getType() == Event.Type.JOB_STARTED) {

// 这里开始执行

processJobCallOnStart(event);

} else if (event.getType() == Event.Type.JOB_FINISHED) {

processJobCallOnFinish(event);

}

} catch (Throwable e) {

// Use job runner logger so user can see the issue in their job

// log

JobRunner jobRunner = (JobRunner) event.getRunner();

jobRunner.getLogger().error("Encountered error while hanlding job callback event", e);

}

} else {

logger.warn("((( Got an unsupported runner: " + event.getRunner().getClass().getName() + " )))");

}

 

}

进去debug,发现还是什么都没有执行。。。怪我咯?

==========================================================================================

JmxJobMBeanManager

这个Listener具体做了啥呢?

if (event.getType() == Event.Type.JOB_STARTED) {

// 只是增加了一个计数器的值。。。

runningJobCount.incrementAndGet();


紧接着

loader.uploadExecutableNode(node, props);

这个到底干嘛的???

final String INSERT_EXECUTION_NODE = "INSERT INTO execution_jobs "

+ "(exec_id, project_id, version, flow_id, job_id, start_time, "

+ "end_time, status, input_params, attempt) VALUES (?,?,?,?,?,?,?,?,?,?)";

执行了这个语句先!

==========================================================================================

stop in azkaban.execapp.JobRunner.insertJobMetadata

==========================================================================================

 下面会构造一个JVM参数

/**

 * Add useful JVM arguments so it is easier to map a running Java process to

 * a flow, execution id and job

 */

private void insertJVMAargs() {

// 看到这里了

String flowName = node.getParentFlow().getFlowId();//

String jobId = node.getId();

// "-Dazkaban.flowid=World -Dazkaban.execid=17 -Dazkaban.jobid=Hello"

 

String jobJVMArgs = String.format("-Dazkaban.flowid=%s -Dazkaban.execid=%s -Dazkaban.jobid=%s", flowName,

executionId, jobId);

// "-Dazkaban.flowid=World -Dazkaban.execid=17 -Dazkaban.jobid=Hello"

 

String previousJVMArgs = props.get(JavaProcessJob.JVM_PARAMS);

jobJVMArgs += (previousJVMArgs == null) ? "" : " " + previousJVMArgs;

//存起来

logger.info("job JVM args: " + jobJVMArgs);

props.put(JavaProcessJob.JVM_PARAMS, jobJVMArgs);

}

==========================================================================================

然后发现,JobRunner实例,需要构造一个具体的job对象

这是在

job = jobtypeManager.buildJobExecutor(this.jobId, props, logger); 实现的

那么到底怎么构造这个对象呢?

鉴于这部分比较重要,我还是另开一个小节吧。


标签: azkaban
  • 打赏
  • 点赞
  • 收藏
  • 分享
共有 人打赏支持
强子哥哥
粉丝 840
博文 555
码字总数 635433
作品 8
×
强子哥哥
如果觉得我的文章对您有用,请随意打赏。您的支持将鼓励我继续创作!
* 金额(元)
¥1 ¥5 ¥10 ¥20 其他金额
打赏人
留言
* 支付类型
微信扫码支付
打赏金额:
已支付成功
打赏金额: