DataX工具在新冠肺炎數(shù)據(jù)上報(bào)中的應(yīng)用

田翠姣，蘇義武

(武漢大學(xué)中南醫(yī)院，湖北 武漢 430071)

1 背 景

2020年2月15日國(guó)務(wù)院醫(yī)療救治組發(fā)文《關(guān)于做好武漢市新冠肺炎患者醫(yī)療數(shù)據(jù)報(bào)送工作的通知》，通知要求武漢市全部新冠肺炎定點(diǎn)收治醫(yī)院自2月15日起每天中午12點(diǎn)前上報(bào)新冠肺炎在院病例和出院病例的全量醫(yī)療數(shù)據(jù)。數(shù)據(jù)包括但不限于電子病歷數(shù)據(jù)、病程記錄數(shù)據(jù)、護(hù)理記錄數(shù)據(jù)、檢驗(yàn)信息數(shù)據(jù)、醫(yī)囑信息數(shù)據(jù)、重癥系統(tǒng)數(shù)據(jù)、影像報(bào)告系統(tǒng)數(shù)據(jù)、手術(shù)麻醉系統(tǒng)數(shù)據(jù)、病案首頁(yè)數(shù)據(jù)。

數(shù)據(jù)上報(bào)采用VPN+FTP方式，數(shù)據(jù)文件要求每張表數(shù)據(jù)存放到獨(dú)立的CSV格式文件，并且文件首行為數(shù)據(jù)表對(duì)應(yīng)字段名，每行的各個(gè)字段之間使用Tab符( )分隔，數(shù)據(jù)字符集要求UTF-8，最后文件用gz壓縮為單個(gè)壓縮文件[1]。

2 方案分析

按照要求對(duì)任務(wù)進(jìn)行了分析，上傳數(shù)據(jù)文件要求為CSV格式文件。CSV格式文件是一種通用的、相對(duì)簡(jiǎn)單的文件格式，在商業(yè)和科學(xué)中應(yīng)用廣泛，其特點(diǎn)是分隔的數(shù)據(jù)格式，字段之間的分割格式有逗號(hào)分隔和空格分隔兩種[2-3]。數(shù)據(jù)內(nèi)容方面主要涉及到9個(gè)信息系統(tǒng)，4種不同的數(shù)據(jù)庫(kù)，進(jìn)一步分解任務(wù)后得出需要上報(bào)的表格數(shù)量為22個(gè)(見表1)。

表1 上報(bào)數(shù)據(jù)信息數(shù)據(jù)庫(kù)格式

按照以往提取數(shù)據(jù)的方式，決定先用最熟悉的Oracle數(shù)據(jù)庫(kù)做測(cè)試，使用Oracle自帶的PL/SQL工具查詢出數(shù)據(jù)后，直接導(dǎo)出CSV文件。經(jīng)過測(cè)試發(fā)現(xiàn)這些CSV文件默認(rèn)使用逗號(hào)分隔符，字符集為GB2312，需要使用特定工具進(jìn)行轉(zhuǎn)換。此種方法不僅繁瑣，費(fèi)時(shí)費(fèi)力，而且會(huì)出現(xiàn)轉(zhuǎn)換失敗的情況。經(jīng)過不斷摸索，最后選擇使用DataX工具進(jìn)行數(shù)據(jù)批量提取的方法。

3 相關(guān)工具及安裝介紹

DataX是阿里巴巴集團(tuán)內(nèi)被廣泛使用的離線數(shù)據(jù)同步工具，能夠?qū)崿F(xiàn)包括MySQL、Oracle、SQL Server、MongoDB、HDFS、Hive、OceanBase、HBase、OTS、ODPS等各種異構(gòu)數(shù)據(jù)源之間高效的數(shù)據(jù)同步功能[4-6]。DataX將復(fù)雜的網(wǎng)狀的同步鏈路變成了星型數(shù)據(jù)鏈路，DataX作為中間傳輸載體負(fù)責(zé)連接各種數(shù)據(jù)源。當(dāng)需要接入一個(gè)新的數(shù)據(jù)源的時(shí)候，只需要將此數(shù)據(jù)源對(duì)接到DataX，便能跟已有的數(shù)據(jù)源做到無縫數(shù)據(jù)同步。更重要的是下載即可用，配置簡(jiǎn)單，支持Linux和Windows，只需要短短幾步就可以完成數(shù)據(jù)的傳輸[7]。

另外需要引入的工具是Python，Python是一種跨平臺(tái)的計(jì)算機(jī)程序設(shè)計(jì)語言，是一種面向?qū)ο蟮膭?dòng)態(tài)類型語言，最初被設(shè)計(jì)用于編寫自動(dòng)化腳本(shell)，隨著版本的不斷更新和語言新功能的添加，越來越多地被用于獨(dú)立的、大型項(xiàng)目的開發(fā)[8]。再加上Java運(yùn)行環(huán)境，組成了DataX+Python+Java環(huán)境，該文以Windows10+Datax3.0+Python2.7+jdk1.8組合環(huán)境進(jìn)行說明。

先安裝jdk環(huán)境并配置系統(tǒng)環(huán)境變量，然后缺省安裝Python軟件并配置系統(tǒng)環(huán)境變量，最后解壓Datax壓縮包到指定目錄(此處路徑為D:datax)，至此環(huán)境配置完成，整個(gè)安裝過程為純傻瓜式安裝，沒有特殊設(shè)置，簡(jiǎn)單易行。

4 DataX運(yùn)行原理

在DataX框架中，數(shù)據(jù)交換通過DataX進(jìn)行中轉(zhuǎn)，任何數(shù)據(jù)源只要和DataX連接上即可以和已實(shí)現(xiàn)的任意數(shù)據(jù)源同步。

DataX作為離線數(shù)據(jù)同步框架，采用Framework+plugins架構(gòu)構(gòu)建。將數(shù)據(jù)同步過程中的讀取和寫入過程抽象為內(nèi)部Reader/Writer插件，如圖1所示。

圖1 DataX框架結(jié)構(gòu)

Reader：數(shù)據(jù)采集模塊的抽象，負(fù)責(zé)采集數(shù)據(jù)源數(shù)據(jù)并發(fā)送給Framework。Writer：數(shù)據(jù)寫入模塊，不斷從Framework取數(shù)據(jù)并寫入目標(biāo)端。Framework：作為Reader和Writer的數(shù)據(jù)傳輸通道，處理緩沖流控、并發(fā)、數(shù)據(jù)轉(zhuǎn)換等核心技術(shù)問題。

框架提供了簡(jiǎn)單的接口與插件接入機(jī)制，只需要任意加上一種插件，就可以無縫對(duì)接其他數(shù)據(jù)源。開放式的框架也讓開發(fā)者可以在短時(shí)間內(nèi)開發(fā)一個(gè)新的插件，以快速支持新的數(shù)據(jù)源同步，以下結(jié)合此次數(shù)據(jù)上報(bào)應(yīng)用具體說明。

此外，DataX內(nèi)部運(yùn)行采用Job模式運(yùn)行機(jī)制(如圖2所示)，DataX完成單個(gè)數(shù)據(jù)同步的作業(yè)，稱之為Job，DataX接受到一個(gè)Job之后，將啟動(dòng)一個(gè)進(jìn)程來完成整個(gè)作業(yè)同步過程。DataX Job模塊是單個(gè)作業(yè)的中樞管理節(jié)點(diǎn)，承擔(dān)了數(shù)據(jù)清理、子任務(wù)切分(將單一作業(yè)計(jì)算轉(zhuǎn)化為多個(gè)子Task)、TaskGroup管理等功能。

圖2 DataX各功能模塊關(guān)系

DataX Job啟動(dòng)后，會(huì)根據(jù)不同的源端切分策略，將Job切分成多個(gè)小的Task(子任務(wù))，以便于并發(fā)執(zhí)行。Task便是DataX作業(yè)的最小單元，每一個(gè)Task都會(huì)負(fù)責(zé)一部分?jǐn)?shù)據(jù)的同步工作。

切分多個(gè)Task之后，DataX Job會(huì)調(diào)用Scheduler模塊，根據(jù)配置的并發(fā)數(shù)據(jù)量，將拆分成的Task重新組合，組裝成TaskGroup(任務(wù)組)。每一個(gè)TaskGroup負(fù)責(zé)以一定的并發(fā)運(yùn)行完畢分配好的所有Task，默認(rèn)單個(gè)任務(wù)組的并發(fā)數(shù)量為5。

每一個(gè)Task都由TaskGroup負(fù)責(zé)啟動(dòng)，Task啟動(dòng)后，會(huì)固定啟動(dòng)Reader→Channel→Writer的線程來完成任務(wù)同步工作。

DataX作業(yè)運(yùn)行起來之后，Job監(jiān)控并等待多個(gè)TaskGroup模塊任務(wù)完成，等待所有TaskGroup任務(wù)完成后Job成功退出。否則，異常退出，進(jìn)程退出值非0。

5 腳本編寫及運(yùn)行

DataX使用一個(gè)JSON文件來描述一個(gè)Job[9-10]，通過配置JSON格式的文件，使用Python命令就可以啟動(dòng)DataX執(zhí)行任務(wù)。這個(gè)Job本身配置并不復(fù)雜，Reader插件和Writer插件基本就是配置連接數(shù)據(jù)庫(kù)使用的連接串、用戶名、密碼、查詢表、字段名。接著，按照J(rèn)SON編碼規(guī)則[11-12]，編寫如下腳本，文件名為demo.json，存儲(chǔ)目錄為D:datax{

"job": {

"content": [

{

"reader": {

"name": "mysqlreader",

"parameter": {

//數(shù)據(jù)庫(kù)用戶名

"username": "db_user",

//數(shù)據(jù)庫(kù)密碼

"password": "db_pass",

//以下為查詢字段，若查詢?nèi)孔侄?，可以使?代表字段列表"

"column": ["patientno","deptcode","name","rptext"],

"connection": [

{

//數(shù)據(jù)源數(shù)據(jù)表名稱

"table": ["pacs_report"],

//數(shù)據(jù)庫(kù)連接串"，更多串配置參考官網(wǎng)示例

//MySQL數(shù)據(jù)庫(kù)連接串

//"jdbcUrl": ["jdbc:mysql://IP地址:3306/database"]

//Oracle數(shù)據(jù)庫(kù)連接串

//"jdbc:oracle:thin:@IP地址:1521/實(shí)例名",

//SQLServer數(shù)據(jù)庫(kù)連接串

"jdbc:sqlserver://IP地址:3433;DatabaseName=dbname"

}

]

}

}

"writer": {

"name": "txtfilewriter",

"parameter": {

//導(dǎo)出數(shù)據(jù)文件輸出路徑

"path": "D:\datax\myoutput ",

//導(dǎo)出數(shù)據(jù)文件名稱

"fileName": "pacs_report",

//數(shù)據(jù)內(nèi)容字符集

"encoding": "UTF-8",

//覆蓋之前同名文件模式

"writeMode": "truncate",

//日期格式

"dateFormat": "yyyy-MM-dd HH:mm:ss",

//導(dǎo)出數(shù)據(jù)文件格式

"fileFormat": "csv",

// csv分隔符為 ,即tab方式分割

"fieldDelimiter": " " ,

//壓縮格式gzip,能減少生成文件的大小

"compress": "gzip",

//表頭設(shè)置

"header":["patientno","deptcode","name","rptext"],

}

}

}

]

}

}

腳本運(yùn)行也非常簡(jiǎn)單，在CMD窗口模式下，輸入如下語句執(zhí)行：python D:dataxindatax.py D:dataxpacs_report.json，其中datax.py為python安裝后自帶的解析腳本。整個(gè)命令大意為使用python命令利用datax.py腳本執(zhí)行pacs_report.json配置的任務(wù)[13-14]。此時(shí)在JSON腳本中設(shè)置的目錄D:\datax\myoutput下會(huì)出現(xiàn)pacs_report.csv.gz壓縮文件，壓縮文件內(nèi)為pacs_report.csv文件。其余的21個(gè)文件，按照相同的方式配置job，最后將所有運(yùn)行腳本形成批處理文件，一鍵執(zhí)行后生成所需22個(gè)數(shù)據(jù)文件。

6 效果對(duì)比

經(jīng)過自己檢測(cè)，數(shù)據(jù)內(nèi)容、文件格式、分隔符、字符集、壓縮形式均符合上報(bào)要求，上報(bào)后經(jīng)國(guó)家醫(yī)政醫(yī)管局工程師檢驗(yàn)，亦沒有問題。效率方面，原先手工生成方式，每次都需要執(zhí)行大量的SQL語句，并且需要轉(zhuǎn)換、壓縮、校驗(yàn)，一個(gè)文件生成至少需要2分鐘，并且是在數(shù)據(jù)量較小的前提下，費(fèi)時(shí)費(fèi)力而且容易出錯(cuò)。使用基于DataX的工具環(huán)境后，實(shí)際執(zhí)行33萬條費(fèi)用明細(xì)數(shù)據(jù)全表導(dǎo)出，只需要27秒即完成，全部22個(gè)數(shù)據(jù)文件生成平均花費(fèi)3分鐘，極大提高了數(shù)據(jù)收集的效率。

7 擴(kuò)展應(yīng)用討論

7.1 疫情數(shù)據(jù)上報(bào)

隨著疫情發(fā)展，疫情數(shù)據(jù)報(bào)送內(nèi)容、對(duì)象、形式不斷變化，高峰時(shí)期對(duì)外報(bào)送報(bào)表數(shù)目多達(dá)20余個(gè)。如每日8點(diǎn)郵件報(bào)送住院人數(shù)、發(fā)熱門診人數(shù)，每日10點(diǎn)網(wǎng)絡(luò)報(bào)送核酸檢測(cè)數(shù)量、中醫(yī)藥治療病例數(shù)，每日18點(diǎn)郵件報(bào)送住院病人明細(xì)等等(部分任務(wù)清單如表2所示)。報(bào)送對(duì)象涉及國(guó)家、省、市、區(qū)的衛(wèi)計(jì)委相關(guān)部門或者各級(jí)防控指揮部。

表2 疫情數(shù)據(jù)上報(bào)任務(wù)清單

從表2任務(wù)清單可知，疫情數(shù)據(jù)上報(bào)大多采用網(wǎng)頁(yè)填報(bào)或郵箱報(bào)送方式，并且對(duì)實(shí)時(shí)性要求高。由于并未提供數(shù)據(jù)接口，初期采用人工方式從各信息系統(tǒng)將數(shù)據(jù)整理出來，十分繁瑣、耗時(shí)。而且涉及到的系統(tǒng)比較多，數(shù)據(jù)庫(kù)類型也各異，為此，利用DataX工具強(qiáng)大的離線同步功能將所需數(shù)據(jù)匯總到一個(gè)專用數(shù)據(jù)庫(kù)(此處使用Oracle)中，然后引入網(wǎng)頁(yè)數(shù)據(jù)填寫成熟技術(shù)或者Python工具，梳理出網(wǎng)頁(yè)填報(bào)數(shù)據(jù)與匯總數(shù)據(jù)庫(kù)中信息數(shù)據(jù)對(duì)應(yīng)關(guān)系。進(jìn)行數(shù)據(jù)統(tǒng)計(jì)上報(bào)時(shí)，自動(dòng)提取匯總數(shù)據(jù)庫(kù)中相關(guān)數(shù)據(jù)，填充到衛(wèi)生部門下發(fā)的各類疫情報(bào)表。最后系統(tǒng)啟動(dòng)代碼或工具將數(shù)據(jù)填寫到網(wǎng)頁(yè)上的相應(yīng)元素域或郵箱中，數(shù)據(jù)填寫完畢后，自動(dòng)提交并退出網(wǎng)頁(yè)[15-16]。通過該方案實(shí)現(xiàn)了大部分?jǐn)?shù)據(jù)的自動(dòng)化填報(bào)，極大減少了數(shù)據(jù)上報(bào)的工作量。

7.2 疫情數(shù)據(jù)展示

精準(zhǔn)及時(shí)的數(shù)據(jù)有利于疫情處理。此次疫情對(duì)醫(yī)療資源消耗巨大。為了提高床位利用效率、緩解醫(yī)療資源緊張的形勢(shì)，根據(jù)醫(yī)院差異化定位分級(jí)接診、促使患者快速分流勢(shì)在必行，這離不開全面數(shù)據(jù)分析的支持。

為了能夠讓院領(lǐng)導(dǎo)實(shí)時(shí)獲取所有院區(qū)各項(xiàng)關(guān)鍵指標(biāo)，便于快速精準(zhǔn)決策，工程師利用大數(shù)據(jù)分析手段開發(fā)了移動(dòng)BI平臺(tái)，用于展示疫情專題數(shù)據(jù)(如圖3所示)。并在此基礎(chǔ)上開發(fā)出院隔離患者初篩報(bào)表，按照診療標(biāo)準(zhǔn)中的出院條件自動(dòng)化標(biāo)識(shí)擬離院患者，然后將報(bào)表實(shí)時(shí)推送給相關(guān)領(lǐng)導(dǎo)，使得管理者對(duì)全院心中有數(shù)、決策有據(jù)，為改善出院管理、床位調(diào)配、轉(zhuǎn)院交接等方面發(fā)揮了作用。重視效率的同時(shí)，質(zhì)量和安全也毫不松懈，針對(duì)住院日、吸氧、ICU入住、轉(zhuǎn)歸等情況開展實(shí)時(shí)統(tǒng)計(jì)，如若發(fā)現(xiàn)負(fù)性指標(biāo)及時(shí)通報(bào)同時(shí)追查。

圖3 方艙醫(yī)院疫情專題主界面

BI平臺(tái)需要實(shí)時(shí)抽取各類系統(tǒng)大量的數(shù)據(jù)，這涉及到多個(gè)無法回避的問題。第一，一些數(shù)據(jù)，比如患者重癥情況、吸氧情況、CT檢查結(jié)果、核酸檢查結(jié)果分布在不同的數(shù)據(jù)庫(kù)中，無法通過聯(lián)合SQL查詢的方法及時(shí)獲取數(shù)據(jù)；第二，由于數(shù)據(jù)的及時(shí)性要求非常高，如果采取直接從生產(chǎn)數(shù)據(jù)庫(kù)中抽取數(shù)據(jù)的方式，勢(shì)必會(huì)造成生產(chǎn)庫(kù)壓力驟增，進(jìn)而影響全院信息系統(tǒng)的運(yùn)行效率，從而對(duì)醫(yī)療業(yè)務(wù)產(chǎn)生影響。

DataX工具在此時(shí)發(fā)揮了巨大的作用，按照數(shù)據(jù)分型編寫好數(shù)據(jù)抽取腳本后，根據(jù)數(shù)據(jù)產(chǎn)生規(guī)律定制化Job計(jì)劃任務(wù)。數(shù)據(jù)實(shí)時(shí)同步到BI數(shù)據(jù)倉(cāng)庫(kù)，BI前端展示只需要查詢數(shù)據(jù)倉(cāng)庫(kù)即可快速查詢數(shù)據(jù)，而且由于是高效率地實(shí)施同步模式，數(shù)據(jù)延遲在可控范圍內(nèi)。

7.3 其他應(yīng)用

疫情結(jié)束后，數(shù)據(jù)上報(bào)會(huì)停止，如果僅僅為了解決這一次上報(bào)問題，花費(fèi)精力去配置工具有些得不償失，但是發(fā)現(xiàn)在醫(yī)院實(shí)際工作中經(jīng)常有各種數(shù)據(jù)上報(bào)，上報(bào)數(shù)據(jù)格式大多數(shù)都是CSV格式。比如國(guó)家衛(wèi)生統(tǒng)計(jì)4-1報(bào)表上報(bào)病案首頁(yè)數(shù)據(jù)，HQMS醫(yī)療數(shù)據(jù)上報(bào)、全國(guó)三級(jí)公立醫(yī)院績(jī)效考核醫(yī)療數(shù)據(jù)上報(bào)、流感數(shù)據(jù)上報(bào)等都涉及到CSV格式數(shù)據(jù)報(bào)送，特別是有一些數(shù)據(jù)還涉及到不同數(shù)據(jù)庫(kù)表聯(lián)合查詢，普通的數(shù)據(jù)庫(kù)查詢導(dǎo)出方式很難實(shí)現(xiàn)，且容易出現(xiàn)數(shù)據(jù)丟失、錯(cuò)位的問題。實(shí)踐表明，DataX不僅能快速導(dǎo)出數(shù)據(jù)，而且能將不同數(shù)據(jù)庫(kù)的數(shù)據(jù)抽取到目標(biāo)庫(kù)，形成數(shù)據(jù)整合，數(shù)據(jù)導(dǎo)出高效、準(zhǔn)確，能很好地滿足醫(yī)療機(jī)構(gòu)日常工作中各類數(shù)據(jù)上報(bào)需求。

8 結(jié)束語

DataX作為一個(gè)服務(wù)于大數(shù)據(jù)的ETL工具，除了提供數(shù)據(jù)快照搬遷功能之外，還提供了豐富數(shù)據(jù)轉(zhuǎn)換的功能，讓數(shù)據(jù)在傳輸過程中可以輕松完成數(shù)據(jù)脫敏、補(bǔ)全、過濾等數(shù)據(jù)轉(zhuǎn)換功能，另外還提供了函數(shù)，讓用戶自定義轉(zhuǎn)換函數(shù)。結(jié)合Python和JDK運(yùn)行環(huán)境[17]，不僅部署簡(jiǎn)單，而且操作方便，能夠根據(jù)需要整合各類格式數(shù)據(jù)，甚至是txt文本、ftp文件，對(duì)于醫(yī)學(xué)統(tǒng)計(jì)專業(yè)常用的dbf格式文件，也能突破常規(guī)256列數(shù)據(jù)的限制，對(duì)于醫(yī)療大數(shù)據(jù)分析意義重大[18]。在此次新型冠狀病毒引起的肺炎疫情中，本院也將該方案推薦給其他醫(yī)療機(jī)構(gòu)，及時(shí)、高效地完成了數(shù)據(jù)上報(bào)工作，為消滅疫情提供了堅(jiān)實(shí)可靠的數(shù)據(jù)支撐。