怎么使用EMR Spark Relational Cache跨集群同步數據

這篇文章給大家介紹怎么使用EMR Spark Relational Cache跨集群同步數據，內容非常詳細，感興趣的小伙伴們可以參考借鑒，希望對大家能有所幫助。

使用Relational Cache加速EMR Spark數據分析

背景

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

使用Relational Cache同步數據

假設我們有A，B兩個集群，需要把activity_log表的數據從集群A同步到集群B中，且在整個過程中，會持續有新的數據插入到activity_log表中，A集群中activity_log的建表語句如下：

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

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

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

為activity_log表建一個Relational Cache：

CACHE TABLE activity_log_syncREFRESH ON COMMITDISABLE REWRITEUSING JSONPARTITIONED 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表示當源表數據發生更新時，自動更新cache數據。通過LOCATION可以指定cache的數據的存儲地址，我們把cache的地址指向B集群的HDFS從而實現數據從集群A到集群B的同步。此外Cache的字段和Partition信息均與源表保持一致。

在集群B中，我們也創建一個activity_log表，創建語句如下：

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

執行MSCK REPAIR TABLE activity_log自動修復相關meta信息，然后執行查詢語句，可以看到在集群B中，已經能夠查到之前集群A的表中插入的兩條數據。

在集群A中繼續插入新的數據：

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

然后在集群B中執行MSCK REPAIR TABLE activity_log并再次查詢activity_log表，可以發現數據已經自動同步到集群B的activity_log表中，對于分區表，當有新的分區數據加入時，Relational Cache可以增量的同步新的分區數據，而不是重新同步全部數據。

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

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

通過Relational Cache在不同大數據集群的數據表之間同步數據，非常簡單便捷。除此之外，Relational Cache也可以應用到很多其他的場景中，比如構建秒級響應的OLAP平臺，交互式的BI，Dashboard應用，加速ETL過程等等。

關于怎么使用EMR Spark Relational Cache跨集群同步數據就分享到這里了，希望以上內容可以對大家有一定的幫助，可以學到更多知識。如果覺得文章不錯，可以把它分享出去讓更多的人看到。