大資料同步利器: 表格存儲全增量一體消費通道

是阿裡雲自研的NoSQL多模型資料庫，提供海量結構化資料存儲以及快速的查詢和分析服務，表格存儲的分布式存儲和強大的索引引擎能夠提供PB級存儲、千萬TPS以及毫秒級延遲的服務能力。

本文會給大家詳細介紹表格存儲重磅推出的一項新功能--全增量一體資料通道。文章會為大家闡述資料通道的主要使用場景，表格存儲資料通道的功能優勢，并帶大家快速入門如何使用資料通道來消費資料。

什麼場景需要資料通道

圖-1 資料通道的主要應用場景

如圖-1展示，資料通道的使用場景主要分為四大類：

資料同步場景，很多時候，使用者的資料除了存儲在表格存儲外還需要同步到其他系統，比如把實時寫入的資料按順序同步到Redis做本地cache，同步到自建的elastic search來支援檢索，同步到OSS做歸檔 [拓展閱讀6 ]，或是同步到容量型表格存儲做冷熱資料分層 [拓展閱讀7 ]等等，通道服務提供了全量、增量、全量加增量三種通道類型來支援不同的同步需求，同時保證資料保序、吞吐速度水準擴充、提供同步延遲監控；
資料移動場景，使用者也會有很多資料遷移的場景，比如将執行個體從一個區域遷移到另一個區域，從容量型替換為高性能執行個體，或是拆表等等，通道服務實作了全量資料消費無縫切換為增量資料消費，不需要業務停寫，大大簡化遷移任務複雜度；
大資料分布式計算架構，常見的大資料場景如爬蟲大資料分析系統 [擴充閱讀9 ]、輿情風控系統、IOT智能制造場景、畫像推薦系統等，其基本架構都是Lambda大資料架構，業務在寫入資料庫的同時需要将資料同步到消息隊列和數倉系統來支援不同的計算引擎，架構依賴多個開源系統，而表格存儲通道服務做到了資料存儲和資料處理all in one，資料隻需寫入表格存儲，對接blink流批一體處理引擎，極大地降低了大資料系統搭建門檻 [利用表格存儲和blink的計算架構可見拓展閱讀5 ]；
事件驅動，很多微服務架構，都會用事件驅動的方式解耦不通的業務邏輯，比如線上業務把資料寫入表格存儲後，異步觸發其他業務的serverless計算函數或者其他事件訂閱系統，訂閱系統可以利用通道服務靈活消費表格存儲的實時資料，函數計算更是直接将表格存儲資料通道作為事件源觸發函數 [拓展閱讀8

總之，表格存儲利用其強大的分布式引擎寫入能力和完備的資料通道功能，做到了讓使用者的資料存儲和資料消費All in one! 使用表格存儲，縮減了使用者系統架構中的外部依賴，也避免引入資料同步和多寫一緻性問題，大大降低上述四大類場景的技術門檻和人力成本。

表格存儲通道服務的優勢

從主流傳統資料庫到NoSQL資料庫，從開源産品到雲産品，都已經有一些資料通道相關的解決方案了，表格存儲資料通道從功能上與主流自建消費通道對比如下：

功能	表格存儲通道服務	MySQL資料同步	Hbase replication架構	AWS Dynamodb stream
同步功能	全量加增量資料同步，全量複制到增量同步無縫切換	分離的全量複制任務和增量同步任務，需要業務方設計切換方案	增量資料同步
資料一緻性	支援保序協定		無保序協定保證一緻性
擴充性	資料規模水準擴充	單機資料庫同步
易用性	多語言SDK	需自建方案或使用開源實作	自建Hbase replication log監聽	使用AWS KCL client
運維監控	RPO消費監控	需自建監控	無RPO監控
計算對接	直讀對接阿裡雲流計算(Blink)	需導出到數倉或消息隊列		需Kinesis擴充卡對接
負載均衡	基于RPO自動負載均衡		無負載均衡	KCL client負載均衡

不少的開源資料系統和計算引擎都實作了從MySQL到自身的資料同步方案，以solr的MySQL全量、增量同步方案為例，使用者需要建立全量導入任務full import和增量同步任務delta import，并且需要根據自身業務安排兩個任務的先後順序、抑或是否需要業務停寫等，同時需要自建同步延遲監控，避免同步滞後和堆積。在對接計算引擎方面，MySQL使用者通常通過資料同步把資料導入到資料倉庫或者消息隊列，進而接入計算引擎，引入了額外的存儲依賴和資料備援，比較複雜。

Hbase上的增量資料可以通過複用Hbase replication架構實作增量資料消費，參照Lily Indexer實作，但是replication會引入離線推送和Hbase線上服務的資源競争，也需要較高的技術門檻解決傳輸優化、熱點問題。同時HBase的日志順序通過資料上的時間戳決定，在時鐘回退和消費逾時時的日志亂序問題難以避免。總體來說，該方案的技術門檻和運維成本都很高，消費場景也需要容忍日志亂序。

AWS Dynamodb stream是Dynamodb的實時資料處了解決方案，DynamoDb會為使用者儲存最近24小時的資料記錄檔，支援使用者以partition粒度并發消費，同時保證增量日志有序。Dynamodb stream不支援導出全量資料，但很多同步場景需要先處理全量的曆史資料随後在開始消費後續的增量資料，另外其複用了KCL實作partition的消費租約、管理partition拓撲關系、持久化每個partition的checkpoint，邏輯很重，沒有對應語言KCL client時使用難度比較大。

相對于以上方案，表格存儲通道服務的主要功能優勢有：

全增量一體，不僅提供增量資料消費，還提供可并行的全量資料消費和全量加增量資料消費，并實作全量複制到增量同步狀态無縫切換；
增量資料保序，通道服務會為使用者資料劃分一到多個可并行消費的邏輯分區，每個邏輯分區的增量資料保序，不同邏輯分區的資料可以并行消費；
消費延遲監控，通道服務通過DescribeTunnel API提供了用戶端消費資料RPO(恢複點目标，recovery point objective)資訊，并在控制台提供了通道資料消費監控；
水準擴充，通道服務會平均配置設定可消費的分區，用通過增加消費端數量，擴充自身實時處理能力；
自動負載均衡，合理消費速度的通道使用者，在其RPO消費延遲增大時，通道服務會自動觸發分區分裂，增大使用者通道的消費并發；

快速入門

通道服務可以在控制台即開即用，計費上同其他資料讀API一樣僅按讀取資料CU計費(按量、預留以及資源包)，并可以在控制台管理和監控通道消費進度和狀态。接下來就帶大家快速體驗一下通道服務的開通、資料消費、消費延遲監控和水準擴充。

首先，我們在控制台選擇已經已有部分資料的表，建立全量加增量資料通道，目的是通過通道先消費完所有存量資料，再按順序實時消費後續新寫入的資料；

大資料同步利器: 表格存儲全增量一體消費通道

圖-2 建立一個全量加增量類型的通道

可以看到建立後通道的通道ID和分區資訊：

圖-3 通道的ID和分區資訊

複制通道ID，通過Java SDK實作通道服務IProcessor接口的消費函數，列印從通道讀取的資料，體驗從通道開始資料消費；

// 使用者自定義資料消費 Callback，即實作IChannelProcessor 接口（process和shutdown）。
private static class SimpleProcessor implements IChannelProcessor {
    @Override
    // ProcessRecordsInput 中包含拉取到的資料。
    public void process(ProcessRecordsInput input) {
        System.out.println("Default record processor, would print records count");
        System.out.println(
            String.format("Process %d records, NextToken: %s", input.getRecords().size(), input.getNextToken()));
        try {
            // Mock Record Process.
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    @Override
    public void shutdown() {
        System.out.println("Mock shutdown");
    }
}

public static void main() throws Exception {
        // 1. 初始化Tunnel Client。
                // endPoint 為表格存儲執行個體 endPoint，如https://instance.cn-hangzhou.ots.aliyuncs.com。
                // accessKeyId 和 accessKeySecret 分别為通路表格存儲服務的 AccessKey 的 Id 和 Secret。
        TunnelClient tunnelClient = new TunnelClient(endPoint, accessKeyId, accessKeySecret, instanceName);

        // 2. 填入建立好的tunnelId，開始資料消費
        // TunnelWorkerConfig裡面還有更多的進階參數，這裡不做展開，會有專門的文檔介紹。
        TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
        TunnelWorker worker = new TunnelWorker(tunnelId, tunnelClient, config);
        try {
            worker.connectAndWorking();
        } catch (Exception e) {
            e.printStackTrace();
            worker.shutdown();
            tunnelClient.shutdown();
        }
    }