Spark做爲ETL工具與SequoiaDB的結合應用

1、前言

ETL一詞較經常使用於數據倉庫，但其對象並不只限於數據倉庫。ETL是指將數據從源系統中通過抽取（Extract）、轉換（Transform）、加載（Load）到目標數據存儲區的過程。常見的ETL工具備Oracle Data Integrator、Informatica PowerCenter、DataStage、Kettle、DataSprider等。

在大數據應用中，海量的數據及對潛在應用的支持是很是重要的方面，並體現出與傳統應用開發的巨大不一樣。所以，在選擇合適的ETL工具時，除了須要考慮數據處理的正確性、完整性、工具易用性、對不一樣數據格式的支持程度以外，還必須考慮數據處理的效率、處理能力的可擴展、容錯性。
Spark是UC Berkeley AMP lab開源的類Hadoop MapReduce的通用的並行計算框架，是一個新興的大數據處理引擎，主要特色是提供了一個集羣的分佈式內存抽象。與Hadoop相比，Spark將中間數據放在內存中，避免頻繁寫盤，所以效率更高，更適合於迭代計算；在操做類型、開發語言支持上更豐富；在分佈式數據集計算時經過checkpoint來實現容錯。並且，因爲Spark的分佈式特性，處理能力的擴展更容易，也更經濟。所以，從總體上，Spark做爲ETL工具能幫助企業實現技術和財務的共贏。

SequoiaDB是新一代NewSQL數據庫，是文檔型分佈式數據的典型表明。SequoiaDB企業版經過深度集成最新的Spark內存計算框架，實現了批處理分析、流處理等貼近應用的功能。存儲層和計算層兩層分離的架構，技術互補，是硅谷大數據新架構的主流，將分佈式計算與分佈式存儲的能力分別發揮到了極致。在Spark最新版本中，SparkSQL對標準SQL的支持也愈來愈完善，更加體現出Spark產品的成熟。所以，在SequoiaDB應用中，利用Spark進行數據加工分析是理想之選。

2、功能概述

做爲ETL工具，必須具有多樣數據源的支持，好比HDFS、HBase、Amazon S三、MongoDB等。在這一點上，Spark支持跟多種數據源的對接，常見的數據源包括HDFS、Cassandra、HBase、Hive、ALLUXIO（即Tachyon）、Amazon S3；Spark也能從全文檢索工具Elasticsearch中讀寫數據。Spark做爲ETL工具能知足工具功能通用性的要求。

以Spark爲ETL處理的數據流圖如圖一所示：

圖一 Spark爲ETL數據流圖
在以上數據流圖中，能夠將存儲於HDFS、Cassandra等系統中的存量數據經過Spark提供的接口抽到Spark中，利用Spark的快速處理能力進行處理，好比數據去重、更新，最後將結構數據存儲到巨杉數據庫中。整個處理過程當中，不須要將數據以數據文件的形式存盤，加快了處理速度。

對於已存儲到巨杉數據庫中的數據，也能夠在Spark中處理，並將處理後的數據落到庫中。

3、環境搭建

3.1 Spark環境搭建

Spark運行模式包括Standalone、Spark on YARN、Spark on Mesos。三種模式的主要區別在於使用的資源管理調度工具不同。這裏以Standalone模式爲例進行說明。

在部署以前，將須要部署Spark的機器兩兩之間的信任關係配置好，並根據Spark版本對JDK版本的需求安裝配置好JDK。而後就能夠開始安裝Spark。

首先，從Spark官網獲取最新版本的Spark安裝文件。下載完成後將其解壓到目標文件夾。

tar -zxvf spark-2.0.0-bin-hadoop2.6.tgz

從解壓出來的文件目錄能夠看到，跟1.6版本相比，2.0版本的目錄結構有一些細微變化，lib目錄被刪除，增長了jars目錄。

而後，修改配置文件。一般須要修改的配置文件包含spark-env.sh、slaves，但爲了後續使用方便，還須要修改或增長hive-site.xml、spark-defaults.conf、log4j.properties。下面分別進行說明。

1. spark-env.sh
   配置Spark環境變量，包括：

SPARK_MASTER_IP：Spark集羣Master節點IP地址；
SPARK_MASTER_PORT：Master節點端口號，默認爲7077；
SPARK_WORKER_INSTANCES：每節點啓動的Worker進程數量；
SPARK_WORKER_CORES：本機上Worker可用核數；
SPARK_WORKER_MEMORY：Worker可分配給executor使用的總內存；
SPARK_WORKER_DIR：Worker工做目錄；
SPARK_LOCAL_DIRS：節點shuffle數據存放目錄；
SPARK_CLASSPATH：Spark默認classpath。
2.slaves
配置Spark集羣中運行Worker的節點，值爲主機名，每一行一個主機名。

1. hive-site.xml
   主要用於元數據庫的配置。Spark默認使用Derby做爲數據庫管理元數據，當咱們須要配置其餘數據庫做爲元數據庫時，須要增長並修改此配置文件。一個例子以下所示：

<configuration>
  <property>
    <name>javax.jdo.option.ConnectionURL</name>
    <value>jdbc:postgresql://192.168.111.129:5432/metastore</value>
    <description>JDBC connect string for a JDBC metastore</description>
  </property>
  <property>
    <name>javax.jdo.option.ConnectionDriverName</name>
    <value>org.postgresql.Driver</value>
    <description>Driver class name for a JDBC metastore</description>
  </property>
  <property>
    <name>javax.jdo.option.ConnectionUserName</name>
    <value>hiveuser</value>
    <description>Username to use against metastore database</description>
  </property>
  <property>
    <name>javax.jdo.option.ConnectionPassword</name>
    <value>mypassword</value>
    <description>password to use against metastore database</description>
  </property>
  <property>
    <name>datanucleus.autoCreateSchema</name>
    <value>false</value>
  </property>
</configuration>

1. spark-defaults.conf
   Spark默認配置。該配置能夠配置spark.master、spark.driver.memory、spark.executor.extraJavaOptions等。當咱們須要經過JDBC使用SparkSQL時，須要首先啓動Thriftserver，啓動時須要指定MASTER_URL，這個MASTER_URL能夠配置到spark-defaults.conf中的spark.master參數中，省去在命令行啓動時都須要輸入MASTER_URL的麻煩。

2. log4j.properties
   配置Spark log日誌。

最後，啓動Spark集羣。配置文件修改好後，就能夠啓動Spark。因爲已經配置好Master及Worker的信息，能夠經過以下命令啓動Spark集羣：

sbin/start-all.sh

3.2 配置Spark與SequoiaDB的鏈接
SequoiaDB開源了Spark鏈接器，能夠在github網站上找到相應的代碼（https://github.com/SequoiaDB/...），打包後獲得鏈接器命名爲spark-sequoiadb-2.0.0.jar。將鏈接器和SequoiaDB Java驅動包sequoiadb.jar一塊兒，拷貝至jars目錄，並在spark-env.sh中配置SPARK_CLASSPATH，將鏈接器及驅動包全路徑配置到SPARK_CLASSPATH環境變量中，如

SPARK_CLASSPATH=/opt/spark-2.0.0-bin-hadoop2.6/jars/spark-sequoiadb-2.0.0.jar:/opt/spark-2.0.0-bin-hadoop2.6/jars/sequoiadb.jar

配置完成後，經過以下命令啓動Thriftserver：

sbin/start-thriftserver.sh

啓動成功後，經過jps命令能夠看到Thriftserver相關的進程：

圖二 Thriftserver相關進程
至此，Spark與SequoiaDB的環境已經搭建完成，能夠開始用Spark處理數據了。

4、SequoiaDB數據處理

4.1 處理流程
SequoiaDB與Spark環境搭建完成以後，能夠根據數據源的不一樣，採起不一樣的方式在Spark中建立針對不一樣數據源的映射，就能夠將數據源與SequoiaDB經過Spark鏈接起來，完成ETL處理。

本節以SparkSQL對SequoiaDB中數據進行ETL爲例，說明Spark與SequoiaDB的結合應用。源數據爲其餘時可使用相似的處理邏輯。

在Spark中建立到SequoiaDB中集合的映射表語法爲：

create table tablename (
f1 string,
f2 string,
f3 int,
f4 double,
f5 string,
f6 long
) using com.sequoiadb.spark OPTIONS ( host 'sdbserver1:11810,sdbserver2:11810,sdbserver3:11810', collectionspace 'foo', collection 'bar');

其中，host爲SequoiaDB的訪問地址，格式爲hostname:svcname，能夠包含多個地址。collectionspace及collection分別表明SequoiaDB中的集合空間及集合。

本例爲利用天天增量數據對已有存量數據進行更新的場景，涉及的表爲：帳戶信息表acct_info爲結果表、帳戶信息中轉表repo_acct_info爲天天增量數據、acct_info_his爲已有存量數據。因爲SparkSQL不支持UPDATE及DELETE操做，所以，涉及到UPDATE及DELETE的場景能夠經過將結果數據存於新表的方式來完成。UPDATE分爲兩步：
第一步：將中轉表中最新數據插入結果表。經過這一步，保證第一次進來的數據和存在更新的數據進到結果表。執行語句爲：

insert into table dst.acct_info
select * from src.repo_acct_info
where tx_date = '2016-06-23' ;"

第二步：將未作任何更新的數據數據插入結果表。執行語句爲：

insert into table dst.acct_info
select distinct a.* from src.acct_info_his a left join src.repo_acct_info b
on a.id = b.id and b.tx_date = '2016-06-23'
where b.id is null ;"

其中id爲acct_info表的主鍵，經過id惟一標識一條記錄。經過以上兩個步驟，結果表acct_info即爲通過去重後的更新數據。
而DELETE操做則只須要將不知足刪除條件的數據插入新表便可。

4.2 性能結果

1.系統配置

硬件環境

軟件環境

2.UPDATE場景

3.DELETE場景

5、結論

Spark可以方便地讀取多樣數據源，做爲一種較爲成熟的新框架，Spark不只支持HDFS、Cassandra、HIVE、Amazon S3這類相對較新的數據源，對傳統行業常見的如Oracle、DB二、Teradata，Spark也能很好地支持，且Spark支持SQL2003，應用Spark也能充分發揮傳統企業在SQL處理上的強項。在大數據應用中，以Spark爲ETL工具能夠充分發揮分佈式計算框架Spark的處理能力的性能優點。

做爲全球得到Databricks認證的14家發行商之一，SequoiaDB企業版深度集成最新的Spark內存計算框架，存儲層和計算層兩層分離的架構、技術互補，是硅谷大數據新架構的主流，將分佈式計算與分佈式存儲的能力分別發揮到了極致。現在，Spark技術已經被大量運用到實時流處理、分析等不一樣領域，後臺數據加工也能夠利用Spark技術得以實現。

SequoiaDB巨杉數據庫2.6最新版下載
SequoiaDB巨杉數據庫技術博客
SequoiaDB巨杉數據庫社區