2. 程式人生 > 其它 >DataX學習指南（二）--外掛開發 DataX學習指南（一）--基礎介紹

DataX學習指南（二）--外掛開發 DataX學習指南（一）--基礎介紹

阿新 • • 發佈：2022-04-21

DataX為什麼採用外掛機制？　　

　　從設計之初，DataX就把異構資料來源同步作為自身的使命，為了應對不同資料來源的差異、同時提供一致的同步原語和擴充套件能力，DataX自然而然地採用了框架 + 外掛 的模式：

  • 外掛只需關心資料的讀取或者寫入本身。
  • 而同步的共性問題，比如：型別轉換、效能、統計，則交由框架來處理。

　　作為外掛開發人員，則需要關注兩個問題：

  1. 資料來源本身的讀寫資料正確性。
  2. 如何與框架溝通、合理正確地使用框架。

如何編寫外掛？

　　外掛開發者不用關心太多，基本只需要關注特定系統讀和寫，以及自己的程式碼在邏輯上是怎樣被執行的，哪一個方法是在什麼時候被呼叫的。在此之前，需要明確以下概念：

  • Job: Job是DataX用以描述從一個源頭到一個目的端的同步作業，是DataX資料同步的最小業務單元。比如：從一張mysql的表同步到odps的一個表的特定分割槽。
  • Task: Task是為最大化而把Job拆分得到的最小執行單元。比如：讀一張有1024個分表的mysql分庫分表的Job，拆分成1024個讀Task，用若干個併發執行。
  • TaskGroup: 描述的是一組Task集合。在同一個TaskGroupContainer執行下的Task集合稱之為TaskGroup
  • JobContainer: Job執行器，負責Job全域性拆分、排程、前置語句和後置語句等工作的工作單元。類似Yarn中的JobTracker
  • TaskGroupContainer: TaskGroup執行器，負責執行一組Task的工作單元，類似Yarn中的TaskTracker。

　　簡而言之， Job拆分成Task，在分別在框架提供的容器中執行，外掛只需要實現Job和Task兩部分邏輯。（框架體系介紹見DataX學習指南（一）--基礎介紹）

定製外掛開發（二次開發）

　　在進行二次開發之前，需要到GitHub上下載原始碼。開發前將datax的開發外掛的手冊認真觀看一遍，對開發有幫助的。地址：https://github.com/alibaba/DataX/blob/master/dataxPluginDev.md

　　下載完後，通過ideaJ開啟，目錄結構如下（標準的分散式包結構），然後就是漫長的導包過程。

　　

 

 　　通過上面的目錄結構，每類資料來源都存在一對Readere和Writer。這些是開源專案已經寫好的，如我們需要二次開發，最方便的就是借鑑已有的業務邏輯，然後根據時間需求定製開發reader/writer（並非每次都得同時開發reader/writer）。

 　　新建一個模組,比如我這叫testreader,然後從其他模組複製plugin.json和plugin_job_template.json這兩個檔案，並做對應修改。（我這裡reader實現的是從文字讀取內容）

　　

 

 　　然後就是引入pom.xml檔案，可以將任意一個專案的pom檔案複製過來，然後根據自己的時間使用情況，修改/引入自己所需的依賴檔案。最後將本模組加入到datax的父pom檔案。

　　Reader/Job內方法的介紹在外掛開發手冊介紹得很詳細了，這裡我就不累述了，直接貼上開發的TestReader 

package com.alibaba.datax.plugin.reader.testreader;

import com.alibaba.datax.common.exception.DataXException;
import com.alibaba.datax.common.plugin.RecordSender;
import com.alibaba.datax.common.spi.Reader;
import com.alibaba.datax.common.util.Configuration;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderErrorCode;
import com.alibaba.datax.plugin.unstructuredstorage.reader.UnstructuredStorageReaderUtil;
import com.google.common.collect.Sets;

import org.apache.commons.io.Charsets;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.nio.charset.UnsupportedCharsetException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;

/**
 * jackpot
 */
public class TestReader extends Reader {
    public static class Job extends Reader.Job {
        private static final Logger LOG = LoggerFactory.getLogger(Job.class);

        private Configuration originConfig = null;

        private List<String> path = null;

        private List<String> sourceFiles;

        private Map<String, Pattern> pattern;

        private Map<String, Boolean> isRegexPath;

        @Override
        public void init() {
            this.originConfig = this.getPluginJobConf();
            this.pattern = new HashMap<String, Pattern>();
            this.isRegexPath = new HashMap<String, Boolean>();
            this.validateParameter();
        }

        private void validateParameter() {
            // Compatible with the old version, path is a string before
            String pathInString = this.originConfig.getNecessaryValue(Key.PATH,
                    TxtFileReaderErrorCode.REQUIRED_VALUE);
            if (StringUtils.isBlank(pathInString)) {
                throw DataXException.asDataXException(
                        TxtFileReaderErrorCode.REQUIRED_VALUE,
                        "您需要指定待讀取的源目錄或檔案");
            }
            if (!pathInString.startsWith("[") && !pathInString.endsWith("]")) {
                path = new ArrayList<String>();
                path.add(pathInString);
            } else {
                path = this.originConfig.getList(Key.PATH, String.class);
                if (null == path || path.size() == 0) {
                    throw DataXException.asDataXException(
                            TxtFileReaderErrorCode.REQUIRED_VALUE,
                            "您需要指定待讀取的源目錄或檔案");
                }
            }

            String encoding = this.originConfig
                    .getString(
                            com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING,
                            com.alibaba.datax.plugin.unstructuredstorage.reader.Constant.DEFAULT_ENCODING);
            if (StringUtils.isBlank(encoding)) {
                this.originConfig
                        .set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING,
                                com.alibaba.datax.plugin.unstructuredstorage.reader.Constant.DEFAULT_ENCODING);
            } else {
                try {
                    encoding = encoding.trim();
                    this.originConfig
                            .set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.ENCODING,
                                    encoding);
                    Charsets.toCharset(encoding);
                } catch (UnsupportedCharsetException uce) {
                    throw DataXException.asDataXException(
                            TxtFileReaderErrorCode.ILLEGAL_VALUE,
                            String.format("不支援您配置的編碼格式 : [%s]", encoding), uce);
                } catch (Exception e) {
                    throw DataXException.asDataXException(
                            TxtFileReaderErrorCode.CONFIG_INVALID_EXCEPTION,
                            String.format("編碼配置異常, 請聯絡我們: %s", e.getMessage()),
                            e);
                }
            }

            // column: 1. index type 2.value type 3.when type is Date, may have
            // format
            List<Configuration> columns = this.originConfig
                    .getListConfiguration(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN);
            // handle ["*"]
            if (null != columns && 1 == columns.size()) {
                String columnsInStr = columns.get(0).toString();
                if ("\"*\"".equals(columnsInStr) || "'*'".equals(columnsInStr)) {
                    this.originConfig
                            .set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COLUMN,
                                    null);
                    columns = null;
                }
            }

            if (null != columns && columns.size() != 0) {
                for (Configuration eachColumnConf : columns) {
                    eachColumnConf
                            .getNecessaryValue(
                                    com.alibaba.datax.plugin.unstructuredstorage.reader.Key.TYPE,
                                    TxtFileReaderErrorCode.REQUIRED_VALUE);
                    Integer columnIndex = eachColumnConf
                            .getInt(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.INDEX);
                    String columnValue = eachColumnConf
                            .getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.VALUE);

                    if (null == columnIndex && null == columnValue) {
                        throw DataXException.asDataXException(
                                TxtFileReaderErrorCode.NO_INDEX_VALUE,
                                "由於您配置了type, 則至少需要配置 index 或 value");
                    }

                    if (null != columnIndex && null != columnValue) {
                        throw DataXException.asDataXException(
                                TxtFileReaderErrorCode.MIXED_INDEX_VALUE,
                                "您混合配置了index, value, 每一列同時僅能選擇其中一種");
                    }
                    if (null != columnIndex && columnIndex < 0) {
                        throw DataXException.asDataXException(
                                TxtFileReaderErrorCode.ILLEGAL_VALUE, String
                                        .format("index需要大於等於0, 您配置的index為[%s]",
                                                columnIndex));
                    }
                }
            }

            // only support compress types
            String compress = this.originConfig
                    .getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COMPRESS);
            if (StringUtils.isBlank(compress)) {
                this.originConfig
                        .set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COMPRESS,
                                null);
            } else {
                Set<String> supportedCompress = Sets
                        .newHashSet("gzip", "bzip2", "zip");
                compress = compress.toLowerCase().trim();
                if (!supportedCompress.contains(compress)) {
                    throw DataXException
                            .asDataXException(
                                    TxtFileReaderErrorCode.ILLEGAL_VALUE,
                                    String.format(
                                            "僅支援 gzip, bzip2, zip 檔案壓縮格式 , 不支援您配置的檔案壓縮格式: [%s]",
                                            compress));
                }
                this.originConfig
                        .set(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.COMPRESS,
                                compress);
            }

            String delimiterInStr = this.originConfig
                    .getString(com.alibaba.datax.plugin.unstructuredstorage.reader.Key.FIELD_DELIMITER);
            // warn: if have, length must be one
            if (null != delimiterInStr && 1 != delimiterInStr.length()) {
                throw DataXException.asDataXException(
                        UnstructuredStorageReaderErrorCode.ILLEGAL_VALUE,
                        String.format("僅僅支援單字元切分, 您配置的切分為 : [%s]",
                                delimiterInStr));
            }

        }

        @Override
        public void prepare() {
            LOG.debug("prepare() begin...");
            // warn:make sure this regex string
            // warn:no need trim
            for (String eachPath : this.path) {
                String regexString = eachPath.replace("*", ".*").replace("?",
                        ".?");
                Pattern patt = Pattern.compile(regexString);
                this.pattern.put(eachPath, patt);
                this.sourceFiles = this.buildSourceTargets();
            }

            LOG.info(String.format("您即將讀取的檔案數為: [%s]", this.sourceFiles.size()));
        }

        @Override
        public void post() {
        }

        @Override
        public void destroy() {
        }

        // warn: 如果源目錄為空會報錯，拖空目錄意圖=>空檔案顯示指定此意圖
        @Override
        public List<Configuration> split(int adviceNumber) {
            LOG.debug("split() begin...");
            List<Configuration> readerSplitConfigs = new ArrayList<Configuration>();

            // warn:每個slice拖且僅拖一個檔案,
            // int splitNumber = adviceNumber;
            int splitNumber = this.sourceFiles.size();
            if (0 == splitNumber) {
                throw DataXException.asDataXException(
                        TxtFileReaderErrorCode.EMPTY_DIR_EXCEPTION, String
                                .format("未能找到待讀取的檔案,請確認您的配置項path: %s",
                                        this.originConfig.getString(Key.PATH)));
            }

            List<List<String>> splitedSourceFiles = this.splitSourceFiles(
                    this.sourceFiles, splitNumber);
            for (List<String> files : splitedSourceFiles) {
                Configuration splitedConfig = this.originConfig.clone();
                splitedConfig.set(Constant.SOURCE_FILES, files);
                readerSplitConfigs.add(splitedConfig);
            }
            LOG.debug("split() ok and end...");
            return readerSplitConfigs;
        }

        // validate the path, path must be a absolute path
        private List<String> buildSourceTargets() {
            // for eath path
            Set<String> toBeReadFiles = new HashSet<String>();
            for (String eachPath : this.path) {
                int endMark;
                for (endMark = 0; endMark < eachPath.length(); endMark++) {
                    if ('*' != eachPath.charAt(endMark)
                            && '?' != eachPath.charAt(endMark)) {
                        continue;
                    } else {
                        this.isRegexPath.put(eachPath, true);
                        break;
                    }
                }

                String parentDirectory;
                if (BooleanUtils.isTrue(this.isRegexPath.get(eachPath))) {
                    int lastDirSeparator = eachPath.substring(0, endMark)
                            .lastIndexOf(IOUtils.DIR_SEPARATOR);
                    parentDirectory = eachPath.substring(0,
                            lastDirSeparator + 1);
                } else {
                    this.isRegexPath.put(eachPath, false);
                    parentDirectory = eachPath;
                }
                this.buildSourceTargetsEathPath(eachPath, parentDirectory,
                        toBeReadFiles);
            }
            return Arrays.asList(toBeReadFiles.toArray(new String[0]));
        }

        private void buildSourceTargetsEathPath(String regexPath,
                String parentDirectory, Set<String> toBeReadFiles) {
            // 檢測目錄是否存在，錯誤情況更明確
            try {
                File dir = new File(parentDirectory);
                boolean isExists = dir.exists();
                if (!isExists) {
                    String message = String.format("您設定的目錄不存在 : [%s]",
                            parentDirectory);
                    LOG.error(message);
                    throw DataXException.asDataXException(
                            TxtFileReaderErrorCode.FILE_NOT_EXISTS, message);
                }
            } catch (SecurityException se) {
                String message = String.format("您沒有許可權檢視目錄 : [%s]",
                        parentDirectory);
                LOG.error(message);
                throw DataXException.asDataXException(
                        TxtFileReaderErrorCode.SECURITY_NOT_ENOUGH, message);
            }

            directoryRover(regexPath, parentDirectory, toBeReadFiles);
        }

        private void directoryRover(String regexPath, String parentDirectory,
                Set<String> toBeReadFiles) {
            File directory = new File(parentDirectory);
            // is a normal file
            if (!directory.isDirectory()) {
                if (this.isTargetFile(regexPath, directory.getAbsolutePath())) {
                    toBeReadFiles.add(parentDirectory);
                    LOG.info(String.format(
                            "add file [%s] as a candidate to be read.",
                            parentDirectory));

                }
            } else {
                // 是目錄
                try {
                    // warn:對於沒有許可權的目錄,listFiles 返回null，而不是丟擲SecurityException
                    File[] files = directory.listFiles();
                    if (null != files) {
                        for (File subFileNames : files) {
                            directoryRover(regexPath,
                                    subFileNames.getAbsolutePath(),
                                    toBeReadFiles);
                        }
                    } else {
                        // warn: 對於沒有許可權的檔案，是直接throw DataXException
                        String message = String.format("您沒有許可權檢視目錄 : [%s]",
                                directory);
                        LOG.error(message);
                        throw DataXException.asDataXException(
                                TxtFileReaderErrorCode.SECURITY_NOT_ENOUGH,
                                message);
                    }

                } catch (SecurityException e) {
                    String message = String.format("您沒有許可權檢視目錄 : [%s]",
                            directory);
                    LOG.error(message);
                    throw DataXException.asDataXException(
                            TxtFileReaderErrorCode.SECURITY_NOT_ENOUGH,
                            message, e);
                }
            }
        }

        // 正則過濾
        private boolean isTargetFile(String regexPath, String absoluteFilePath) {
            if (this.isRegexPath.get(regexPath)) {
                return this.pattern.get(regexPath).matcher(absoluteFilePath)
                        .matches();
            } else {
                return true;
            }

        }

        private <T> List<List<T>> splitSourceFiles(final List<T> sourceList,
                int adviceNumber) {
            List<List<T>> splitedList = new ArrayList<List<T>>();
            int averageLength = sourceList.size() / adviceNumber;
            averageLength = averageLength == 0 ? 1 : averageLength;

            for (int begin = 0, end = 0; begin < sourceList.size(); begin = end) {
                end = begin + averageLength;
                if (end > sourceList.size()) {
                    end = sourceList.size();
                }
                splitedList.add(sourceList.subList(begin, end));
            }
            return splitedList;
        }

    }

    public static class Task extends Reader.Task {
        private static Logger LOG = LoggerFactory.getLogger(Task.class);

        private Configuration readerSliceConfig;
        private List<String> sourceFiles;

        @Override
        public void init() {
            this.readerSliceConfig = this.getPluginJobConf();
            this.sourceFiles = this.readerSliceConfig.getList(
                    Constant.SOURCE_FILES, String.class);
        }

        @Override
        public void prepare() {

        }

        @Override
        public void post() {

        }

        @Override
        public void destroy() {

        }

        @Override
        public void startRead(RecordSender recordSender) {
            LOG.debug("start read source files...");
            for (String fileName : this.sourceFiles) {
                LOG.info(String.format("reading file : [%s]", fileName));
                InputStream inputStream;
                try {
                    inputStream = new FileInputStream(fileName);
                    UnstructuredStorageReaderUtil.readFromStream(inputStream,
                            fileName, this.readerSliceConfig, recordSender,
                            this.getTaskPluginCollector());
                    recordSender.flush();
                } catch (FileNotFoundException e) {
                    // warn: sock 檔案無法read,能影響所有檔案的傳輸,需要使用者自己保證
                    String message = String
                            .format("找不到待讀取的檔案 : [%s]", fileName);
                    LOG.error(message);
                    throw DataXException.asDataXException(
                            TxtFileReaderErrorCode.OPEN_FILE_ERROR, message);
                }
            }
            LOG.debug("end read source files...");
        }

    }
}

　

相關推薦

DataX學習指南（二）--外掛開發 DataX學習指南（一）--基礎介紹

DataX為什麼採用外掛機制？　　 　　從設計之初，DataX就把異構資料來源同步作為自身的使命，為了應對不同資料來源的差異、同時提供一致的同步原語和擴充套件能力，DataX自然而然地採用了框架 + 外掛 的模式：

【外掛開發】VSCode外掛開發全攻略（二）HelloWord

文章轉載於：https://www.cnblogs.com/liuxianan/p/vscode-plugin-hello-world.html 寫著前面 學習一門新的語言或者生態首先肯定是從HelloWord開始。

併發程式設計學習筆記（二十六、ConcurrentHashMap，Java8 HashMap簡述）

目錄： 學習準備 類核心屬性、內部類、建構函式介紹 雜湊衝突（雜湊碰撞） put()方法原始碼分析

【外掛開發】VSCode外掛開發全攻略（七）WebView

什麼是Webview 大家都知道，整個VSCode編輯器就是一張大的網頁，其實，我們還可以在Visual Studio Code中建立完全自定義的、可以間接和nodejs通訊的特殊網頁（通過一個acquireVsCodeApi特殊方法），這個網頁就叫WebV

【外掛開發】VSCode外掛開發全攻略（四）命令、選單、快捷鍵

文章轉載於：https://www.cnblogs.com/liuxianan/p/vscode-plugin-command-and-menu.html 命令 我們在前面HelloWord章節中已經提到了命令寫法，這裡再重溫一下。

【外掛開發】VSCode外掛開發全攻略（五）跳轉到定義、自動補全、懸停提示

跳轉到定義 跳轉到定義其實很簡單，通過vscode.languages.registerDefinitionProvider註冊一個provider，這個provider如果返回了new vscode.Location()就表示當前游標所在單詞支援跳轉，並且跳轉到對應location。

【外掛開發】VSCode外掛開發全攻略（六）開發除錯技巧

文章轉載於：http://blog.haoji.me/vscode-plugin-develop-tips.html 前言 在介紹完一些比較簡單的內容點之後，我覺得有必要先和大家介紹一些開發中遇到的一些細節問題以及技巧，特別是後面一章節將要介紹WebView的知

課時二：外掛開發

課時二：外掛開發 參考文章： 瑞泰資訊 微軟MVP(15-18)羅勇 CG_Technology 基本概念 D365平臺與傳統平臺

視訊投影（二維視訊投影到三維模型上）

視訊投影分享 目錄視訊投影分享前置知識物體與紋理物體組成紋理對映3D 轉換1. 存在一個物體2. 存在一個相機，用於觀察物體3. 根據相機引數的不同，得到不同的結構4. 最終得到規範立方體（canonical view volume）5.

音視訊開發進階指南（二）

1. 　　在ffplay中音畫同步的實現方式其實有三種，分別是：　　 以音訊為主時間軸作為同步源；（ffplay的預設方式），ubuntu16下測試偶有卡頓，效果比下面兩種的好

solr學習之（二）_在solr4.2中部署IKAnalyzer中文分詞外掛

前幾天弄過solr的中文分詞外掛，當時看了一個部落格，一步步來，感覺挺簡單的，就沒在意，今天覺得應該一步一步把這個細節知識點都過一遍。

核心模組開發學習筆記（二）

技術標籤：驅動 在開發板上如何編譯 1.開發板上怎麼rmmod 將Makefile裡ubuntu的編譯系統的路徑修改成開發板上對應的核心原始碼的路徑. 將*.ko複製到開發板的根檔案系統，執行insmod就可以載入了。 執行lsmod | g

《SystemVerilog驗證測試平臺編寫指南》學習筆記——隨機化（二）

技術標籤：SystemVerilog晶片verilogsystemverilog 一、解的概率1、沒有約束的類2、關係操作3、關係操作和雙向約束4、使用solve...before約束引導概率分佈

《SystemVerilog驗證測試平臺編寫指南》學習筆記——連線設計和測試平臺（二）

技術標籤：SystemVerilog 一、介面的驅動和取樣1、介面同步2、介面訊號取樣3、介面訊號驅動4、通過時鐘塊驅動介面訊號5、介面中的雙向訊號6、為什麼在程式中不允許使用always塊7、時鐘發生器

Android開發學習手記：從零開始學習android開發（二）：嘗試上手第一個App：HelloWorld

技術標籤：我的安卓開發學習手記android （By Mokiki） “Hello World”，一定是每個程式設計師都不陌生的詞，我們從這裡開始，正式進入Android開發的學習旅程。

數學建模學習筆記（二十一）時間序列小白指南（下）

技術標籤：數學建模學習筆記 上兩篇整理了ARIMA的建模和程式設計，此篇再加以拓展，考慮季節性

ESP32S2開發學習之路--新建工程（二）

技術標籤：esp32 ESP32S2開發學習之路--新建工程 檔案構成MakefileCMakeLists.txtmain-CMakeLists.txtcomponent.mkdemo.c

Django搭建開發‘學習筆記’專案二（建立應用程式）

Django專案由一系列應用程式組成，它們協同工作讓專案成為一個整體。 進入目錄learn_log，並激活虛擬環境，再執行命令python3 manage.py startapp learning_logs，建立應用程式

VUE移動端音樂APP學習【二十五】：歌曲列表元件開發（二）

實現點選垃圾桶，清除所有歌曲列表功能 和之前清除所有搜尋歷史列表一樣，引用confirm元件攔截提示使用者的操作

Vue學習（二十）外掛

Vue.use() 方法接收一個引數。這個引數必須具有install方法。Vue.use函式內部會呼叫引數的install方法。