聯童科技基于incubator-dolphinscheduler從0到1構建大資料調度平臺之路-有解無憂

聯童科技是一家智能化母嬰童產業平臺，從事母嬰童行業以及互聯網技術多年，擁有豐富的母嬰門店運營和系統開發經驗，在會員經營和商品經營方面，能夠圍繞會員需求，深入場景，更貼近合作伙伴和消費者，提供最優服務產品，公司致力于以技術來驅動母嬰童產業的發展，公司也希望借助于大資料為客戶提供更多智能資料分析和決策分析，大資料是公司重點發展的一部分，公司從成立初期起就搭建了大資料團隊，有了大資料團隊后，大資料調度平臺的構建自然是最基礎也是最重要的環節，

一、為什么選擇incubator-dolphinscheduler

1、incubator-dolphinscheduler是一個由國內公司發起的開源專案，中國本土社區成員非常活躍，更加容易去進行社區溝通，同時聯童也希望能加入到這個社區中，一起把這個由本土成員為主成立的社區做的更好，

2、incubator-dolphinscheduler 能夠支撐非常多的應用場景

以DAG圖的方式將Task按照任務的依賴關系關聯起來，可實時可視化監控任務的運行狀態
支持豐富的任務型別：Shell、MR、Spark、SQL(mysql、postgresql、hive、sparksql),Python,Sub_Process、Procedure，flink，datax，sqoop，http等
支持作業流定時調度、依賴調度、手動調度、手動暫停/停止/恢復，同時支持失敗重試/告警、從指定節點恢復失敗、Kill任務等操作
支持作業流優先級、任務優先級及任務的故障轉移及任務超時告警/失敗
支持作業流全域引數及節點自定義引數設定
支持資源檔案的在線上傳/下載，管理等，支持在線檔案創建、編輯
支持任務日志在線查看及滾動、在線下載日志等
實作集群HA，通過Zookeeper實作Master集群和Worker集群去中心化
支持對Master/Worker cpu load，memory，cpu在線查看
支持作業流運行歷史樹形/甘特圖展示、支持任務狀態統計、流程狀態統計
支持補數
支持多租戶
支持國際化

其中DAG圖借鑒自spark ，在dolphinscheduler 一個作業流可以對應多個作業任務，每一個作業任務對應一個DAG中的節點，

3、incubator-dolphinscheduler在保證了高并發和高可用的設計時，架構思路也相對簡單，技術架構中沒有引入非常多的復雜技術組件，降低了學習和維護的成本，

備注：此架構圖摘自社區官方網站

incubator-dolphinscheduler在設計時，除了zookeeper外，沒有引入太多復雜的技術組件，整個架構以zookeeper 作為集群管理，采用去中心化思想進行設計，

二、incubator-dolphinscheduler功能的不足

1、無法支持串行調度策略

incubator-dolphinscheduler 在一開始設計時，只支持并行調度，不支持串行調度，而在聯童中，大部分場景都是需要串行運行的，也就是每一個作業流任務都只能有一個實體在運行，同一個作業流任務中必須要等前一個實體執行結束，下一個實體才能開始執行，這種場景大多出現在準實時任務中，

2、任務依賴不夠強大，只能支持被動等待依賴執行成功，無法主動觸發下游作業流實體運行

如下圖所示，只能支持在創建任務時，被動去等待依賴執行成功，無法在當前任務執行成功后，主動去觸發別的作業流任務執行，

3、部分模塊中用戶體驗不足，并且在資料量大時，部分模塊資料查詢性能較慢

4、缺少比較完備的監控體系

在 incubator-dolphinscheduler 只提供了一些簡單的監控，當有多大幾千個任務在運行時，很難做到完備監控，更是缺少對每一個任務運行的性能分析，

三、我們對于incubator-dolphinscheduler的功能升級開發

1、增加串行調度的支持

如下圖所示，我們在原有并行執行的基礎上，增加了串行執行方式，

在串行執行時，我們還增加了串行執行的佇列功能，每一任務都可以指定佇列的長度大小，

2、增加主動觸發下游作業流實體運行

如下圖所示，我們在原有并行執行的基礎上，增加主動觸發下游一個或者多個作業流實體運行，

運行后效果如下：

3、一些較大的Bug修復

聯童在使用 incubator-dolphinscheduler時，同樣也踩過不少的坑，這里我們舉其中一個例子，比如在內部使用時，同事反饋最多的問題就是調度任務的日志重繪不及時，有時候很久才能重繪出日志，后來經過原始碼分析，發現是原始碼中存在了一些不太健壯的處理導致了這個問題，

incubator-dolphinscheduler 中AbstractCommandExecutor.java 部分原始碼

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.dolphinscheduler.server.worker.task;

import static org.apache.dolphinscheduler.common.Constants.EXIT_CODE_FAILURE;
import static org.apache.dolphinscheduler.common.Constants.EXIT_CODE_KILL;
import static org.apache.dolphinscheduler.common.Constants.EXIT_CODE_SUCCESS;

import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.LoggerUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.server.entity.TaskExecutionContext;
import org.apache.dolphinscheduler.server.utils.ProcessUtils;
import org.apache.dolphinscheduler.server.worker.cache.TaskExecutionContextCacheManager;
import org.apache.dolphinscheduler.server.worker.cache.impl.TaskExecutionContextCacheManagerImpl;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.slf4j.Logger;

/**
 * abstract command executor
 */
public abstract class AbstractCommandExecutor {
    /**
     * rules for extracting application ID
     */
    protected static final Pattern APPLICATION_REGEX = Pattern.compile(Constants.APPLICATION_REGEX);

    protected StringBuilder varPool = new StringBuilder();
    /**
     * process
     */
    private Process process;

    /**
     * log handler
     */
    protected Consumer<List<String>> logHandler;

    /**
     * logger
     */
    protected Logger logger;

    /**
     * log list
     */
    protected final List<String> logBuffer;

    /**
     * taskExecutionContext
     */
    protected TaskExecutionContext taskExecutionContext;

    /**
     * taskExecutionContextCacheManager
     */
    private TaskExecutionContextCacheManager taskExecutionContextCacheManager;

    public AbstractCommandExecutor(Consumer<List<String>> logHandler,
                                   TaskExecutionContext taskExecutionContext,
                                   Logger logger) {
        this.logHandler = logHandler;
        this.taskExecutionContext = taskExecutionContext;
        this.logger = logger;
        this.logBuffer = Collections.synchronizedList(new ArrayList<>());
        this.taskExecutionContextCacheManager = SpringApplicationContext.getBean(TaskExecutionContextCacheManagerImpl.class);
    }

    /**
     * build process
     *
     * @param commandFile command file
     * @throws IOException IO Exception
     */
    private void buildProcess(String commandFile) throws IOException {
        // setting up user to run commands
        List<String> command = new LinkedList<>();

        //init process builder
        ProcessBuilder processBuilder = new ProcessBuilder();
        // setting up a working directory
        processBuilder.directory(new File(taskExecutionContext.getExecutePath()));
        // merge error information to standard output stream
        processBuilder.redirectErrorStream(true);

        // setting up user to run commands
        command.add("sudo");
        command.add("-u");
        command.add(taskExecutionContext.getTenantCode());
        command.add(commandInterpreter());
        command.addAll(commandOptions());
        command.add(commandFile);

        // setting commands
        processBuilder.command(command);
        process = processBuilder.start();

        // print command
        printCommand(command);
    }

    ..........


    /**
     * get the standard output of the process
     *
     * @param process process
     */
    private void parseProcessOutput(Process process) {
        String threadLoggerInfoName = String.format(LoggerUtils.TASK_LOGGER_THREAD_NAME + "-%s", taskExecutionContext.getTaskAppId());
        ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
        parseProcessOutputExecutorService.submit(new Runnable() {
            @Override
            public void run() {
                BufferedReader inReader = null;

                try {
                    inReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
                    String line;

                    long lastFlushTime = System.currentTimeMillis();

                    while ((line = inReader.readLine()) != null) {
                        if (line.startsWith("${setValue(")) {
                            varPool.append(line.substring("${setValue(".length(), line.length() - 2));
                            varPool.append("$VarPool$");
                        } else {
                            logBuffer.add(line);
                            lastFlushTime = flush(lastFlushTime);
                        }
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                } finally {
                    clear();
                    close(inReader);
                }
            }
        });
        parseProcessOutputExecutorService.shutdown();
    }

................

    /**
     * when log buffer siz or flush time reach condition , then flush
     *
     * @param lastFlushTime last flush time
     * @return last flush time
     */
    private long flush(long lastFlushTime) {
        long now = System.currentTimeMillis();

        /**
         * when log buffer siz or flush time reach condition , then flush
         */
        if (logBuffer.size() >= Constants.DEFAULT_LOG_ROWS_NUM || now - lastFlushTime > Constants.DEFAULT_LOG_FLUSH_INTERVAL) {
            lastFlushTime = now;
            /** log handle */
            logHandler.accept(logBuffer);

            logBuffer.clear();
        }
        return lastFlushTime;
    }

    /**
     * close buffer reader
     *
     * @param inReader in reader
     */
    private void close(BufferedReader inReader) {
        if (inReader != null) {
            try {
                inReader.close();
            } catch (IOException e) {
                logger.error(e.getMessage(), e);
            }
        }
    }

    protected List<String> commandOptions() {
        return Collections.emptyList();
    }

    protected abstract String buildCommandFilePath();

    protected abstract String commandInterpreter();

    protected abstract void createCommandFileIfNotExists(String execCommand, String commandFile) throws IOException;
}

在這段原始碼中，parseProcessOutput(Process process) 方法是負責任務日志的獲取以及Flush，但是由于采用了BufferedReader 中的readLine() 方法來讀取任務行程的process.getInputStream()日志，由于readLine() 是一個阻塞方法，

flush(long lastFlushTime) 方法在處理時有一個判斷條件if (logBuffer.size() >= Constants.DEFAULT_LOG_ROWS_NUM || now - lastFlushTime > Constants.DEFAULT_LOG_FLUSH_INTERVAL)，只有當日志條數達到64潭訓者間隔1s時才會

flush，按理說，代碼其實是要實作至少每隔1s會flash 一次日志，但是由于readLine() 是一個阻塞方法，所以并不會一直在執行，而是readLine()必須是讀取到新資料后，才會執行flush方法，那么在出現1s內產生的任務日志不滿足64條，而任務又很久沒有新日志出現時，就會觸發這個bug，例如執行如下一個shell 腳本任務，由于每個執行步驟產生的日志少，而且每個步驟執行的時間又很久，時間間隔很大，就會出現很久都不會重繪上一次產生的日志，

#!/bin/bash
echo "hello world"
exec 10m
sleep 100000s
echo "hello world2"
exec 10m
sleep 100000s
echo "hello world3"
exec 10m
sleep 100000s

之后我們對這段原始碼進行了重寫，采用了兩個執行緒進行處理，一個執行緒負責readline()，一個執行緒負責flush.做到在readline()方法的執行緒阻塞時，不影響flush執行緒的處理，

/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


public abstract class AbstractCommandExecutor {
    /**
     * rules for extracting application ID
     */
    protected static final Pattern APPLICATION_REGEX = Pattern.compile(Constants.APPLICATION_REGEX);

    /**
     * process
     */
    private Process process;

    /**
     * log handler
     */
    protected Consumer<List<String>> logHandler;

    /**
     * logger
     */
    protected Logger logger;

    /**
     * log list
     */
    protected final List<String> logBuffer;

    protected boolean logOutputIsScuccess = false;

    /**
     * taskExecutionContext
     */
    protected TaskExecutionContext taskExecutionContext;

    /**
     * taskExecutionContextCacheManager
     */
    private TaskExecutionContextCacheManager taskExecutionContextCacheManager;

.........
 /**
     * get the standard output of the process
     *
     * @param process process
     */
    private void parseProcessOutput(Process process) {
        String threadLoggerInfoName = String.format(LoggerUtils.TASK_LOGGER_THREAD_NAME + "-%s", taskExecutionContext.getTaskAppId());
        ExecutorService getOutputLogService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName + "-" + "getOutputLogService");
        getOutputLogService.submit(() -> {
            BufferedReader inReader = null;

            try {
                inReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
                String line;while ((line = inReader.readLine()) != null) {
                    logBuffer.add(line);
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            } finally {
                logOutputIsScuccess = true;
                close(inReader);
            }
        });
        getOutputLogService.shutdown();

        ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
        parseProcessOutputExecutorService.submit(() -> {
            try {
                long lastFlushTime = System.currentTimeMillis();
                while (logBuffer.size() > 0 || !logOutputIsScuccess) {
                    if (logBuffer.size() > 0) {
                        lastFlushTime = flush(lastFlushTime);
                    } else {
                        Thread.sleep(Constants.DEFAULT_LOG_FLUSH_INTERVAL);
                    }
                }
            } catch (Exception e) {
                logger.error(e.getMessage(), e);
            } finally {
                clear();
            }
        });
        parseProcessOutputExecutorService.shutdown();
    }
.......
    /**
     * when log buffer siz or flush time reach condition , then flush
     *
     * @param lastFlushTime last flush time
     * @return last flush time
     */
    private long flush(long lastFlushTime) throws InterruptedException {
        long now = System.currentTimeMillis();

        /**
         * when log buffer siz or flush time reach condition , then flush
         */
        if (logBuffer.size() >= Constants.DEFAULT_LOG_ROWS_NUM || now - lastFlushTime > Constants.DEFAULT_LOG_FLUSH_INTERVAL) {
            lastFlushTime = now;
            /** log handle */
            logHandler.accept(logBuffer);

            logBuffer.clear();
        }

        return lastFlushTime;
    }
.......
}

4、將調度系統的監控接入到prometheus和grafana中

incubator-dolphinscheduler 只提供了一些如下的簡單實時監控，尤其缺少對任務的監控，

聯童在此基礎上，引入了prometheus和grafana，

使用prometheus和grafana 不但可以監控到調度系統任務的總體運行，也可以監控到單個任務的運行耗時曲線等，

5、對incubator-dolphinscheduler 的性能優化

待稍后晚點補充

四、聯童對于開源社區的擁抱和回饋

聯童雖然是一家新興起的母嬰童公司，但是在成立的初始，就秉承著以技術來驅動母嬰童產業的發展，公司擁有一個非常好的技術團隊，也一直在擁抱開源社區，目前已經引入了incubator-dolphinscheduler、prometheus、grafana 、hadoop、spark、flink、hive、presto......等很多開源專案來支撐公司的技術驅動，在未來，聯童也一定回不斷的去回饋開源社區，去提供更多的Pull requests，貢獻自己的一份力量，

【備注】本篇文章摘選自【聯童直通車】公眾號： https://mp.weixin.qq.com/s?__biz=MzI5NDM3MzU0MQ==&mid=2247483996&idx=1&sn=cdb180446fa215705b21e09bdfb632fd&chksm=ec629396db151a80ee6c8b7f860cf4061b72e2bf0f1f099ef281bc914d8f5bcd60e70c54c2da&scene=126&&sessionid=1614058343&version=3.1.1.3006&platform=win#rd

歡迎加入APachec dolphinscheduler社區

https://github.com/apache/incubator-dolphinscheduler (請記得fork和star)
訂閱郵件串列
用自己的郵箱向[email protected]發送一封郵件，主題和內容任意，
接收確認郵件并回復，完成步驟1后，將收到一封來自[email protected]的確認郵件（如未收到，請確認郵件是否被自動歸入垃圾郵件、推廣郵件、訂閱郵件等檔案夾），然后直接回復該郵件，或點擊郵件里的鏈接快捷回復即可，主題和內容任意，
接識訓迎郵件，完成以上步驟后，會收到一封主題為WELCOME to [email protected]的歡迎郵件，至此已成功訂閱Apache DolphinScheduler（Incubating）的郵件串列，

轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/262873.html

標籤：大數據

上一篇：數碼管顯示

下一篇：大資料盤點2020年直播帶貨那些事兒