解码 DolphinScheduler：Flink 任务如何 “跑” 起来？

./sbin/start-dfs.sh# resource storage type: HDFS, S3, NONEresource.storage.type=HDFS# resource store on HDFS/S3 path, resource file will store to this hadoop hdfs path, self configuration, please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommendedresource.storage.upload.base.path=/dolphinscheduler# whether to startup kerberoshadoop.security.authentication.startup.state=false# resource view suffixs#resource.view.suffixs=txt,log,sh,bat,conf,cfg,py,java,sql,xml,hql,properties,json,yml,yaml,ini,js# if resource.storage.type=HDFS, the user must have the permission to create directories under the HDFS root pathresource.hdfs.root.user=lizu# if resource.storage.type=S3, the value like: s3a://dolphinscheduler; if resource.storage.type=HDFS and namenode HA is enabled, you need to copy core-site.xml and hdfs-site.xml to conf dirresource.hdfs.fs.defaultFS=hdfs://0.0.0.0:9000# use sudo or not, if set true, executing user is tenant user and deploy user needs sudo permissions; if set false, executing user is the deploy user and doesn't need sudo permissionssudo.enable=true

# JAVA_HOME, will use it to start DolphinScheduler serverexport JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk1.8.0_281.jdk/Contents/Homeexport PATH=$PATH:$JAVA_HOME/binexport M2_HOME=/Users/lizu/app/apache-maven-3.6.3export PATH=$PATH:$M2_HOME/binexport SCALA_HOME=/Users/lizu/app/scala-2.11.12#export SCALA_HOME=/Users/lizu/app/scala-2.12.13export PATH=$PATH:$SCALA_HOME/bin#export SPARK_HOME=/Users/lizu/app/spark-2.4.3-bin-hadoop2.7export SPARK_HOME=/Users/lizu/app/spark-3.0.2-bin-hadoop2.7export PATH=$PATH:$SPARK_HOME/binexport FLINK_HOME=/Users/lizu/app/flink-1.13.6export PATH=$PATH:$FLINK_HOME/binexport PATH=/Library/Frameworks/Python.framework/Versions/3.6/bin:$PATHexport HADOOP_HOME=/Users/lizu/app/hadoop-2.7.6export PATH=$PATH:$HADOOP_HOME/bin#export HIVE_HOME=/Users/lizu/app/apache-hive-1.2.1-binexport HIVE_HOME=/Users/lizu/app/hive-2.3.4export PATH=$PATH:$HIVE_HOME/bin#export NODE_HOME=/Users/lizu/app/node-v14.16.0export NODE_HOME=/Users/lizu/app/node-v16.17.0export PATH=$PATH:$NODE_HOME/binexport PATH=$PATH:/Users/lizu/app/go/binexport PATH=$PATH:/Users/lizu/app/gradle-6.3/binalias python="/usr/local/bin/python3"export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/pyspark:$SPARK_HOME/python/lib/py4j-0.10.9-src.zip:$PYTHONPATH

./start-cluster.sh

nc -l 9999

@PostConstructpublic void run() {    this.workerRpcServer.start();    this.workerRpcClient.start();    this.taskPluginManager.loadPlugin();    this.workerRegistryClient.setRegistryStoppable(this);    this.workerRegistryClient.start();    this.workerManagerThread.start();    this.messageRetryRunner.start();    /*     * registry hooks, which are called before the process exits     */    Runtime.getRuntime().addShutdownHook(new Thread(() -> {        if (!ServerLifeCycleManager.isStopped()) {            close("WorkerServer shutdown hook");        }    }));}

this.nettyRemotingServer.registerProcessor(CommandType.TASK_DISPATCH_REQUEST, taskDispatchProcessor);

final String workflowMasterAddress = taskDispatchCommand.getMessageSenderAddress();logger.info("Receive task dispatch request, command: {}", taskDispatchCommand);//....忽略部分代码	WorkerDelayTaskExecuteRunnable workerTaskExecuteRunnable = WorkerTaskExecuteRunnableFactoryBuilder      .createWorkerDelayTaskExecuteRunnableFactory(                taskExecutionContext,                workerConfig,                workflowMasterAddress,                workerMessageSender,                alertClientService,                taskPluginManager,                storageOperate)      .createWorkerTaskExecuteRunnable();// submit task to managerboolean offer = workerManager.offer(workerTaskExecuteRunnable);if (!offer) {    logger.warn("submit task to wait queue error, queue is full, current queue size is {}, will send a task reject message to master", workerManager.getWaitSubmitQueueSize());    workerMessageSender.sendMessageWithRetry(taskExecutionContext, workflowMasterAddress, CommandType.TASK_REJECT);} else {    logger.info("Submit task to wait queue success, current queue size is {}", workerManager.getWaitSubmitQueueSize());}

@Overridepublic void run() {    Thread.currentThread().setName("Worker-Execute-Manager-Thread");    while (!ServerLifeCycleManager.isStopped()) {        try {            if (!ServerLifeCycleManager.isRunning()) {                Thread.sleep(Constants.SLEEP_TIME_MILLIS);            }            if (this.getThreadPoolQueueSize() <= workerExecThreads) {                final WorkerDelayTaskExecuteRunnable workerDelayTaskExecuteRunnable = waitSubmitQueue.take();                workerExecService.submit(workerDelayTaskExecuteRunnable);            } else {                WorkerServerMetrics.incWorkerOverloadCount();                logger.info("Exec queue is full, waiting submit queue {}, waiting exec queue size {}",                        this.getWaitSubmitQueueSize(), this.getThreadPoolQueueSize());                ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);            }        } catch (Exception e) {            logger.error("An unexpected interrupt is happened, "                  + "the exception will be ignored and this thread will continue to run", e);        }    }}

//work线程池workerExecService = new WorkerExecService(        ThreadUtils.newDaemonFixedThreadExecutor("Worker-Execute-Thread", workerConfig.getExecThreads()),        taskExecuteThreadMap);

@Overridepublic void run() {    try {        // set the thread name to make sure the log be written to the task log file        Thread.currentThread().setName(taskExecutionContext.getTaskLogName());        LoggerUtils.setWorkflowAndTaskInstanceIDMDC(taskExecutionContext.getProcessInstanceId(),                taskExecutionContext.getTaskInstanceId());        logger.info("Begin to pulling task");        initializeTask();        if (Constants.DRY_RUN_FLAG_YES == taskExecutionContext.getDryRun()) {            taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.SUCCESS);            taskExecutionContext.setEndTime(new Date());            TaskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());            workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress,                    CommandType.TASK_EXECUTE_RESULT);            logger.info(                    "The current execute mode is dry run, will stop the subsequent process and set the taskInstance status to success");            return;        }        beforeExecute();        TaskCallBack taskCallBack = TaskCallbackImpl.builder().workerMessageSender(workerMessageSender)              .masterAddress(masterAddress).build();        executeTask(taskCallBack);        afterExecute();    } catch (Throwable ex) {        logger.error("Task execute failed, due to meet an exception", ex);        afterThrowing(ex);    } finally {        LoggerUtils.removeWorkflowAndTaskInstanceIdMDC();    }}

@Overridepublic void executeTask(TaskCallBack taskCallBack) throws TaskException {    if (task == null) {        throw new TaskException("The task plugin instance is not initialized");    }    task.handle(taskCallBack);}

// todo split handle to submit and track@Overridepublic void handle(TaskCallBack taskCallBack) throws TaskException {    try {        // SHELL task exit code        TaskResponse response = shellCommandExecutor.run(buildCommand());        setExitStatusCode(response.getExitStatusCode());        // set appIds        setAppIds(String.join(TaskConstants.COMMA, getApplicationIds()));        setProcessId(response.getProcessId());    } catch (InterruptedException ex) {        Thread.currentThread().interrupt();        logger.info("The current yarn task has been interrupted", ex);        setExitStatusCode(TaskConstants.EXIT_CODE_FAILURE);        throw new TaskException("The current yarn task has been interrupted", ex);    } catch (Exception e) {        logger.error("yarn process failure", e);        exitStatusCode = -1;        throw new TaskException("Execute task failed", e);    }}

public TaskResponse run(String execCommand) throws IOException, InterruptedException {    TaskResponse result = new TaskResponse();    int taskInstanceId = taskRequest.getTaskInstanceId();    if (null == TaskExecutionContextCacheManager.getByTaskInstanceId(taskInstanceId)) {        result.setExitStatusCode(EXIT_CODE_KILL);        return result;    }    if (StringUtils.isEmpty(execCommand)) {        TaskExecutionContextCacheManager.removeByTaskInstanceId(taskInstanceId);        return result;    }    String commandFilePath = buildCommandFilePath();    // create command file if not exists    createCommandFileIfNotExists(execCommand, commandFilePath);    // build process    buildProcess(commandFilePath);    // parse process output    parseProcessOutput(process);    int processId = getProcessId(process);    result.setProcessId(processId);    // cache processId    taskRequest.setProcessId(processId);    boolean updateTaskExecutionContextStatus =            TaskExecutionContextCacheManager.updateTaskExecutionContext(taskRequest);    if (Boolean.FALSE.equals(updateTaskExecutionContextStatus)) {        ProcessUtils.kill(taskRequest);        result.setExitStatusCode(EXIT_CODE_KILL);        return result;    }    // print process id    logger.info("process start, process id is: {}", processId);    // if timeout occurs, exit directly    long remainTime = getRemainTime();    // waiting for the run to finish    boolean status = process.waitFor(remainTime, TimeUnit.SECONDS);    // if SHELL task exit    if (status) {        // SHELL task state        result.setExitStatusCode(process.exitValue());    } else {        logger.error("process has failure, the task timeout configuration value is:{}, ready to kill ...",                taskRequest.getTaskTimeout());        ProcessUtils.kill(taskRequest);        result.setExitStatusCode(EXIT_CODE_FAILURE);    }    int exitCode = process.exitValue();    String exitLogMessage = EXIT_CODE_KILL == exitCode ? "process has killed." : "process has exited.";    logger.info(exitLogMessage          + " execute path:{}, processId:{} ,exitStatusCode:{} ,processWaitForStatus:{} ,processExitValue:{}",            taskRequest.getExecutePath(), processId, result.getExitStatusCode(), status, exitCode);    return result;}

[INFO] 2023-06-22 23:49:12.520 +0800 - flink task command : flink run -p 1 -sae -c org.apache.flink.streaming.examples.socket.SocketWindowWordCount flink-example/SocketWindowWordCount.jar --hostname localhost --port 9999[INFO] 2023-06-22 23:49:12.520 +0800 - Begin to create command file:/tmp/dolphinscheduler/exec/process/lizu/9986178674720/9986206731168_3/5/5/5_5.command[INFO] 2023-06-22 23:49:12.521 +0800 - Success create command file, command: #!/bin/bashBASEDIR=$(cd `dirname $0`; pwd)cd $BASEDIRsource /Users/lizu/idea/scheduler/dolphinscheduler/dolphinscheduler-standalone-server/target/classes/dolphinscheduler_env.shflink run -p 1 -sae -c org.apache.flink.streaming.examples.socket.SocketWindowWordCount flink-example/SocketWindowWordCount.jar --hostname localhost --port 9999[INFO] 2023-06-22 23:49:12.526 +0800 - task run command: sudo -u lizu -E bash /tmp/dolphinscheduler/exec/process/lizu/9986178674720/9986206731168_3/5/5/5_5.command[INFO] 2023-06-22 23:49:12.547 +0800 - process start, process id is: 23572[INFO] 2023-06-22 23:49:15.560 +0800 -  -> Job has been submitted with JobID 192c0f1b984f2bc10cd2ec6d39525fbb

protected void sendTaskResult() {    taskExecutionContext.setCurrentExecutionStatus(task.getExitStatus());    taskExecutionContext.setEndTime(new Date());    taskExecutionContext.setProcessId(task.getProcessId());    taskExecutionContext.setAppIds(task.getAppIds());    taskExecutionContext.setVarPool(JSONUtils.toJsonString(task.getParameters().getVarPool()));    workerMessageSender.sendMessageWithRetry(taskExecutionContext, masterAddress, CommandType.TASK_EXECUTE_RESULT);    logger.info("Send task execute result to master, the current task status: {}",            taskExecutionContext.getCurrentExecutionStatus());}

[INFO] 2023-06-22 23:49:15.560 +0800 -  -> Job has been submitted with JobID 192c0f1b984f2bc10cd2ec6d39525fbb[INFO] 2023-06-23 00:11:35.405 +0800 - process has exited. execute path:/tmp/dolphinscheduler/exec/process/lizu/9986178674720/9986206731168_3/5/5, processId:23572 ,exitStatusCode:1 ,processWaitForStatus:true ,processExitValue:1[INFO] 2023-06-23 00:11:35.411 +0800 - Send task execute result to master, the current task status: TaskExecutionStatus{code=6, desc='failure'}