王浩，葛昂，趙晴

　?。ㄈA北計算機系統(tǒng)工程研究所，北京 100083）

       摘要：為了實現(xiàn)將傳統(tǒng)關系型數(shù)據(jù)庫中的增量數(shù)據(jù)快速導入同構或者異構目的庫，在使用已有的增量提取方法的基礎上，提出了通過增加并行度和流式計算的方法加快同步速度。此方法不僅支持插入、更新和刪除的增量數(shù)據(jù)同步，而且可以抽取出數(shù)據(jù)庫表結構信息動態(tài)支持表結構變更。與傳統(tǒng)單點抽取方式相比，大大提高了目的庫數(shù)據(jù)的新鮮度。

　　關鍵詞：增量同步; Spark; 流式計算

0引言

　　隨著大數(shù)據(jù)技術的發(fā)展，越來越多的企業(yè)開始構建大數(shù)據(jù)平臺進行數(shù)據(jù)處理。然而如何將保存在關系型數(shù)據(jù)庫中的數(shù)據(jù)快速同步到大數(shù)據(jù)平臺組件（例如HBase、HDFS）中，正成為很多企業(yè)面臨的問題。Sqoop是常用的數(shù)據(jù)同步工具，其實質(zhì)是MapReduce任務，延時較高，而且需要通過定時任務來達到自動化流程效果。本文在觸發(fā)器記錄數(shù)據(jù)變化的基礎上，提出了一種使用Spark Streaming將增量數(shù)據(jù)抽取出來，然后根據(jù)需要寫入到不同的目的庫的方法。由于只提取增量數(shù)據(jù)，所以較Sqoop減少了數(shù)據(jù)量。另外由于是流式處理方式，降低了延時。

1增量提取

　　1.1增量提取的概念

　　增量提取是針對上一次提取而言，將上一次提取時間點到現(xiàn)在數(shù)據(jù)庫中插入、更新、刪除的數(shù)據(jù)提取出來［1］。

　　1.2常用的增量提取方法

　　1.2.1基于業(yè)務系統(tǒng)日志

　　在業(yè)務中將數(shù)據(jù)庫DML（Data Manipulation Language）語句輸出以日志的方式存儲，然后通過解析日志將DML語句在目的庫中重放以達到目的。此方法需要侵入業(yè)務系統(tǒng)，對于已經(jīng)成型的業(yè)務系統(tǒng)不適用。

　　1.2.2基于數(shù)據(jù)庫日志

　　解析數(shù)據(jù)庫日志也能達到增量提取的目的，但是各大數(shù)據(jù)庫廠商不對外開放數(shù)據(jù)庫系統(tǒng)的日志格式，這就使得解析日志變成了問題。而且各數(shù)據(jù)庫的日志格式還不盡相同，難以達到通用性。

　　1.2.3基于觸發(fā)器

　　基于觸發(fā)器的方式，目前被廣泛運用于數(shù)據(jù)庫增量提取。它通過在源表上建立插入、更新、刪除觸發(fā)器來記錄對數(shù)據(jù)的操作。每當有數(shù)據(jù)變化時，就會觸發(fā)相應的觸發(fā)器，然后運行觸發(fā)器定義的邏輯，將變化記錄到增量表。

　　1.3基于觸發(fā)器方法的具體實現(xiàn)

　　由于觸發(fā)器方法具有實現(xiàn)邏輯簡單，對業(yè)務無入侵，數(shù)據(jù)庫通用等優(yōu)點，所以本文采用了基于觸發(fā)器方式的增量提取方法。具體實現(xiàn)方法如下：

　?。?）創(chuàng)建名為dml_log的數(shù)據(jù)庫表，字段為id、table_name、record_id、execute_date、dml_type。其中id為自增id，table_name存儲要同步的源表表名稱，record_id是源表中發(fā)生變化的記錄的唯一標識，execute_date為觸發(fā)器執(zhí)行時的時間戳，dml_type為I、U、D分別代表insert、update、delete操作。

　?。?）在源表上創(chuàng)建插入、更新、刪除類型的觸發(fā)器。創(chuàng)建語句在此省略。

2構建Spark Streaming程序

　　2.1Spark Streaming

　　Spark是目前大數(shù)據(jù)處理領域比較常用的計算框架。它將中間計算結果維護在內(nèi)存中，這樣不僅可以做到中間結果的重用，而且減少了磁盤IO，大大加快了計算速度。Spark Streaming是構建于Spark core之上的流式處理模塊。其原理是將流式數(shù)據(jù)切分成一個個小的片段，以mini batch的形式來處理這一小部分數(shù)據(jù)，從而模擬流式計算達到準實時的效果。

　　2.2JdbcRDD

　　彈性分布式數(shù)據(jù)集(Resilient Distributed Datasets,RDD)，它是Spark數(shù)據(jù)抽象的基石。RDD是一個只讀的分區(qū)記錄集合，分區(qū)分散在各個計算節(jié)點［2］。RDD提供了transformation和action兩類操作，其中transformation是lazy級別的，主要對數(shù)據(jù)處理流程進行標記，而不立即進行運算。action操作會觸發(fā)作業(yè)的提交，然后進行回溯導致transformation操作進行運算。

　　JdbcRDD擴展自RDD，是RDD的子類。內(nèi)部通過JDBC（Java Data Base Connectivity）操作以數(shù)據(jù)庫為源頭構建RDD。其構造函數(shù)簽名為：

　　class JdbcRDD［T: ClassTag］(

　　sc: SparkContext,

　　getConnection:()=> Connection,

　　sql: String,

　　lowerBound: Long,

　　upperBound: Long,

　　numPartitions: Int,

　　mapRow:(ResultSet) => T =

　　JdbcRDD.resultSetToObjectArray _)

　　extends RDD［T］(sc, Nil) with Logging {…}

　　2.3具體實現(xiàn)

　　Spark官方提供用于構建Spark Streaming的數(shù)據(jù)源沒有對數(shù)據(jù)庫進行支持，所以本文自己實現(xiàn)對數(shù)據(jù)庫的支持。編寫繼承自InputDStream類的DirectJdbcInputDStream類，其簽名為：

　　class DirectJdbcInputDStream［T: ClassTag］(

　　@transient ssc_ : StreamingContext,

　　param: JdbcParam) extends

　　InputDStream［Row］ (ssc_) with Logging {…}

　　對start()、compute()和stop()方法進行重寫。

　　(1)在start函數(shù)中注冊JDBC驅(qū)動，用于JDBC獲取初始化信息（構造JdbcRDD時的參數(shù)）；

　　(2)compute函數(shù)會被框架間隔指定的時間反復調(diào)用，其實質(zhì)是如何返回一個JdbcRDD。首先通過JDBC獲取本次需要拉取的trigger記錄的id的上下界以及表的Schema信息；然后以這些信息為參數(shù)生成提取真實數(shù)據(jù)的SQL，其邏輯為用選中的trigger表中的記錄和原表在record_id上進行左連接；最后使用該SQL當做參數(shù)構建JdbcRDD。值得說明的是，構建JdbcRDD時是可以指定并行度的，每個worker節(jié)點都會建立到數(shù)據(jù)庫的JDBC連接，由多個節(jié)點并行去數(shù)據(jù)庫拉取屬于自己的那一部分數(shù)據(jù)，這就大大增加了提取和處理速度。

　　(3)在stop函數(shù)中關閉JDBC連接?？傮w來看，就是在driver程序中執(zhí)行的JDBC程序獲取初始化參數(shù)，在executor中執(zhí)行的JDBC程序拉取真實的數(shù)據(jù)。

　　(4)編寫driver程序：

　　val sc = new SparkContext(new SparkConf)

　　val ssc = new StreamingContext(sc, Seconds(30))

　　val directStream = new DirectJdbcInputDStream［Row］(ssc, jdbcParam)

　　directStream.foreachRDD(rdd => {

　　…//對數(shù)據(jù)進行處理

　　})

　　2.4限流

　　假設當前時間點到上次提取的時間點之間新增數(shù)據(jù)量太大，就會導致在新一次作業(yè)提交時，上一次的作業(yè)仍然沒有完成，可能會因此造成作業(yè)積壓使得系統(tǒng)不穩(wěn)定。本文使用了基于規(guī)則的限流方法，綜合考慮集群處理能力以及間隔時間，可以配置化設置每次最大提取條數(shù)。如果當前需要提取的數(shù)據(jù)條數(shù)大于最大提取條數(shù)，則本次就只提取最大條數(shù)，剩下的延時到下次再進行提取。這樣做的好處是削減了峰流對系統(tǒng)造成的影響。

3測試分析

　　測試環(huán)境：VMware虛擬機，處理器設置為4核心，2 GB內(nèi)存， 64位CentOS 6.5操作系統(tǒng)，Spark 1.5.1，Oracle 11g。使用4臺虛擬機搭建成Spark集群，1臺為Master，3臺為Worker。數(shù)據(jù)庫表分別設置為20、40個字段，每次最大抽取記錄數(shù)分別設置為10 000、50 000、500 000。將抽取出來的數(shù)據(jù)寫成parquet格式的文件存儲到hdfs上。測試結果如表1所示。

4結束語

　　本文在基于數(shù)據(jù)庫觸發(fā)器記錄數(shù)據(jù)變化的基礎上，通過自己構造DirectJdbcStream類提供Spark Streaming對數(shù)據(jù)庫的支持，達到準實時從數(shù)據(jù)庫中抽取出增量數(shù)據(jù)的目的。并且可以對抽取出來的數(shù)據(jù)進行過濾、清洗等操作，根據(jù)需求靈活地寫入到不同的目的庫。

　　參考文獻

　?。?］ 郭亮. 基于MD5與HASH的數(shù)據(jù)庫增量提取算法及其應用［D］. 長沙：湖南大學，2013.

　?。?］ ZAHARIA M， CHOWDHURY M, DAS T, et al. Resilient distributed datasets: a fault tolerant abstraction for in memory cluster computing［C］. Usenix Conference on Networked Systems Design & Implementation, 2012, 70(2):141146.

　?。?］ DEAN J， GHEMAWAT S. MapReduce: simplified dataprocessing on large clusters［C］. USENIX Association OSDI′04: 6th Symposium on Operating Systems Design and Implementation, 2004:137149.

　?。?］ MARTIN O. Programming in scala［M］. California: Artima Press,2010.

　?。?］ YADAV R. Spark cookbook［M］. UK: Packt Publishing Ltd, 2015.

　?。?］ KARAU H. Learning spark［M］. America: O’Reilly Media, Inc. 2015.

　?。?］ 梁剛. 企業(yè)大數(shù)據(jù)管理解決方案［J］. 微型機與應用，2013,32(24)：7 10,13.