背景

Relational Cache是EMR Spark支持的一個(gè)重要特性，主要通過對(duì)數(shù)據(jù)進(jìn)行預(yù)組織和預(yù)計(jì)算加速數(shù)據(jù)分析，提供了類似傳統(tǒng)數(shù)據(jù)倉庫物化視圖的功能。除了用于提升數(shù)據(jù)處理速度，Relational Cache還可以應(yīng)用于其他很多場(chǎng)景，本文主要介紹如何使用Relational Cache跨集群同步數(shù)據(jù)表。
通過統(tǒng)一的Data Lake管理所有數(shù)據(jù)是許多公司追求的目標(biāo)，但是在現(xiàn)實(shí)中，由于多個(gè)數(shù)據(jù)中心，不同網(wǎng)絡(luò)Region，甚至不同部門的存在，不可避免的會(huì)存在多個(gè)不同的大數(shù)據(jù)集群，不同集群的數(shù)據(jù)同步需求普遍存在，此外，集群遷移，搬站涉及到的新老數(shù)據(jù)同步也是一個(gè)常見的問題。數(shù)據(jù)同步的工作通常是一個(gè)比較痛苦的過程，遷移工具的開發(fā)，增量數(shù)據(jù)處理，讀寫的同步，后續(xù)的數(shù)據(jù)比對(duì)等等，需要很多的定制開發(fā)和人工介入。基于Relational Cache，用戶可以簡(jiǎn)化這部分的工作，以較小的代價(jià)實(shí)現(xiàn)跨集群的數(shù)據(jù)同步。
下面我們以具體示例展示如何通過EMR Spark Relational Cache實(shí)現(xiàn)跨集群的數(shù)據(jù)同步。

使用Relational Cache同步數(shù)據(jù)

假設(shè)我們有A，B兩個(gè)集群，需要把a(bǔ)ctivity_log表的數(shù)據(jù)從集群A同步到集群B中，且在整個(gè)過程中，會(huì)持續(xù)有新的數(shù)據(jù)插入到activity_log表中，A集群中activity_log的建表語句如下：

CREATE TABLE activity_log (user_id STRING,act_type STRING,module_id INT,d_year INT) USING JSON PARTITIONED BY (d_year)

插入兩條信息代表歷史信息：

INSERT INTO TABLE activity_log PARTITION (d_year = 2017) VALUES("user_001", "NOTIFICATION", 10), ("user_101", "SCAN", 2)

為activity_log表建一個(gè)Relational Cache：

CACHE TABLE activity_log_sync REFRESH ON COMMIT DISABLE REWRITE USING JSON PARTITIONED BY (d_year) LOCATION "hdfs://192.168.1.36:9000/user/hive/data/activity_log" AS SELECT user_id, act_type, module_id, d_year FROM activity_log

REFRESH ON COMMIT表示當(dāng)源表數(shù)據(jù)發(fā)生更新時(shí)，自動(dòng)更新cache數(shù)據(jù)。通過LOCATION可以指定cache的數(shù)據(jù)的存儲(chǔ)地址，我們把cache的地址指向B集群的HDFS從而實(shí)現(xiàn)數(shù)據(jù)從集群A到集群B的同步。此外Cache的字段和Partition信息均與源表保持一致。

在集群B中，我們也創(chuàng)建一個(gè)activity_log表，創(chuàng)建語句如下：

CREATE TABLE activity_log (user_id STRING,act_type STRING,module_id INT,d_year INT) USING JSON PARTITIONED BY (d_year) LOCATION "hdfs:///user/hive/data/activity_log"

執(zhí)行MSCK REPAIR TABLE activity_log自動(dòng)修復(fù)相關(guān)meta信息，然后執(zhí)行查詢語句，可以看到在集群B中，已經(jīng)能夠查到之前集群A的表中插入的兩條數(shù)據(jù)。

在集群A中繼續(xù)插入新的數(shù)據(jù)：

INSERT INTO TABLE activity_log PARTITION (d_year = 2018) VALUES("user_011", "SUBCRIBE", 24);

然后在集群B中執(zhí)行MSCK REPAIR TABLE activity_log并再次查詢activity_log表，可以發(fā)現(xiàn)數(shù)據(jù)已經(jīng)自動(dòng)同步到集群B的activity_log表中，對(duì)于分區(qū)表，當(dāng)有新的分區(qū)數(shù)據(jù)加入時(shí)，Relational Cache可以增量的同步新的分區(qū)數(shù)據(jù)，而不是重新同步全部數(shù)據(jù)。

?

如果集群A中activity_log的新增數(shù)據(jù)不是通過Spark插入的，而是通過Hive或其他方式外部導(dǎo)入到Hive表中，用戶可以通過REFRESH TABLE activity_log_sync語句手工或通過腳本觸發(fā)同步數(shù)據(jù)，如果新增數(shù)據(jù)是按照分區(qū)批量導(dǎo)入，還可以通過類似REFRESH TABLE activity_log_sync WITH TABLE activity_log PARTITION (d_year=2018)語句增量同步分區(qū)數(shù)據(jù)。

Relational Cache可以保證集群A和集群B中activity_log表的數(shù)據(jù)一致性，依賴activity_log表的下游任務(wù)或應(yīng)用可以隨時(shí)切換到集群B，同時(shí)用戶也可以隨時(shí)將寫入數(shù)據(jù)到集群A中activity_log表的應(yīng)用或服務(wù)暫停，指向集群B中的activity_log表并重啟服務(wù)，從而完成上層應(yīng)用或服務(wù)的遷移。完成后清理集群A中的activity_log和activity_log_sync即可。

總結(jié)

本文介紹了如何通過Relational Cache在不同大數(shù)據(jù)集群的數(shù)據(jù)表之間同步數(shù)據(jù)，非常簡(jiǎn)單便捷。除此之外，Relational Cache也可以應(yīng)用到很多其他的場(chǎng)景中，比如構(gòu)建秒級(jí)響應(yīng)的OLAP平臺(tái)，交互式的BI，Dashboard應(yīng)用，加速ETL過程等等，之后我們也會(huì)和大家分享在更多場(chǎng)景中Relational Cache的最佳實(shí)踐。

原文鏈接
本文為云棲社區(qū)原創(chuàng)內(nèi)容，未經(jīng)允許不得轉(zhuǎn)載。

總結(jié)

以上是生活随笔為你收集整理的使用EMR Spark Relational Cache跨集群同步数据的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。