拈花云科基于 Apache DolphinScheduler 在文旅業態下的實踐-有解無憂

file

作者｜云科NearFar X Lab團隊左益、周志銀、洪守偉、陳超、武超

一、導讀

無錫拈花云科技服務有限公司（以下簡稱：拈花云科）是由拈花灣文旅和北京滴普科技共同范訓的文旅目的地數智化服務商，2022年底，拈花云科NearFar X Lab團隊開始測驗DolphinScheduler作為交付型專案和產品專案的任務調度工具，本文主要分享了拈花云科在任務調度工具的選擇、迭代和實踐程序中的經驗，希望對大家有所啟發，

二、業務背景

我們的服務物件主要是國內各個景區、景點，業務范圍涵蓋文旅行業的多個板塊，如票務、交通、零售、住宿、餐飲、演繹、游樂、影院、KTV、租賃、服務、會務、康樂、康養、電商、客服、營銷、分銷、安防等，由于業務系統來源較多，多系統下的資料源型別差異化較大，所以在實施資料專案時我們需要能夠支持多種資料來源（Mysql、Oracle、SqlServer、Hive、Excel……）的資料集成任務，同時根據大部分景區為國有化的特點，我們也需要具備能夠提供私有化交付部署及SAAS化資料中臺產品解決方案的雙重服務支撐能力，

三、DolphinScheduler 調度系統選型程序

在團隊成立之初為了快速構建MVP業務版本，我們沿用了團隊同事之前用過的Kettle調度方案，該方案下通過Kettle完成可視化調度的配置及對于異構資料的集成任務，通過Python 呼叫HQL腳本完成基于Hive的傳參資料計算，
file
基于MVP的構建，我們也開始思考，在我們的整體中臺架構下該需要一個什么樣的調度系統，以及除了調度這件事本身我們還需要哪些功能和能力，帶著這些問題我們開始整理自己的需求，以及每個需求下有什么樣的產品可以適配，

調度系統需要支撐的應用場景

文旅業態下的資料使用場景與其它業態下的使用場景大體相同，主要分為以下四類：

file

調度系統需要支撐的專案型別

我們選擇的調度系統需要同時具備實施類專案、SAAS產品兩種需求下的資料中臺支撐能力
file

基于以上需求我們進行了調度系統的選型對比，網上有非常多關于Oozie、Azkaban、Airflow、DolphinScheduler、Xxl-job、Kettle等調度選型的文章及介紹，在此不過多的展開他們的優缺點，我們覺得每個產品的設計都有它自身的考量，都有適用與不適用的場景，結合我們自身的使用需求最終我們選擇了使用DolphinScheduler作為資料中臺的調度平臺，

主要原因如下：

High Reliability（高可靠性）
高可靠性是我們看重的第一要點，因為不管是實施專案還是SAAS產品，只有系統穩定產品才可以正常運行，DolphinScheduler通過去中心化設計、原生 HA 任務佇列支持、過載容錯能力支持提供了高度穩健的環境，在我們半年的使用程序中也驗證了其非常穩定，
High Scalability：（高擴展性）
由于我們要支持實施專案/SAAS產品兩種場景下的使用，DolphinScheduler的多租戶支持很好的契合了SAAS場景下資源隔離的使用需求，同時其擴縮容能力、高度的調度任務上限（10萬+）都能很好的支撐我們業務的擴展性需求，
豐富的資料集成能力
DolphinScheduler提供的任務型別已經遠遠涵蓋了我們經常使用的任務型別（DataX、SeaTunnel的支持本身就涵蓋了較多的Source2Target同步/推送場景），
支持Kubernetes部署
上文提到在私有化的部署方式下客戶的部署環境大不相同，對于實施團隊來說，如果能夠簡單、高效、一致的完成部署則會極大的提高專案投遞部署效率，同時也能減少很多因環境原因而產生的問題，
不僅僅是調度
在調研DolphinScheduler的程序中我們發現，除了調度本身這個環節，結合DCMM（資料管理能力成熟度評估模型）的國標8個能力域，DolphinScheduler在資料質量模塊也做了很多實作，這無疑又命中了我們對于資料質量能力建設的需求，同時Apache DolphinScheduler的服務架構中還有AlertServer服務，作為整體資料中臺方案來說后期完全可以將所有報警需求集成在Apache DolphinScheduler的報警服務中，避免多系統重復造輪子，從這些點考慮DolphinScheduler它不僅僅是一個調度工具而更像是一個資料開發平臺，（期待隨著社區的迭代會有更完整的生態實作）
問題處理難度
DolphinScheduler社區非常的活躍，在加入DolphinScheduler社區群后每天都可以看到非常多的伙伴在交流關于Apache DolphinScheduler使用程序中的問題，社區人員也都積極的予以回復，同時Dolphinscheduler又是咱們國產開源軟體，所以完全不必擔心存在溝通上的障礙，

四、基于DolphinScheduler的專案實踐

1、DolphinScheduler ON Kubernetes

DolphinScheduler支持多種部署方式:單機部署(Standalone)、偽集群部署(Pseudo-Cluster)、集群部署(Cluster)、Kubernetes部署(Kubernetes)，在專案實施的場景下由于客戶提供的部署環境千變萬化，我們需要一種穩定、快速、不挑環境的部署方式，Apache DolphinScheduler on K8S的部署方式很好的滿足了我們的需求，此部署方式能極大的提高整體專案的部署效率及動態擴展性，

Kubernetes是一個開源的容器編排平臺，可以實作容器的自動化部署、擴縮容、服務發現、負載均衡，可以提高DolphinScheduler的可用性、可擴展性和可維護性
Kubernetes可以支持多種存盤型別，包括本地存盤、網路存盤和云，可以滿足DolphinScheduler多節點共享持久化存盤需求
Kubernetes可以支持多種調度策略，包括親和性、反親和性、污點和容忍，可以優化DolphinScheduler的資源利用率，提高任務執行效率，
Kubernetes可以支持多種監控和日志方案，包括Prometheus、Grafana、ELK等等，可以方便地對DolphinScheduler的運行狀態和性能進行監控，分析和告警

在部署Apache DolphinScheduler on K8S 的程序中我們也曾遇到過一些問題，下面是我們總結的一些Kubernetes部署要點：

自定義鏡像

FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-alert-server:版本號
# 如果你想支持 MySQL 資料源
COPY ./mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/libs

dolphinscheduler-api

FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-api:版本號

# 如果你想支持 MySQL 資料源
COPY ./mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/libs

# 如果你想支持 Oracle 資料源
COPY ./ojdbc8-19.9.0.0.jar /opt/dolphinscheduler/libs

dolphinscheduler-master

FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-master:版本號

# 如果你想支持 MySQL 資料源
COPY ./mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/libs

dolphinscheduler-tools

FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-tools:版本號

# 如果你想支持 MySQL 資料源
COPY ./mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/libs

dolphinscheduler-worker

FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:版本號

# 如果你想支持 MySQL 資料源
COPY ./mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/libs

# 如果你想支持 Oracle 資料源
COPY ./ojdbc8-19.9.0.0.jar /opt/dolphinscheduler/libs

# 如果你想支持 hadoop 資料源
COPY ./hadoop-common-2.7.3.jar /opt/dolphinscheduler/libs
COPY ./hadoop-core-1.2.1.jar /opt/dolphinscheduler/libs

# 如果你想支持 hive 資料源
COPY ./hive-common.jar /opt/dolphinscheduler/libs
COPY ./hive-jdbc.jar /opt/dolphinscheduler/libs
COPY ./hive-metastore.jar /opt/dolphinscheduler/libs
COPY ./hive-serde.jar /opt/dolphinscheduler/libs
COPY ./hive-service.jar /opt/dolphinscheduler/libs

# 安裝python3環境
RUN apt-get update && \
    apt-get install -y --no-install-recommenApache DolphinScheduler curl && \
    rm -rf /var/lib/apt/lists/*

RUN apt-get update && \
    apt-get install -y --no-install-recommenApache DolphinScheduler libcurl4-openssl-dev libssl-dev && \
    rm -rf /var/lib/apt/lists/*

RUN apt-get update && \
    apt-get install -y --no-install-recommenApache DolphinScheduler python3 && \
    rm -rf /var/lib/apt/lists/*

RUN apt-get update && \
    apt-get install -y --no-install-recommenApache DolphinScheduler python3-pip && \
    rm -rf /var/lib/apt/lists/*
    
# 安裝dataX 并且解壓縮
COPY ./datax.tar.gz /home
RUN tar -zxvf /home/datax.tar.gz -C /opt

組態檔修改

官方教程中的helm進行安裝，在安裝程序中需要修改原始碼中 "/deploy/kubernetes/dolphinscheduler/" 路徑下的 "values.yaml，Chart.yaml" 里的相關鏡像和版本(注:原配置沒有指定持久化儲存類，會使用默認的存盤類，建議是修改并使用可多節點讀寫并且可以彈性擴展的，多節點讀寫便于worker節點共用同一套lib)

生產配置

dolphinscheduler-api 3 副本（注：默認是 1副本，但是實際使用中發現，平臺頁面在使用程序中會有卡頓，增加副本數之后，會有明顯改善）
dolphinscheduler-alert 1副本
dolphinscheduler-zookeeper 1副本
dolphinscheduler-worker 3副本
dolphinscheduler-master 3副本

k8s部署總結

采用k8s部署后，最大感受就是可排除環境干擾，快速擴展，遷移，部署，幫助我司實作了資料中臺私有化中的調度標準化，該方案已在多個景區中進行快速落地并應用，

2、基于SQL腳本血緣的DolphinScheduler作業流自動化實作

專案背景

基于景區中各個業務系統（票務、營銷、安防、酒店、商業、能耗、停車等）在景區機房下建設資料中臺，完成以下應用需求：

滿足各個業務部門日常的報表需求
支持各類大屏看板展示
為景區的管理層決策提供資料依據

技術選型

資料倉庫：Doris
調度工具：DolphinScheduler 使用版本：3.0.0
版本管理：Gitlab
容器編排：Kubernete

處理流程

業務分析與指標確認：與業務部門溝通，了解業務需求和目標，明確資料指標的定義、計算邏輯和展示方式，
資料倉庫分層和設計：根據資料倉庫的四層架構（ODS、DWD、DWS、ADS），設計資料模型和表結構，規范命名和注釋，
資料腳本開發：撰寫資料抽取、清洗、轉換、加載等腳本，實作資料從源系統到目標表的流轉和處理，
資料任務調度：入倉后的執行腳本通過血緣識別依賴自動錄入DolphinScheduler形成作業流任務調度（設定任務依賴、觸發條件、重試策略等引數）監控任務運行狀態和日志，
資料輸出和檔案：輸出標準指標庫和相關檔案，供BI、可視化報表等使用，同時撰寫資料開發檔案，記錄資料開發程序中的關鍵資訊和問題，

下面介紹下我們根據SQL血緣識別自動生成Apache DolphinScheduler調度任務的實作程序：

在DolphinScheduler平臺上開發資料調度作業流的程序中我們遇到一個問題：如果一個作業流下的任務量太多，在原有的UI界面中想通過界面方式完成配置更改以及血緣關系的建立等操作會非常不便捷，即便是通過任務定義去配置，當上百個任務同時需要配置依賴關系時也是一個不小的作業量開銷而且還容易出錯，后期的更新迭代也較為不便，

我們結合作業流下SQL任務本身的特點（數倉SQL任務是整體按照Apache DolphinScheduler、DWD、DWS、Apache DolphinScheduler 的計算流程進行計算，每個表對應一個Apache DolphinScheduler的Task既每個Task都會包含SourceTable及TargetTabe，通過這層關系我們就可以構建起完整的數倉任務血緣依賴），基于以上想法我們實作了從資料腳本自動生成對應的作業流和任務的自動化方案：

資料入倉后的開發腳本以每個表單為單位對應生成一個TaskSQL腳本提交到git，
自動化工具生成作業流及任務前自動從git庫中獲取最新的資料腳本，
自動化工具拉取到最新代碼后識別和分析所有資料腳本之間的血緣關系，
自動化工具通過血緣關系自動將所有的任務關聯并組裝到一個作業流中：每個任務執行完成后，會立即觸發下游任務，最大化地利用服務器資源，

以下是該實作的核心代碼：

sql決議

SqlParse是使用阿里的druid中的組件MySqlStatementParser 
/**
 * sql決議
 * @param sqlStr
 * @return
 */
public static Map<String, Set<String>> sqlParser(String sqlStr) {
    List<String> sqlList = StrUtil.split(sqlStr, ";");
    Map<String, Set<String>> map = new HashMap<>();
    for (String sql : sqlList) {
        if (StrUtil.isBlank(sql)) {
            continue;
        }
        // 這里使用的時 mysql 決議
        MySqlStatementParser parser = new MySqlStatementParser(sql);
        SQLStatement sqlStatement = parser.parseStatement();
        MySqlSchemaStatVisitor visitor = new MySqlSchemaStatVisitor();
        sqlStatement.accept(visitor);
        Map<TableStat.Name, TableStat> tableStatMap = visitor.getTables();
        for (Map.Entry<TableStat.Name, TableStat> tableStatEntry : tableStatMap.entrySet()) {
            String name = tableStatEntry.getKey().getName();
            // 這里的 value 有兩種 Select（父級）、Insert（子級）
            String value = https://www.cnblogs.com/DolphinScheduler/p/tableStatEntry.getValue().toString();
            if (map.containsKey(value)) {
                map.get(value).add(name);
            } else {
                Set list = new HashSet<>();
                list.add(name);
                map.put(value, list);
            }
        }
    }
    return map;
}

節點物件定義

/**
 * 任務節點定義
 */
public class Apache DolphinSchedulerTaskNode implements Serializable {

    private static final long serialVersionUID = 1L;

    /**
     * 源表
     */
    private List<String> sourceTableName = new ArrayList<>();

    /**
     * 目標表
     */
    private String targetTableName;

    /**
     * 源sql
     */
    private String sql;

    /**
     * 用sql做一個MD5簽名
     */
    private String md5;

    /**
     * 用sql名稱
     */
    private String name;

    /**
     * 任務code
     */
    private Long taskCode;

    ...
}

/**
 * 樹型節點定義
 */
public class MyTreeNode extenApache DolphinScheduler Apache DolphinSchedulerTaskNode {
    /**
     * 添加一個子節點串列屬性
     */
    private List<MyTreeNode> children;

    ...
}

樹型結構組裝

/**
 * 決議sql,并組裝node
 *
 * @param files
 * @return
 */
private static List<MyTreeNode> treeNodeProcess(List<File> files) {
    List<MyTreeNode> sourceList = new ArrayList<>();
    for (File sqlFile : files) {
        // 1 取出里面的 sql 腳本內容
        String sql = FileUtil.readUtf8String(sqlFile);
        // 2 決議里面的腳本內容
        Map<String, Set<String>> map = null;
        try {
            // 血緣決議
            map = AutoCreateTask.sqlParser(sql);
        } catch (Exception e) {
            log.error(" table-parser error: {}", sqlFile.getPath());
        }
        // 3 將每一個sql的 source , target 決議出來
        if (ObjectUtil.isNotNull(map)) {
            MyTreeNode treeNode = new MyTreeNode();
            treeNode.setName(sqlFile.getName());
            if (map.containsKey("Select")) {
                Set<String> select = map.get("Select");
                treeNode.setSourceTableName(new ArrayList<>(select));
            }
            treeNode.setSql(sql);
            if (map.containsKey("Insert")) {
                Set<String> insert = map.get("Insert");
                treeNode.setTargetTableName(new ArrayList<>(insert).get(0));
            }
            sourceList.add(treeNode);
        }
    }
    // 將sql按照 source , target 組成 樹狀結構
    return TreeUtil.getTree(sourceList);
}

/**
 * 組成樹狀結構
 * @param sourceList
 * @return
 */
public static List<MyTreeNode> getTree(List<MyTreeNode> sourceList) {
    Map<String, Set<MyTreeNode>> sourceMap = new HashMap<>();
    Map<String, Set<MyTreeNode>> targetMap = new HashMap<>();
    for (MyTreeNode node : sourceList) {
        //源表Map
        List<String> subSourceTableList = node.getSourceTableName();
        if (IterUtil.isNotEmpty(subSourceTableList)) {
            for (String subSourceTable : subSourceTableList) {
                if (sourceMap.containsKey(subSourceTable)) {
                    sourceMap.get(subSourceTable).add(node);
                } else {
                    Set<MyTreeNode> nodeSet = new HashSet<>();
                    nodeSet.add(node);
                    sourceMap.put(subSourceTable, nodeSet);
                }
            }
        }

        //目標表Map
        String targetTableName = node.getTargetTableName();
        if (targetMap.containsKey(targetTableName)) {
            targetMap.get(targetTableName).add(node);
        } else {
            Set<MyTreeNode> nodeSet = new HashSet<>();
            nodeSet.add(node);
            targetMap.put(targetTableName, nodeSet);
        }
    }
    //創建一個串列，用于存盤根節點
    List<MyTreeNode> rootList = new ArrayList<>();
    for (MyTreeNode node : sourceList) {
        // 將子節點處理好
        String targetTableName = node.getTargetTableName();
        if (sourceMap.containsKey(targetTableName)) {
            List<MyTreeNode> childrenList = node.getChildren();
            if (IterUtil.isEmpty(childrenList)) {
                childrenList = new ArrayList<>();
                node.setChildren(childrenList);
            }
            childrenList.addAll(sourceMap.get(targetTableName));
        }

        List<String> subSourceTableList = node.getSourceTableName();
        boolean isRoot = true;
        for (String subSourceTable : subSourceTableList) {
            if (targetMap.containsKey(subSourceTable)) {
                isRoot = false;
                break;
            }
        }
        if (isRoot) {
            rootList.add(node);
        }
    }
    return rootList;
}

部分效果圖展示：

自動化生成的任務定義

file
自動化生成的作業流定義圖
file
增量運行結果
file

自動化實作總結

數倉調度任務的秒級上線與切換
通過該方式將數倉開發與DolphinScheduler解耦，實作了整體資料調度任務的秒級上線與切換，這樣，我們可以快速復制標品化資料建模，極大地簡化了實施的難度，
血緣進行的任務關聯與生成
通過血緣進行的任務關聯與生成，大大提升了整體的資源利用率，也減少了人工的投入和成本，
血緣監控和管理
通過血緣監控和管理，可以幫助我們及時發現和糾正任務執行程序中的問題和錯誤，保證資料質量和準確性，

五、未來規劃

離在線統一調度 :實作基于Apache SeaTunnel的離線與實時資料同步調度，使得兩個場景在一個平臺完成，
應用基線管理：根據各任務的上下游依賴構建資料應用基線，并監控各目標任務及其依賴任務是否按時成功運行，以確保目標任務的準時產出，
任務指標監控：支持實時查看每個組件的指標，例如輸入記錄數、輸出記錄數和執行時間等，
資料血緣關系：上報資料源、目標表、欄位等資訊，構建資料血緣關系圖，方便追溯資料的來源和影響，
資源檔案版本管理：支持資源檔案等的簡單多版本管理，可以查看歷史版本、回滾到指定版本等，

六、總結與致謝

通過使用DolphinScheduler替換原有的調度工具，使得資料開發運維實作了平臺統一化，基于Apache DolphinScheduler提供的強大集成擴展插件能力大大提升了資料集成與資料開發的效率，
DolphinScheduler自帶的告警管理功能非常全面，配合此功能我們建立了運維值班制度以及微信告警通知的功能，故障發生時，運維人員可以第一時間收到報警通知，有效提高了故障的感知能力，
基于DolphinScheduler調度技術方案在多個專案中的優異表現，使得我們更好的推動了公司的資料驅動戰略，從實踐中沉淀出公司的資料實施SOP，為各個客戶、業務部門提供了及時、準確、全面的資料分析和決策支持，
我們第一次部署時使用的是3.0.0 版本，目前社區已經發布了 3.1.7 迭代速度非常快，如果你們的專案正在選型調度工具，我們強烈建議使用DolphinScheduler，加入社區進群會有非常多的前輩、大佬帶你起飛，DolphinScheduler 值得大力推薦，希望大家都能從中受益，祝愿DolphinScheduler生態越來越繁榮，越來越好！

本文由白鯨開源提供發布支持！

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

標籤：大數據

上一篇：什么是hive的高級分組聚合，它的用法和注意事項以及性能分析

下一篇：返回列表