更關(guān)鍵的是，數(shù)據(jù)來源的業(yè)務(wù)系統(tǒng)也是在不斷地更新維護(hù)中的，任何一個(gè)變更都會(huì)對下游的數(shù)據(jù)分析程序產(chǎn)生巨大的影響。因此，有了ETL過程作為一個(gè)緩沖區(qū)，當(dāng)上游的業(yè)務(wù)系統(tǒng)變更時(shí)，只需要對ETL過程進(jìn)行相應(yīng)變更，下游的數(shù)據(jù)分析就能夠比較穩(wěn)定，從而降低系統(tǒng)維護(hù)成本。

ETL，是英文 Extract-Transform-Load 的縮寫，用來描述將數(shù)據(jù)從來源端經(jīng)過抽?。╡xtract）、轉(zhuǎn)換（transform）、加載（load）至目的端的過程。

1、數(shù)據(jù)清洗

首先進(jìn)行數(shù)據(jù)清洗，對原始數(shù)據(jù)中的錯(cuò)誤予以糾正，或者對缺失數(shù)據(jù)進(jìn)行補(bǔ)填。譬如，現(xiàn)在要建設(shè)一個(gè)增值稅發(fā)票的數(shù)據(jù)中臺(tái)。這時(shí)，系統(tǒng)從許多不同的來源采集與增值稅發(fā)票相關(guān)的數(shù)據(jù)。當(dāng)收集完這些原始數(shù)據(jù)以后，進(jìn)行數(shù)據(jù)清洗工作。增值稅發(fā)票的數(shù)據(jù)結(jié)構(gòu)如下圖所示。

增值稅發(fā)票的數(shù)據(jù)結(jié)構(gòu)圖

在正常的增值稅發(fā)票的數(shù)據(jù)結(jié)構(gòu)中，每張進(jìn)項(xiàng)發(fā)票都應(yīng)當(dāng)有至少一條發(fā)票明細(xì)。然而，可能由于采集的數(shù)據(jù)不一致，發(fā)票與明細(xì)經(jīng)常不是同時(shí)到來，可能相差幾天，造成用發(fā)票分析的數(shù)據(jù)與用發(fā)票明細(xì)分析的數(shù)據(jù)不一致。這時(shí)，必須要先補(bǔ)填一個(gè)發(fā)票明細(xì)，雖然商品名稱與數(shù)量不知道，但至少要保證發(fā)票明細(xì)的金額之和要等于發(fā)票金額，才不至于影響后續(xù)的分析質(zhì)量。至于商品名稱，可以暫時(shí)補(bǔ)填一個(gè)“未知商品”。這樣，當(dāng)該發(fā)票真正的發(fā)票明細(xì)到來時(shí)，再覆蓋原有補(bǔ)填的明細(xì)。

此外，原本每張發(fā)票都應(yīng)當(dāng)有購方納稅人與銷方納稅人，然而由于納稅人信息的基礎(chǔ)數(shù)據(jù)來源于不同的系統(tǒng)，可能造成該發(fā)票的納稅人信息不在納稅人信息表中的情況。這時(shí)，必須要補(bǔ)填一條納稅人信息，使得發(fā)票表與納稅人能夠?qū)?yīng)上，不會(huì)造成數(shù)據(jù)無法關(guān)聯(lián)而缺失數(shù)據(jù)。

同理，每個(gè)納稅人都應(yīng)當(dāng)有各自的稅務(wù)機(jī)關(guān)、地域和行業(yè)，這些信息都可能缺失。對于稅務(wù)機(jī)關(guān)和地域，可以通過納稅人社會(huì)信用代碼中的內(nèi)容進(jìn)行推測。但是，行業(yè)信息是無法推測的。即使無法推測，也不能將其置為null，而是填一個(gè)默認(rèn)值X99999，對應(yīng)到行業(yè)表中的“未知行業(yè)”。

數(shù)據(jù)清洗的過程通過SparkSQL來實(shí)現(xiàn)。通過SparkSQL從原始表中查詢數(shù)據(jù)，然后經(jīng)過以下處理過程，最終寫入ETL臨時(shí)表中：

1/**
 2
 3 * @author fangang
 4
 5 */
 6
 7object ZzsfpJx {
 8
 9  def main(args: Array[String]): Unit = {
10
11    val task = LogUtils.start("zzsfpJxQd")
12
13    try {
14
15      val spark = SparkUtils.init("zzsfpJx")
16
17      val ETLFPMAPNUM = PropertyFile.getProperty("ETLFPMAPNUM").toInt
18
19      spark.udf.register("getJxfpId", (fpdm:String, fphm:String, kprq:String) => 
20      if(null==kprq) fpdm+"X"+fphm+"X" else fpdm+"X"+fphm+"X"+kprq)
21      UdfRegister.fillNsr(spark)
22      UdfRegister.fillSwjg(spark)
23      UdfRegister.cutSL(spark)
24
25
26      val result = spark.sql("SELECT getJxfpId(D.FPDM,D.FPHM,D.KPRQ) JXFP_ID,D.FPDM,D.
27      FPHM,'YB' FP_LB,D.JE JE,cast(cutSL(D.SE/D.JE) as double) SL,D.
28      SE SE,fillNsr(D.XFSBH) XF_NSRSBH, D.XFMC XF_NSRMC, fillNsr(D.GFSBH) GF_NSRSBH,
29      D.GFMC GF_NSRMC,D.KPRQ, D.KPRQ RZSJ, D.XF_QXSWJG_DM SWJG_DM,
30      from_unixtime(unix_timestamp(),'yyyy-MM-dd HH:mm:ss') CZSJ,NSR.SWJG_KEY GF_SWJG_DM,
31      getSwjg(D.XF_QXSWJG_DM,fillNsr(D.XFSBH)) XF_SWJG_DM, case trim(D.fpzt_dm) 
32      when '0' then 'N' when '1' then 'N' else 'Y' end ZFBZ,'' SKM,'' SHRSBH,'
33     ' SHRMC,'' FHRSBH,'' FHRMC,'' QYD,'' SKPH, D.JSHJ,'' CZCH,'' CCDW,'' 
34     YSHWXX,D.BZ,D.tspz_dm as TSPZBZ,CASE WHEN 
35     length(trim(D.zfrq))>15 THEN D.zfrq ELSE NULL END ZFSJ FROM dzdz.DZDZ_FPXX_ZZSFP 
36     D JOIN DW.DW_DM_NSR NSR ON D.GFSBH = NSR.NSR_KEY and 
37     NSR.WDBZ='1'").repartition(ETLFPMAPNUM)
38
39     DataFrameUtils.saveAppend(result, "etl", "etl_jxfp")
40
41     LogUtils.end(task)
42    } catch { case ex:Exception  => LogUtils.error(task, ex) }
43  }
44}
45

在以上SparkSQL程序中，首先從原始數(shù)據(jù)dzdz.DZDZ_FPXX_ZZSFP中查詢數(shù)據(jù)，通過公用方法UdfRegister.fillNsr(spark)與UdfRegister.fillSwjg(spark)對納稅人與稅務(wù)機(jī)關(guān)進(jìn)行補(bǔ)填，保證發(fā)票在與納稅人信息、稅務(wù)機(jī)關(guān)信息關(guān)聯(lián)時(shí)不會(huì)因?yàn)閿?shù)據(jù)為null而造成數(shù)據(jù)缺失。最終，將結(jié)果數(shù)據(jù)寫入etl_jxfp的臨時(shí)表中。

此外，在處理發(fā)票明細(xì)時(shí)加入了這樣一段語句：

1val result1 = spark.sql("SELECT getJxfpqdId(R.FPDM,R.FPHM,R.KPRQ,'00','1') JXFPQD_ID, 
 2
 3getJxfpId(R.FPDM,R.FPHM,R.KPRQ) JXFP_ID ,1.0 HH,'YB' FP_LB,
 4
 5'無商品明細(xì)' WP_MC,'' WP_DW,'' WP_XH,1.0 WP_SL,R.JE DJ,R.JE, cast(cutSL(R.SL) as double) SL,R.SE,R.RZSJ, from_unixtime(unix_timestamp(),'yyyy-MM-dd HH:mm:ss') CZSJ,R.KPRQ,'00' QDBZ,'' SKPH,'' SFZHM,'' CD,'' HGZS,'' JKZMSH,'' SJDH,'' FDJHM, 
 6
 7'' CJHM,'' DH,'' ZH,'' KHYH,'' DW,'' XCRS,0.0 JSHJ,'9999999999999999999' spbm "+s"FROM dzdz.DZDZ_HWXX_ZZSFP D 
 8
 9RIGHT JOIN etl.ETL_JXFP R ON (D.FPDM = R.FPDM AND D.FPHM = R.FPHM) 
10
11WHERE (D.FPDM is null or D.FPHM is null) and R.FP_LB='YB' ").repartition(ETLFPMAPNUM)
12
13DataFrameUtils.saveAppend(result1, "etl", "etl_jxfp_qd")
14

通過該語句在發(fā)票明細(xì)中加入了名為“無商品明細(xì)”的記錄，保證發(fā)票明細(xì)、發(fā)票的金額與稅額沒有缺失，保障后續(xù)數(shù)據(jù)分析的準(zhǔn)確性。

2.數(shù)據(jù)轉(zhuǎn)換

以上一系列的數(shù)據(jù)清洗，可以有效杜絕因?yàn)槿笔?shù)據(jù)或關(guān)聯(lián)不上造成的數(shù)據(jù)分析質(zhì)量問題。接著，就是數(shù)據(jù)轉(zhuǎn)換與集成。

數(shù)據(jù)中臺(tái)的數(shù)據(jù)來源于不同的業(yè)務(wù)系統(tǒng)，因此數(shù)據(jù)格式、計(jì)算口徑都可能存在差異。當(dāng)把它們都抽取到數(shù)據(jù)中臺(tái)以后，應(yīng)當(dāng)將其轉(zhuǎn)換成統(tǒng)一口徑，并規(guī)范計(jì)算口徑。譬如，如何識別代開發(fā)票，不同的系統(tǒng)有不同的判斷邏輯，但經(jīng)過數(shù)據(jù)轉(zhuǎn)換以后，可以在表中增加一個(gè)“是否代開發(fā)票”字段，這樣后續(xù)的分析業(yè)務(wù)就不必再去判斷了，直接看該字段即可。此外，同樣是稅務(wù)機(jī)關(guān)代碼，有的系統(tǒng)是9位，有的系統(tǒng)是11位，應(yīng)該將它們都統(tǒng)一成11位。以上這些工作就是數(shù)據(jù)轉(zhuǎn)換。

3.數(shù)據(jù)集成

清洗和轉(zhuǎn)換工作完成以后，將相同或者相似的數(shù)據(jù)都集成在一起。譬如，從各個(gè)不同路徑采集的納稅人信息，包括納稅人的基礎(chǔ)信息、認(rèn)證信息、核定信息、資格信息，都集成到了納稅人表中；從各個(gè)不同路徑采集的各種不同的增值稅發(fā)票，如增值稅專票、增值稅普票、機(jī)動(dòng)車統(tǒng)一銷售發(fā)票、電子發(fā)票等類型的發(fā)票，都統(tǒng)一集成到發(fā)票信息表中。它們都來源于不同的業(yè)務(wù)系統(tǒng)，字段與類型都各不相同。因此，在集成的過程中，需要進(jìn)行轉(zhuǎn)換或補(bǔ)填，彼此格式一致，并最終存入同一張表中。譬如，其他發(fā)票都有發(fā)票明細(xì)，但機(jī)動(dòng)車統(tǒng)一銷售發(fā)票沒有，因此需要給它補(bǔ)填一條發(fā)票明細(xì)，商品就是那輛汽車，金額與稅額都是那張發(fā)票的金額與稅額。

在具體設(shè)計(jì)實(shí)現(xiàn)上，就是為每一種發(fā)票都編寫一個(gè)發(fā)票與發(fā)票明細(xì)的SparkSQL程序。它們分別從各自的原始數(shù)據(jù)中獲取，但經(jīng)過一個(gè)SQL語句的轉(zhuǎn)換，最終都存入名為etl_jxfp與etl_jxfp_qd的發(fā)票與發(fā)票明細(xì)臨時(shí)表中。

本書摘編自《架構(gòu)真意：企業(yè)級應(yīng)用架構(gòu)設(shè)計(jì)方法論與實(shí)踐》，經(jīng)出版方授權(quán)發(fā)布。