王 浩，葛 昂，趙 晴

(華北計算機系統(tǒng)工程研究所，北京 100083)

?

基于Spark的數(shù)據(jù)庫增量準實時同步

王 浩，葛 昂，趙 晴

(華北計算機系統(tǒng)工程研究所，北京 100083)

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

(1)創(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操作。

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

2 構建Spark Streaming程序

2.1 Spark Streaming

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

2.2 JdbcRDD

彈性分布式數(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上。測試結(jié)果如表1所示。

表1 運行時間 (單位：s)

4 結(jié)束語

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

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

[2] 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.

[3] 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.

[4] MARTIN O.Programming in scala[M].California: Artima Press,2010.

[5] YADAV R.Spark cookbook[M].UK: Packt Publishing Ltd,2015.

[6] KARAU H.Learning spark[M].America: O’Reilly Media,Inc.2015.

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

Spark-based database increment near-real-time synchronization

Wang Hao,Ge Ang,Zhao Qing

(National Computer System Engineering Research Institute of China,Beijing 100083,China)

In order to export incremental data stored in traditional database to homogeneous or heterogeneous destination,on the basis of existing incremental extraction method,we proposed a solution to speed up synchronization by increasing parallelism and using streaming instead of batch.This approach supports incremental data of inserting,updating and deleting,and can extract the database table schema information to support dynamic table structure changes.Compared with traditional single-point mode,it makes data more fresh.

increment extraction; Spark; streaming computing

TP311.1

A DOI：10.19358/j.issn.1674-7720.2016.19.002

王浩，葛昂，趙晴.基于Spark的數(shù)據(jù)庫增量準實時同步[J].微型機與應用，2016,35(19)：9-10,13.

2016-05-05)

王浩(1989-)，男，碩士，主要研究方向：大數(shù)據(jù)。

葛昂(1972-)，男，碩士，高級工程師，主要研究方向：企業(yè)軟件架構、多維數(shù)據(jù)綜合應用。

趙晴(1964-)，男，學士，高級工程師，主要研究方向：工業(yè)控制物聯(lián)網(wǎng)。