數(shù)據(jù)集成數(shù)據(jù)同步任務(wù)調(diào)度后端【免費(fèi)下載鏈接】datax-webDataX集成可視化頁(yè)面選擇數(shù)據(jù)源即可一鍵生成數(shù)據(jù)同步任務(wù)支持RDBMS、Hive、HBase、ClickHouse、MongoDB等數(shù)據(jù)源批量創(chuàng)建RDBMS數(shù)據(jù)同步任務(wù)集成開(kāi)源調(diào)度系統(tǒng)支持分布式、增量同步數(shù)據(jù)、實(shí)時(shí)查看運(yùn)行日志、監(jiān)控執(zhí)行器資源、KILL運(yùn)行進(jìn)程、數(shù)據(jù)源信息加密等。項(xiàng)目地址https://gitcode.com/gh_mirrors/da/datax-web點(diǎn)擊查看免費(fèi)下載DataX Web 是在阿里開(kāi)源同步引擎 DataX 之上開(kāi)發(fā)的分布式數(shù)據(jù)同步與調(diào)度平臺(tái)它把 DataX 的任務(wù) JSON 構(gòu)建、數(shù)據(jù)源管理、定時(shí)調(diào)度、日志查看與進(jìn)程終止全部搬到頁(yè)面上顯著降低 DataX 的使用門(mén)檻。本文以本倉(cāng)庫(kù) README.md 為骨架結(jié)合 datax-admin、datax-executor、datax-core 等模塊源碼系統(tǒng)講解 DataX Web 的架構(gòu)設(shè)計(jì)、環(huán)境要求、執(zhí)行器與調(diào)度中心配置、JSON 構(gòu)建、增量同步、動(dòng)態(tài)分區(qū)、任務(wù)運(yùn)行與監(jiān)控等完整鏈路讀者可據(jù)此快速上手并深入理解其底層實(shí)現(xiàn)。一、DataX Web 是什么DataX Web 是在 DataX 之上開(kāi)發(fā)的分布式數(shù)據(jù)同步工具提供簡(jiǎn)單易用的操作界面核心目標(biāo)是降低用戶使用 DataX 的學(xué)習(xí)成本、縮短任務(wù)配置時(shí)間、避免配置過(guò)程中出錯(cuò)。用戶通過(guò)頁(yè)面選擇數(shù)據(jù)源即可創(chuàng)建數(shù)據(jù)同步任務(wù)支持 RDBMS、Hive、HBase、ClickHouse、MongoDB 等數(shù)據(jù)源RDBMS 數(shù)據(jù)源可批量創(chuàng)建數(shù)據(jù)同步任務(wù)支持實(shí)時(shí)查看數(shù)據(jù)同步進(jìn)度與日志并提供終止同步功能同時(shí)集成并二次開(kāi)發(fā)了 xxl-job可根據(jù)時(shí)間、自增主鍵增量同步數(shù)據(jù)。任務(wù)執(zhí)行器支持集群部署支持執(zhí)行器多節(jié)點(diǎn)路由策略選擇支持超時(shí)控制、失敗重試、失敗告警、任務(wù)依賴以及執(zhí)行器 CPU、內(nèi)存、負(fù)載的監(jiān)控。項(xiàng)目后續(xù)規(guī)劃還包括更多數(shù)據(jù)源支持、數(shù)據(jù)轉(zhuǎn)換 UDF、表結(jié)構(gòu)同步、數(shù)據(jù)同步血緣等復(fù)雜業(yè)務(wù)場(chǎng)景見(jiàn) README.md。從倉(cāng)庫(kù)結(jié)構(gòu)看項(xiàng)目由四個(gè) Maven 子模塊組成職責(zé)邊界清晰模塊職責(zé)關(guān)鍵目錄datax-admin調(diào)度中心Web 管理端負(fù)責(zé)任務(wù)管理、數(shù)據(jù)源管理、JSON 構(gòu)建、調(diào)度觸發(fā)、日志與報(bào)表com.wugui.datax.admindatax-executor任務(wù)執(zhí)行器接收調(diào)度指令拉起 DataX 進(jìn)程并回傳日志與結(jié)果com.wugui.datax.executordatax-core公共核心提供執(zhí)行器、JobHandler、Glue、日志、線程模型等基礎(chǔ)能力com.wugui.datatx.coredatax-rpcRPC 通信與注冊(cè)中心舊版在old包新版含 netty、netty_http、hessian 序列化等com.wugui.datax.rpc二、系統(tǒng)要求與環(huán)境按 README.md 的系統(tǒng)要求部署前需確認(rèn)以下環(huán)境語(yǔ)言環(huán)境Java 8JDK 版本建議 1.8.201 以上Python 2.7如需支持 Python3需修改替換datax/bin下的三個(gè) python 文件替換文件位于 doc/datax-web/datax-python3 目錄內(nèi)含datax.py、dxprof.py、perftrace.py。操作系統(tǒng)MacOS、Windows、Linux。數(shù)據(jù)庫(kù)Mysql5.7。兩個(gè)核心服務(wù)的默認(rèn)端口與連接配置在 Spring Boot 配置文件中調(diào)度中心 datax-admin/src/main/resources/application.yml端口由server.port環(huán)境變量指定默認(rèn)注釋為 8080MySQL 連接通過(guò)DB_HOST、DB_PORT、DB_DATABASE、DB_USERNAME、DB_PASSWORD環(huán)境變量注入默認(rèn)庫(kù)名dataxweb連接池使用 HikariCPmaximum-pool-size: 10、minimum-idle: 5、connection-timeout: 30000。執(zhí)行器 datax-executor/src/main/resources/application.yml端口由executor.port指定默認(rèn) 9999調(diào)度中心地址由datax.job.admin.addresses指定默認(rèn)http://127.0.0.1:${datax.admin.port}JSON 臨時(shí)目錄由datax.executor.jsonpath指定DataX 腳本路徑由datax.pypath指定。三、核心功能特性一覽README.md 累計(jì)羅列了 41 項(xiàng)功能特性歸納為以下幾大類任務(wù)構(gòu)建與數(shù)據(jù)源通過(guò) Web 構(gòu)建 DataX JsonJSON 保存在數(shù)據(jù)庫(kù)中方便任務(wù)遷移與管理。數(shù)據(jù)源支持 Hive、MySQL、Oracle、PostgreSQL、SqlServer、HBase、MongoDB、ClickHouseJDBC 添加 Hive 數(shù)據(jù)源支持可在構(gòu)建 JSON 頁(yè)面選擇數(shù)據(jù)源生成 column 信息并簡(jiǎn)化配置。添加 HBase 數(shù)據(jù)源支持JSON 構(gòu)建可通過(guò) HBase 數(shù)據(jù)源獲取 hbaseConfig、column添加 MongoDB 數(shù)據(jù)源支持僅需選擇 collectionName 即可完成 JSON 構(gòu)建JSON 構(gòu)建增加 ClickHouse 數(shù)據(jù)源支持RDBMS 數(shù)據(jù)源增加批量任務(wù)創(chuàng)建功能選擇數(shù)據(jù)源、表即可根據(jù)模板批量生成 DataX 同步任務(wù)提供 24 類插件 DataX JSON 配置樣例。數(shù)據(jù)源信息加密存儲(chǔ)數(shù)據(jù)源管理對(duì)用戶名和密碼進(jìn)行加密JSON 文件中的用戶名密碼加密存儲(chǔ)執(zhí)行 DataX 任務(wù)時(shí)解密。調(diào)度與執(zhí)行支持 DataX 定時(shí)任務(wù)支持動(dòng)態(tài)修改任務(wù)狀態(tài)、啟動(dòng)/停止任務(wù)、終止運(yùn)行中任務(wù)即時(shí)生效。調(diào)度采用中心式設(shè)計(jì)支持集群部署任務(wù)分布式執(zhí)行執(zhí)行器支持集群部署執(zhí)行器周期性自動(dòng)注冊(cè)調(diào)度中心自動(dòng)發(fā)現(xiàn)并觸發(fā)執(zhí)行。路由策略第一個(gè)、最后一個(gè)、輪詢、隨機(jī)、一致性 HASH、最不經(jīng)常使用LFU、最近最久未使用LRU、故障轉(zhuǎn)移、忙碌轉(zhuǎn)移。阻塞處理策略單機(jī)串行默認(rèn)、丟棄后續(xù)調(diào)度、覆蓋之前調(diào)度。任務(wù)超時(shí)控制可自定義超時(shí)時(shí)間運(yùn)行超時(shí)主動(dòng)中斷任務(wù)對(duì)超時(shí)任務(wù) kill datax 進(jìn)程可配合重試策略避免網(wǎng)絡(luò)問(wèn)題導(dǎo)致的 DataX 卡死。任務(wù)失敗重試可自定義重試次數(shù)失敗告警默認(rèn)提供郵件方式同時(shí)預(yù)留擴(kuò)展接口。任務(wù)依賴支持配置子任務(wù)依賴父任務(wù)成功后自動(dòng)觸發(fā)子任務(wù)多個(gè)子任務(wù)用逗號(hào)分隔。任務(wù)類型由 DataX 任務(wù)擴(kuò)展到 Shell 任務(wù)、Python 任務(wù)、PowerShell 任務(wù)腳本類型任務(wù)支持停止功能。運(yùn)維與監(jiān)控Web 實(shí)時(shí)查看抽取日志類似 Jenkins 的日志控制臺(tái)輸出DataX 運(yùn)行記錄展示可頁(yè)面操作停止 DataX 作業(yè)。執(zhí)行器 CPU、內(nèi)存、負(fù)載監(jiān)控頁(yè)面圖形化展示運(yùn)行報(bào)表支持實(shí)時(shí)查看運(yùn)行數(shù)據(jù)與調(diào)度報(bào)表調(diào)度日期分布圖、調(diào)度成功分布圖。日志頁(yè)面增加 DataX 執(zhí)行結(jié)果統(tǒng)計(jì)數(shù)據(jù)公共字段創(chuàng)建時(shí)間、創(chuàng)建人、修改時(shí)間、修改者插入或更新時(shí)自動(dòng)填充對(duì)應(yīng) MybatisMetaObjectHandler.javaswagger 接口進(jìn)行 token 驗(yàn)證。四、快速開(kāi)始與部署README.md 提供了三個(gè)入口快速開(kāi)始指南userGuid.md。Linux 一鍵部署文檔doc/datax-web/datax-web-deploy.md。倉(cāng)庫(kù)另有 doc/datax-web/datax-web-deploy-V2.1.1.md 供參考。部署時(shí)優(yōu)先通過(guò)環(huán)境變量獲取 DataX 文件目錄集群部署時(shí)不用指定 JSON 及日志目錄頁(yè)面可配置 DataX 啟動(dòng) JVM 參數(shù)。五、執(zhí)行器配置基于 xxl-job5.1 執(zhí)行器列表與在線狀態(tài)README.md 描述了執(zhí)行器頁(yè)面的兩個(gè)關(guān)鍵區(qū)域調(diào)度中心 OnLine右側(cè)顯示在線的調(diào)度中心列表。任務(wù)執(zhí)行結(jié)束后將以 failover 的模式回調(diào)調(diào)度中心通知執(zhí)行結(jié)果避免回調(diào)的單點(diǎn)風(fēng)險(xiǎn)。執(zhí)行器列表顯示在線的執(zhí)行器列表可通過(guò) OnLine 機(jī)器 查看對(duì)應(yīng)執(zhí)行器的集群機(jī)器。5.2 執(zhí)行器屬性說(shuō)明README.md 給出了新增執(zhí)行器時(shí)的屬性定義AppName與 datax-executor/src/main/resources/application.yml 中datax.job.executor.appname默認(rèn)datax-executor保持一致。它是每個(gè)執(zhí)行器集群的唯一標(biāo)識(shí)執(zhí)行器會(huì)周期性以 AppName 為對(duì)象進(jìn)行自動(dòng)注冊(cè)調(diào)度中心通過(guò)該配置自動(dòng)發(fā)現(xiàn)注冊(cè)成功的執(zhí)行器供任務(wù)調(diào)度使用。名稱執(zhí)行器的可讀名稱AppName 受字母數(shù)字限制可讀性不強(qiáng)名稱用于提高可讀性。排序執(zhí)行器的排序任務(wù)新增等需要執(zhí)行器的地方會(huì)按該排序讀取可用執(zhí)行器列表。注冊(cè)方式調(diào)度中心獲取執(zhí)行器地址的方式。自動(dòng)注冊(cè)執(zhí)行器自動(dòng)注冊(cè)調(diào)度中心通過(guò)底層注冊(cè)表動(dòng)態(tài)發(fā)現(xiàn)執(zhí)行器機(jī)器地址手動(dòng)錄入人工手動(dòng)錄入執(zhí)行器地址多地址用逗號(hào)分隔。機(jī)器地址注冊(cè)方式為手動(dòng)錄入時(shí)有效支持人工維護(hù)執(zhí)行器地址信息。從源碼看執(zhí)行器向調(diào)度中心注冊(cè)時(shí)會(huì)上報(bào) CPU 使用率、內(nèi)存使用率與系統(tǒng)負(fù)載見(jiàn) AdminBizImpl.java 的registry方法這些數(shù)據(jù)正是監(jiān)控頁(yè)面圖形化的數(shù)據(jù)來(lái)源。5.3 路由策略的源碼實(shí)現(xiàn)README 列出的 9 種路由策略在 ExecutorRouteStrategyEnum.java 中均有對(duì)應(yīng)實(shí)現(xiàn)類位于com.wugui.datax.admin.core.route.strategy包下策略枚舉項(xiàng)實(shí)現(xiàn)類第一個(gè)FIRSTExecutorRouteFirst最后一個(gè)LASTExecutorRouteLast輪詢ROUNDExecutorRouteRound隨機(jī)RANDOMExecutorRouteRandom一致性 HASHCONSISTENT_HASHExecutorRouteConsistentHash最不經(jīng)常使用LEAST_FREQUENTLY_USEDExecutorRouteLFU最近最久未使用LEAST_RECENTLY_USEDExecutorRouteLRU故障轉(zhuǎn)移FAILOVERExecutorRouteFailover忙碌轉(zhuǎn)移BUSYOVERExecutorRouteBusyover枚舉還包含SHARDING_BROADCAST分片廣播其 router 為 null屬于特殊的廣播型路由。策略名稱通過(guò)I18nUtil國(guó)際化加載支持中英文界面。六、創(chuàng)建數(shù)據(jù)源與任務(wù)模板6.1 創(chuàng)建數(shù)據(jù)源README 中創(chuàng)建數(shù)據(jù)源是任務(wù)配置鏈路的第一步對(duì)應(yīng) README.md之后在 JSON 構(gòu)建頁(yè)面第 4 步和批量創(chuàng)建任務(wù)時(shí)使用。數(shù)據(jù)源信息安全值得注意v2.1.1 起數(shù)據(jù)源管理對(duì)用戶名和密碼進(jìn)行加密存儲(chǔ)v2.1.2 修改了加密算法并優(yōu)化代碼。加密實(shí)現(xiàn)見(jiàn) AESUtil.java基于 AES 算法密鑰算法AES、隨機(jī)數(shù)算法SHA1PRNG加密后的密文以 Base64 輸出對(duì)稱密鑰配置在 application.yml 的datasource.aes.key默認(rèn)AD42F6697B035B75。數(shù)據(jù)庫(kù)實(shí)體 JobDatasource.java 配合 AESEncryptHandler.java 實(shí)現(xiàn)字段級(jí)加解密。注意由于 2.1.1 與 2.1.2 加密方式變更2.1.1 版本不建議升級(jí)否則已加密的數(shù)據(jù)源會(huì)解密失敗如需升級(jí)請(qǐng)重建數(shù)據(jù)源與任務(wù)見(jiàn) README.md。數(shù)據(jù)源配置成功后還提供手動(dòng)測(cè)試功能對(duì)應(yīng)DatasourceQueryService及其實(shí)現(xiàn) DatasourceQueryServiceImpl.java查詢工具類在tool/query包下如 MySQLQueryTool、HiveQueryTool、HBaseQueryTool 等。6.2 創(chuàng)建任務(wù)模板創(chuàng)建任務(wù)模板README.md允許對(duì)常用任務(wù)配置模板在構(gòu)建完 JSON 之后可選擇關(guān)聯(lián)模板創(chuàng)建任務(wù)。模板管理由 JobTemplateController.java、JobTemplateService 支撐實(shí)體為 JobTemplate.java。七、構(gòu)建 JSON 腳本JSON 構(gòu)建是 DataX Web 的核心交互之一README.md流程如下步驟一、步驟二選擇第 6.1 節(jié)創(chuàng)建的數(shù)據(jù)源。JSON 構(gòu)建目前支持的數(shù)據(jù)源有hive、mysql、oracle、postgresql、sqlserver、hbase、mongodb、clickhouse其它數(shù)據(jù)源的 JSON 構(gòu)建仍在開(kāi)發(fā)中暫時(shí)需要手動(dòng)編寫(xiě)。字段映射選擇讀取字段與寫(xiě)入字段的映射關(guān)系。點(diǎn)擊構(gòu)建生成 JSON此時(shí)可選擇復(fù)制 JSON 然后創(chuàng)建任務(wù)選擇 DataX 任務(wù)將 JSON 粘貼到文本框也可以點(diǎn)擊選擇模板直接生成任務(wù)。JSON 構(gòu)建的后端實(shí)現(xiàn)集中在 DataxJsonService 與 DataxJsonServiceImpl.java插件化設(shè)計(jì)位于tool/datax包讀取插件tool/datax/readerMysqlReader、OracleReader、PostgresqlReader、SqlServerReader、HiveReader、HBaseReader、MongoDBReader、ClickHouseReader均繼承 BaseReaderPlugin.java。寫(xiě)入插件tool/datax/writerMysqlWriter、OraclelWriter、PostgresqllWriter、SqlServerlWriter、HiveWriter、HBaseWriter、MongoDBWriter、ClickHouseWriter均繼承 BaseWriterPlugin.java。統(tǒng)一的構(gòu)建入口為 DataxJsonHelper.java其單元測(cè)試見(jiàn) DataxJsonHelperTest.java。各數(shù)據(jù)源的類型元數(shù)據(jù)、查詢建表語(yǔ)句、生成列的細(xì)節(jié)分布在tool/meta如 MySQLDatabaseMeta、OracleDatabaseMeta、HiveDatabaseMeta、ClickHouseDataBaseMeta 等與tool/query中構(gòu)建 JSON 時(shí)會(huì)動(dòng)態(tài)獲取目標(biāo)表結(jié)構(gòu)與列信息。7.1 RDBMS JSON 構(gòu)建的增強(qiáng)細(xì)節(jié)v2.1.2 對(duì) RDBMS JSON 構(gòu)建做了多項(xiàng)增強(qiáng)增加postSql支持并支持構(gòu)建多個(gè)preSql、postSql。PostgreSql、SQLServer、Oracle 數(shù)據(jù)源 JSON 構(gòu)建增加 schema name 選擇。RDBMS 數(shù)據(jù)源增量抽取增加主鍵自增方式并優(yōu)化頁(yè)面參數(shù)配置。時(shí)間增量同步支持更多時(shí)間格式HIVE JSON 構(gòu)建增加頭尾選項(xiàng)參數(shù)。八、批量創(chuàng)建任務(wù)對(duì) RDBMS 數(shù)據(jù)源可以選擇數(shù)據(jù)源與表根據(jù)模板批量生成 DataX 同步任務(wù)README.md。批量構(gòu)建由 DataXBatchJsonBuildDto.java 承載請(qǐng)求參數(shù)批量任務(wù)創(chuàng)建邏輯在 JobProjectController 與 JobInfoController 中暴露極大縮短了多表同步的配置時(shí)間。九、任務(wù)創(chuàng)建與調(diào)度策略9.1 支持的任務(wù)類型任務(wù)創(chuàng)建支持四種類型README.mdDataX 任務(wù)、Shell 任務(wù)、Python 任務(wù)、PowerShell 任務(wù)。其中 DataX 任務(wù)由 ExecutorJobHandler.javaJobHandler(value executorJobHandler)執(zhí)行流程為根據(jù) JSON 生成臨時(shí)文件 → 拼接命令行并Runtime.exec啟動(dòng)進(jìn)程 → 記錄 DataX 進(jìn)程號(hào)并回調(diào)調(diào)度中心 → 雙線程分別解析標(biāo)準(zhǔn)輸出與錯(cuò)誤流 →process.waitFor()等待退出碼0 表示成功。9.2 阻塞處理策略阻塞處理策略用于調(diào)度過(guò)于密集、執(zhí)行器來(lái)不及處理時(shí)的取舍README.md單機(jī)串行調(diào)度請(qǐng)求進(jìn)入單機(jī)執(zhí)行器后進(jìn)入 FIFO 隊(duì)列并以串行方式運(yùn)行丟棄后續(xù)調(diào)度發(fā)現(xiàn)執(zhí)行器存在運(yùn)行的調(diào)度任務(wù)時(shí)本次請(qǐng)求被丟棄并標(biāo)記為失敗覆蓋之前調(diào)度發(fā)現(xiàn)執(zhí)行器存在運(yùn)行的調(diào)度任務(wù)時(shí)終止運(yùn)行中的任務(wù)并清空隊(duì)列然后運(yùn)行本次調(diào)度任務(wù)。增量同步建議將阻塞策略設(shè)置為丟棄后續(xù)調(diào)度或單機(jī)串行。README 特別提醒設(shè)置單機(jī)串行時(shí)應(yīng)合理設(shè)置重試次數(shù)遵循失敗重試的次數(shù) × 每次執(zhí)行時(shí)間 任務(wù)調(diào)度周期。例如任務(wù) 30 秒執(zhí)行一次、每次執(zhí)行需要 20 秒、設(shè)置重試 3 次若任務(wù)失敗第一個(gè)重試的時(shí)間段為 1577755680-1577756680重試任務(wù)未結(jié)束時(shí)新任務(wù)又開(kāi)啟新任務(wù)的時(shí)間段會(huì)是 1577755680-1577758680重試次數(shù)過(guò)多會(huì)導(dǎo)致數(shù)據(jù)重復(fù)README.md。9.3 超時(shí)、重試與告警任務(wù)增加超時(shí)時(shí)間對(duì)超時(shí)任務(wù) kill datax 進(jìn)程可配合重試策略避免網(wǎng)絡(luò)問(wèn)題導(dǎo)致的 DataX 卡死README.md。超時(shí) kill 邏輯見(jiàn) KillJob.java由 AdminBizImpl.java 在收到FAIL_TIMEOUT回調(diào)時(shí)觸發(fā)。失敗重試次數(shù)可自定義失敗告警默認(rèn)提供郵件方式并預(yù)留擴(kuò)展接口。任務(wù)依賴支持子任務(wù)父任務(wù)執(zhí)行成功后自動(dòng)觸發(fā)子任務(wù)執(zhí)行AdminBizImpl.java 中對(duì)子任務(wù)回調(diào)的處理。9.4 任務(wù)列表與運(yùn)行狀態(tài)任務(wù)列表頁(yè)面展示全部任務(wù)及其狀態(tài)支持動(dòng)態(tài)修改狀態(tài)、啟動(dòng)/停止任務(wù)、終止運(yùn)行中任務(wù)即時(shí)生效。任務(wù)實(shí)體為 JobInfo.java調(diào)度觸發(fā)鏈路由 JobTrigger.java、TriggerTypeEnum.java 與調(diào)度線程池 JobTriggerPoolHelper.java 支撐。十、增量同步時(shí)間自增與主鍵自增增量同步是 DataX Web 最有價(jià)值的特性之一完整配置說(shuō)明見(jiàn) doc/datax-web/increment-desc.mdREADME 中以鏈接形式提供了 增量參數(shù)設(shè)置 與 分區(qū)參數(shù)設(shè)置。10.1 按日期增量抽取頁(yè)面任務(wù)配置5 個(gè)步驟任務(wù)類型選DataX 任務(wù)輔助參數(shù)選擇時(shí)間自增配置增量開(kāi)始時(shí)間即 SQL 中查詢時(shí)間的開(kāi)始時(shí)間方便第一次全量同步。第一次同步完成后該時(shí)間被更新為上一次的任務(wù)觸發(fā)時(shí)間任務(wù)失敗不更新配置增量時(shí)間字段-DlastTime%s -DcurrentTime%s。解析如下increment-desc.md-D是 DataX 參數(shù)的標(biāo)識(shí)符必配-D后面的lastTime、currentTime是 DataX JSON 中 where 條件的時(shí)間字段標(biāo)識(shí)符必須與 JSON 中的變量名稱保持一致%s是項(xiàng)目用于替換時(shí)間的占位符必配且格式必須完全一致注意-DlastTime%s與-DcurrentTime%s中間有一個(gè)空格空格必須保留且只能是一個(gè)空格配置時(shí)間格式可選擇數(shù)據(jù)庫(kù)中時(shí)間的格式也可通過(guò) JSON 中配置 SQL 時(shí)間轉(zhuǎn)換函數(shù)處理。JSON 配置示例節(jié)選自 increment-desc.md{ job: { setting: { speed: { channel: 16 } }, content: [ { reader: { name: mysqlreader, parameter: { splitPk: id, username: root, password: root, column: [*], connection: [ { jdbcUrl: [jdbc:mysql://localhost:3306/test?characterEncodingutf8], querySql: [ select * from test_list where operationDate FROM_UNIXTIME(${lastTime}) and operationDate FROM_UNIXTIME(${currentTime}) ] } ] } }, writer: { name: mysqlwriter, parameter: { username: root, password: 123456, column: [*], batchSize: 4096, connection: [ { jdbcUrl: jdbc:mysql://localhost:3307/test?characterEncodingutf8, table: [test_list] } ] } } } ] } }querySql 解析關(guān)鍵點(diǎn)在${lastTime}、${currentTime}${}是 DataX 動(dòng)態(tài)參數(shù)的固定格式變量名必須與頁(yè)面配置-D中的標(biāo)識(shí)符一致。如果頁(yè)面時(shí)間類型選擇為時(shí)間戳而數(shù)據(jù)庫(kù)時(shí)間格式不是時(shí)間戳例如2019-11-26 11:40:57可以用FROM_UNIXTIME(${lastTime})進(jìn)行轉(zhuǎn)換increment-desc.md。10.2 按自增主鍵增量抽取頁(yè)面任務(wù)配置4 個(gè)步驟increment-desc.md任務(wù)類型選 DataX 任務(wù)輔助參數(shù)選擇主鍵自增配置增量主鍵開(kāi)始 ID即 SQL 中查詢 ID 的開(kāi)始 ID方便第一次全量同步。第一次同步完成后該 ID 被更新為上一次任務(wù)觸發(fā)時(shí)最大的 ID任務(wù)失敗不更新配置增量字段-DstartId%s -DendId%s-D為 DataX 參數(shù)標(biāo)識(shí)符必配startId、endId是 JSON 中 where 條件的 id 字段標(biāo)識(shí)符必須與 JSON 變量名一致endId是任務(wù)每次執(zhí)行時(shí)獲取的當(dāng)前表 maxId也是下一次任務(wù)的startId%s是占位符格式必須完全一致兩個(gè)-D參數(shù)間必須保留且僅有一個(gè)空格選擇 reader 數(shù)據(jù)源并配置需要同步數(shù)據(jù)的表名及該表的主鍵。JSON 配置示例節(jié)選自 increment-desc.md{ job: { setting: { speed: { channel: 3, byte: 1048576 }, errorLimit: { record: 0, percentage: 0.02 } }, content: [ { reader: { name: mysqlreader, parameter: { username: yRjwDFuoPKlqya9h9H2Amg, password: yRjwDFuoPKlqya9h9H2Amg, splitPk: , connection: [ { querySql: [select * from job_log where id ${startId} and id ${endId}], jdbcUrl: [jdbc:mysql://localhost:3306/datax_web] } ] } }, writer: { name: mysqlwriter, parameter: { username: mCFDp1IMsa0rHicbQohcA, password: PhYxJmA/nuBJD1OxKTRzZH8sxuRddOv83hdqDOVRi0, column: [id, job_group, job_id, job_desc, executor_address, executor_handler, executor_param, executor_sharding_param, executor_fail_retry_count, trigger_time, trigger_code, trigger_msg, handle_time, handle_code, handle_msg, alarm_status, process_id, max_id], connection: [ { table: [job_log], jdbcUrl: jdbc:mysql://47.98.125.243:3306/datax_web } ] } } } ] } }注意示例中 reader 的用戶名密碼為密文AES 加密后寫(xiě)入 JSON執(zhí)行 DataX 任務(wù)時(shí)由執(zhí)行器解密。querySql 解析select * from job_log where id ${startId} and id ${endId}startId、endId對(duì)應(yīng)頁(yè)面配置-DstartId%s -DendId%s中的標(biāo)識(shí)符。10.3 增量狀態(tài)的持久化源碼佐證增量游標(biāo)更新由調(diào)度中心在回調(diào)階段完成任務(wù)成功回調(diào)時(shí)AdminBizImpl.java 的updateIncrementParam方法根據(jù)增量類型調(diào)用主鍵自增jobInfoMapper.incrementIdUpdate(jobId, maxId)更新任務(wù)記錄中的incStartId時(shí)間自增jobInfoMapper.incrementTimeUpdate(jobId, triggerTime)更新incStartTime。Mapper 定義見(jiàn) JobInfoMapper.java。這正對(duì)應(yīng)文檔中第一次同步完成后時(shí)間/ID 被更新為上一次任務(wù)觸發(fā)時(shí)間/最大 ID任務(wù)失敗不更新的行為——因?yàn)橛螛?biāo)只在成功回調(diào)路徑上更新。參數(shù)拼接與替換的執(zhí)行器實(shí)現(xiàn)位于 BuildCommand.java時(shí)間自增IncrementTypeEnum.TIME默認(rèn)按 Timestamp 處理startTime 任務(wù)的增量開(kāi)始時(shí)間/1000、endTime 觸發(fā)時(shí)間/1000通過(guò)String.format(replaceParam, startTime, endTime)替換占位符若配置了時(shí)間格式replaceParamType則按SimpleDateFormat格式化后再替換。主鍵自增IncrementTypeEnum.IDstartId tgParam.getStartId()、endId tgParam.getEndId()同樣通過(guò)占位符替換。JVM 參數(shù)與動(dòng)態(tài)參數(shù)最終拼裝進(jìn)命令行數(shù)組python datax.py -j -Xms2G -Xmx2G -p -DlastTime... -DcurrentTime... datax.json見(jiàn) BuildCommand.java。增量類型枚舉定義在 datax-core 的 IncrementTypeEnum.java。十一、增量數(shù)據(jù)抽取到動(dòng)態(tài)分區(qū)README 中 分區(qū)參數(shù)設(shè)置 給出了增量動(dòng)態(tài)參數(shù)抽取數(shù)據(jù)到動(dòng)態(tài)分區(qū)的完整方案通過(guò)動(dòng)態(tài)參數(shù)配置指定 Hive 分區(qū)可配合增量實(shí)現(xiàn)增量數(shù)據(jù)動(dòng)態(tài)插入分區(qū)。DataX JSON 配置要點(diǎn)節(jié)選自 partition-dynamic-param.mdreader 使用 mysqlreaderquerySql中updateTime FROM_UNIXTIME(${lastTime}) and operationDate FROM_UNIXTIME(${currentTime})writer 使用 hdfswriterpath寫(xiě)為/user/hive/warehouse/offline.db/test_order/${partition}即分區(qū)目錄由動(dòng)態(tài)參數(shù)${partition}決定${lastTime}增量開(kāi)始時(shí)間定時(shí)任務(wù)啟動(dòng)后第一次的開(kāi)始時(shí)間為頁(yè)面輸入時(shí)間任務(wù)執(zhí)行成功后更新為上一次的任務(wù)觸發(fā)時(shí)間任務(wù)失敗不更新${currentTime}為任務(wù)觸發(fā)時(shí)間分區(qū)字段${partition}為固定格式不能自定義拼接結(jié)果示例-p-DlastTime1572537600 -DcurrentTime1579317145 -Dpartitiondatety2020-01-18JVM 啟動(dòng)參數(shù)拼接結(jié)果-j -Xms2G -Xmx2G。DataX 啟動(dòng)命令partition-dynamic-param.mdpython datax.py -j -Xms2G -Xmx2G -p -DlastTime1577009172 -DcurrentTime1579317145 -Dpartitiondatety2020-01-18 datax.json從源碼看分區(qū)參數(shù)由頁(yè)面解析為分區(qū)字段,時(shí)間偏移,時(shí)間格式三段逗號(hào)分隔執(zhí)行器在 BuildCommand.java 中按IncrementTypeEnum.PARTITION處理buildPartition取偏移量與格式通過(guò)DateUtil.addDays(new Date(), timeOffset)計(jì)算分區(qū)日期并拼接為字段日期值最終以-Dpartition...形式注入。十二、任務(wù)列表、運(yùn)行日志與進(jìn)程終止12.1 實(shí)時(shí)日志與執(zhí)行結(jié)果統(tǒng)計(jì)任務(wù)列表頁(yè)面支持點(diǎn)擊查看日志、實(shí)時(shí)獲取日志信息并可在頁(yè)面終止正在執(zhí)行的 DataX 進(jìn)程README.md。日志頁(yè)面還增加了 DataX 執(zhí)行結(jié)果統(tǒng)計(jì)數(shù)據(jù)README.md該統(tǒng)計(jì)由執(zhí)行器端的 AnalysisStatistics.java 解析 DataX 輸出流計(jì)算得出并通過(guò) ExecutorJobHandler.java 作為任務(wù)結(jié)果消息回傳。12.2 日志線程模型ExecutorJobHandler.java 的日志處理模型標(biāo)準(zhǔn)輸出用FutureTaskLogStatistics異步解析錯(cuò)誤流單獨(dú)起線程解析最終以退出碼判斷成敗0 成功非 0 失敗臨時(shí) JSON 文件在 finally 中刪除。日志文件與進(jìn)程的關(guān)聯(lián)通過(guò)ProcessCallbackThread將 DataX 進(jìn)程號(hào)回傳調(diào)度中心供頁(yè)面終止進(jìn)程使用。12.3 超時(shí)任務(wù) Kill調(diào)度中心對(duì)超時(shí)任務(wù)執(zhí)行 kill datax 進(jìn)程對(duì)應(yīng) README 第 31 條特性AdminBizImpl.java 在回調(diào)結(jié)果為FAIL_TIMEOUT且存在進(jìn)程號(hào)時(shí)調(diào)用KillJob.trigger(...)主動(dòng)終止。十三、執(zhí)行器資源監(jiān)控執(zhí)行器支持 CPU、內(nèi)存、負(fù)載的監(jiān)控頁(yè)面v2.1.2 起監(jiān)控頁(yè)面圖形化README.md。監(jiān)控?cái)?shù)據(jù)來(lái)自執(zhí)行器注冊(cè)時(shí)上報(bào)的cpuUsage、memoryUsage、loadAverage見(jiàn) AdminBizImpl.java采集端實(shí)現(xiàn)在 SystemUtils.java注冊(cè)與心跳線程見(jiàn) ExecutorRegistryThread.java。十四、用戶管理與權(quán)限admin 可以創(chuàng)建用戶、編輯用戶信息README.md。系統(tǒng)支持管理員、普通用戶兩種角色README 第 14 條特性用戶實(shí)體為 JobUser.java鑒權(quán)基于 JWT過(guò)濾器實(shí)現(xiàn)見(jiàn) JWTAuthenticationFilter.java 與 JWTAuthorizationFilter.java令牌生成與校驗(yàn)見(jiàn) JwtTokenUtils.java。十五、版本演進(jìn)v2.1.2 與 v2.1.1README 末尾記錄了最近的版本變更README.md可作為選型與升級(jí)依據(jù)v2.1.2 新增添加項(xiàng)目管理模塊可對(duì)任務(wù)分類管理RDBMS 數(shù)據(jù)源批量任務(wù)創(chuàng)建功能選數(shù)據(jù)源、表按模板批量生成任務(wù)JSON 構(gòu)建增加 ClickHouse 數(shù)據(jù)源支持執(zhí)行器 CPU、內(nèi)存、負(fù)載監(jiān)控頁(yè)面圖形化RDBMS 增量抽取增加主鍵自增方式并優(yōu)化頁(yè)面參數(shù)配置更換 MongoDB 數(shù)據(jù)源連接方式重構(gòu) HBase 數(shù)據(jù)源 JSON 構(gòu)建模塊腳本類型任務(wù)增加停止功能rdbms json 構(gòu)建增加 postSql支持構(gòu)建多個(gè) preSql、postSql合并 datax-registry 模塊到 datax-rpc 中數(shù)據(jù)源信息加密算法修改及代碼優(yōu)化時(shí)間增量同步支持更多時(shí)間格式日志頁(yè)面增加 DataX 執(zhí)行結(jié)果統(tǒng)計(jì)數(shù)據(jù)。v2.1.2 升級(jí)PostgreSql、SQLServer、Oracle 數(shù)據(jù)源 JSON 構(gòu)建增加 schema name 選擇DataX JSON 字段名稱與數(shù)據(jù)源關(guān)鍵詞一致問(wèn)題優(yōu)化任務(wù)管理頁(yè)面按鈕展示優(yōu)化日志管理頁(yè)面增加任務(wù)描述信息JSON 構(gòu)建前端 form 表單不能緩存數(shù)據(jù)問(wèn)題修復(fù)HIVE JSON 構(gòu)建增加頭尾選項(xiàng)參數(shù)。升級(jí)備注2.1.1 版本不建議直接升級(jí)——數(shù)據(jù)源信息加密方式變更會(huì)導(dǎo)致之前已加密的數(shù)據(jù)源解密失敗、任務(wù)運(yùn)行失敗如需升級(jí)請(qǐng)重建數(shù)據(jù)源與任務(wù)README.md。v2.1.1 新增HBase 數(shù)據(jù)源支持JSON 構(gòu)建可通過(guò) HBase 數(shù)據(jù)源獲取 hbaseConfig、columnMongoDB 數(shù)據(jù)源支持僅需選擇 collectionName 即可完成 JSON 構(gòu)建執(zhí)行器 CPU、內(nèi)存、負(fù)載監(jiān)控頁(yè)面24 類插件 DataX JSON 配置樣例公共字段創(chuàng)建時(shí)間、創(chuàng)建人、修改時(shí)間、修改者自動(dòng)填充swagger 接口 token 驗(yàn)證任務(wù)超時(shí)時(shí)間設(shè)置超時(shí) kill datax 進(jìn)程可配合重試策略避免網(wǎng)絡(luò)問(wèn)題導(dǎo)致的 DataX 卡死。v2.1.1 升級(jí)與修復(fù)數(shù)據(jù)源管理對(duì)用戶名和密碼加密提高安全性JSON 文件中的用戶名密碼加密執(zhí)行時(shí)解密頁(yè)面菜單整理、圖標(biāo)升級(jí)、提示信息優(yōu)化日志輸出取消項(xiàng)目類名等無(wú)關(guān)信息減小文件大小優(yōu)化大文件輸出與頁(yè)面展示logback 改為從 yml 獲取日志路徑配置修復(fù)任務(wù)日志過(guò)大時(shí)查看日志報(bào)錯(cuò)、請(qǐng)求超時(shí)的問(wèn)題。十六、開(kāi)源協(xié)議與社區(qū)DataX Web 采用 MIT LicenseCopyright (c) 2020 WeiYe產(chǎn)品開(kāi)源免費(fèi)并提供持續(xù)的社區(qū)技術(shù)支持個(gè)人或企業(yè)內(nèi)部可自由接入和使用README.md。項(xiàng)目歡迎貢獻(xiàn)可提交 Pull Request 修復(fù) bug或新建 Issue 討論新特性與變更README.md。前端工程為獨(dú)立的 datax-web-ui 項(xiàng)目。結(jié)語(yǔ)DataX Web 的定位非常清晰把 DataX 從手工編寫(xiě) JSON 命令行執(zhí)行升級(jí)為頁(yè)面勾選 集中調(diào)度 分布式執(zhí)行 可視化監(jiān)控的完整鏈路。從 README 的 41 項(xiàng)特性到本文對(duì)照的源碼實(shí)現(xiàn)可以看到它在數(shù)據(jù)源插件化、JSON 構(gòu)建、增量游標(biāo)持久化、超時(shí) Kill、執(zhí)行器自動(dòng)注冊(cè)與資源上報(bào)等方面都已形成閉環(huán)。若需進(jìn)一步深入建議依次閱讀 userGuid.md、doc/datax-web/datax-web-deploy.md、doc/datax-web/increment-desc.md 與 doc/datax-web/partition-dynamic-param.md并結(jié)合 datax-admin 與 datax-executor 源碼驗(yàn)證文中各條調(diào)用鏈。贊分享數(shù)據(jù)集成數(shù)據(jù)同步任務(wù)調(diào)度后端【免費(fèi)下載鏈接】datax-webDataX集成可視化頁(yè)面選擇數(shù)據(jù)源即可一鍵生成數(shù)據(jù)同步任務(wù)支持RDBMS、Hive、HBase、ClickHouse、MongoDB等數(shù)據(jù)源批量創(chuàng)建RDBMS數(shù)據(jù)同步任務(wù)集成開(kāi)源調(diào)度系統(tǒng)支持分布式、增量同步數(shù)據(jù)、實(shí)時(shí)查看運(yùn)行日志、監(jiān)控執(zhí)行器資源、KILL運(yùn)行進(jìn)程、數(shù)據(jù)源信息加密等。項(xiàng)目地址https://gitcode.com/gh_mirrors/da/datax-web點(diǎn)擊查看免費(fèi)下載相關(guān)推薦DataX 分區(qū)同步實(shí)戰(zhàn)基于 datax-web 的動(dòng)態(tài)分區(qū)參數(shù)配置與調(diào)度實(shí)現(xiàn)DataX 分區(qū)同步實(shí)戰(zhàn)基于 datax web 的動(dòng)態(tài)分區(qū)參數(shù)配置與調(diào)度實(shí)現(xiàn) 本文圍繞 datax web 倉(cāng)庫(kù)中的分區(qū)同步方案文檔 partition數(shù)據(jù)集成數(shù)據(jù)同步任務(wù)調(diào)度后端MediaFusion核心功能詳解從多源聚合到個(gè)性化流媒體體驗(yàn)MediaFusion核心功能詳解從多源聚合到個(gè)性化流媒體體驗(yàn) MediaFusion是一款終極開(kāi)源流媒體平臺(tái)通過(guò)多源內(nèi)容聚合與智能個(gè)性化推薦為用戶打造無(wú)python-OBD命令集詳解PIDs、解碼器與單位轉(zhuǎn)換實(shí)用手冊(cè)python OBD命令集詳解PIDs、解碼器與單位轉(zhuǎn)換實(shí)用手冊(cè) python OBD是一個(gè)強(qiáng)大的OBD II串行模塊用于讀取發(fā)動(dòng)機(jī)數(shù)據(jù)。本文將為你全面解析物聯(lián)網(wǎng)智能硬件上一篇PaddleOCR Android 部署實(shí)戰(zhàn)基于 ONNX Runtime 的 PP-OCRv6 移動(dòng)端 SDK 集成指南下一篇MkDocs Material 多語(yǔ)言配置實(shí)戰(zhàn)站點(diǎn)語(yǔ)言、語(yǔ)言選擇器與自定義翻譯創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考