使用EMR Spark Relational Cache跨集群同步数据
背景
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 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表建一個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_logREFRESH 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 JSON PARTITIONED 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過程等等,之后我們也會和大家分享在更多場景中Relational Cache的最佳實踐。
原文鏈接
本文為云棲社區原創內容,未經允許不得轉載。
總結
以上是生活随笔為你收集整理的使用EMR Spark Relational Cache跨集群同步数据的全部內容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: 开发者说:Seata 0.7.0 版本,
- 下一篇: 历时五天用 SwiftUI 做了一款 A