在 Halodoc，我們始終致力于為最終用戶簡化醫(yī)療保健服務(wù)，隨著公司的發(fā)展，我們不斷構(gòu)建和提供新功能。我們兩年前建立的可能無法支持我們今天管理的數(shù)據(jù)量，以解決我們決定改進(jìn)數(shù)據(jù)平臺(tái)架構(gòu)的問題。在這篇文章中，我們將討論我們的新架構(gòu)、涉及的組件和不同的策略，以擁有一個(gè)可擴(kuò)展的數(shù)據(jù)平臺(tái)。

一、新架構(gòu)

讓我們首先看一下經(jīng)過改進(jìn)的新數(shù)據(jù)平臺(tái) 2.0 的高級架構(gòu)。

我們將架構(gòu)分為 4 層：

1. 數(shù)據(jù)攝取/提取層

該層更關(guān)心在原始區(qū)域?qū)又袛z取數(shù)據(jù)，這些數(shù)據(jù)可以稍后在已處理區(qū)域中使用和卸載。大多數(shù)點(diǎn)擊流捕獲工具都支持來自其產(chǎn)品的內(nèi)部數(shù)據(jù)攝取服務(wù)，從而可以輕松獲取或加入原始區(qū)域以進(jìn)行進(jìn)一步處理。對于 MySQL、Postgres 等事務(wù)性數(shù)據(jù)源，我們開始利用基于 CDC 的方法進(jìn)行數(shù)據(jù)提取。由于我們的基礎(chǔ)設(shè)施主要托管在 AWS 中，因此我們選擇了數(shù)據(jù)遷移服務(wù) (DMS) 來執(zhí)行基于 CDC 的遷移。

2. 處理層

這里我們沒有執(zhí)行任何繁重的轉(zhuǎn)換，而是將原始數(shù)據(jù)轉(zhuǎn)換為 HUDI 數(shù)據(jù)集。源數(shù)據(jù)以不同的格式(CSV、JSON)攝取，需要將其轉(zhuǎn)換為列格式(例如parquet)，以將它們存儲(chǔ)在 Data Lake 中以進(jìn)行高效的數(shù)據(jù)處理。數(shù)據(jù)類型基于數(shù)據(jù)湖兼容性進(jìn)行類型轉(zhuǎn)換，時(shí)區(qū)調(diào)整為 WIB 時(shí)間戳。

3. 轉(zhuǎn)換層

數(shù)據(jù)工程的一大挑戰(zhàn)是有效地處理大量數(shù)據(jù)并保持成本不變。我們選擇 Apache Spark 進(jìn)行處理，因?yàn)樗С址植际綌?shù)據(jù)處理，并且可以輕松地從千兆字節(jié)擴(kuò)展到 TB 級數(shù)據(jù)處理。轉(zhuǎn)換層在數(shù)據(jù)倉庫中生成數(shù)據(jù)模型，并成為報(bào)表使用數(shù)據(jù)并支持儀表板或報(bào)表用例的基礎(chǔ)。

4. 報(bào)告層

報(bào)告層主要從維度和事實(shí)表中聚合數(shù)據(jù)，并在這些數(shù)據(jù)庫之上提供視圖供下游用戶使用。大多數(shù)儀表板將建立在這些報(bào)告表和物化視圖之上，從而減少為重復(fù)性任務(wù)和報(bào)告用例連接不同表的計(jì)算成本。一旦我們將平臺(tái)實(shí)現(xiàn)為不同的層，下一個(gè)挑戰(zhàn)就是選擇能夠支持我們大多數(shù)下游用例的組件。當(dāng)我們調(diào)研市場上的數(shù)據(jù)工程工具/產(chǎn)品時(shí)，我們可以輕松找到大量工具。我們計(jì)劃利用 AWS 云和開源項(xiàng)目構(gòu)建內(nèi)部解決方案，而不是購買第三方許可工具。

讓我們更深入地了解上述平臺(tái)中使用的組件。

涉及的組件：

(1) 管理系統(tǒng)

DMS 代表數(shù)據(jù)遷移服務(wù)。這是一項(xiàng) AWS 服務(wù)，可幫助在 MySQL、Postgres 等數(shù)據(jù)庫上執(zhí)行 CDC(更改數(shù)據(jù)捕獲)。我們利用 DMS 從 MySQL DB 讀取二進(jìn)制日志并將原始數(shù)據(jù)存儲(chǔ)在 S3 中。我們已經(jīng)自動(dòng)化了在 Flask 服務(wù)器和 boto3 實(shí)現(xiàn)的幫助下創(chuàng)建的 DMS 資源。我們可以輕松地在控制表中配置的原始區(qū)域參數(shù)中加入新表。

(2) S3 - 原始區(qū)域

DMS 捕獲的所有 CDC 數(shù)據(jù)都存儲(chǔ)在 S3 中適當(dāng)分區(qū)的原始區(qū)域中。該層不執(zhí)行數(shù)據(jù)清洗。只要源系統(tǒng)中發(fā)生插入或更新，數(shù)據(jù)就會(huì)附加到新文件中。原始區(qū)域?qū)τ谠谛枰獣r(shí)執(zhí)行數(shù)據(jù)集的任何回填非常重要。這還存儲(chǔ)從點(diǎn)擊流工具或任何其他數(shù)據(jù)源攝取的數(shù)據(jù)。原始區(qū)域充當(dāng)處理區(qū)域使用數(shù)據(jù)的基礎(chǔ)層。

(3) EMR - HUDI + PySpark

Apache HUDI 用于對位于 Data Lake 中的數(shù)據(jù)利用 UPSERT 操作。我們正在運(yùn)行 PySpark 作業(yè)，這些作業(yè)按預(yù)定的時(shí)間間隔運(yùn)行，從原始區(qū)域讀取數(shù)據(jù)，處理并存儲(chǔ)在已處理區(qū)域中。已處理區(qū)域復(fù)制源系統(tǒng)的行為。這里只是發(fā)生了一個(gè) UPSERT 操作并轉(zhuǎn)換為 HUDI 數(shù)據(jù)集。

(4) S3 - 處理區(qū)

S3 處理層是 Halodoc 的數(shù)據(jù)湖。我們存儲(chǔ)可變和不可變數(shù)據(jù)集。HUDI 被用于維護(hù)可變數(shù)據(jù)集。CSV 或 JSON 數(shù)據(jù)等不可變數(shù)據(jù)集也被轉(zhuǎn)換為列格式(parquet)并存儲(chǔ)在該區(qū)域中。該層還維護(hù)或糾正分區(qū)以有效地查詢數(shù)據(jù)集。

(5) Glue數(shù)據(jù)目錄

AWS Glue 數(shù)據(jù)目錄用于注冊表，并可通過 Athena 進(jìn)行查詢以進(jìn)行臨時(shí)分析。

(6) Athena

Athena 是一個(gè)無服務(wù)器查詢引擎，支持查詢 S3 中的數(shù)據(jù)。用戶利用 Athena 對位于數(shù)據(jù)湖中的數(shù)據(jù)集進(jìn)行任何臨時(shí)分析。

(7) Redshift

Redshift 用作數(shù)據(jù)倉庫來構(gòu)建數(shù)據(jù)模型。所有報(bào)告/BI 用例均由 Redshift 提供服務(wù)。我們在 Redshift 中創(chuàng)建了 2 個(gè)圖層。一層負(fù)責(zé)存儲(chǔ)包含事實(shí)和維度的 PD、CD、Appointments、Insurance 和 Labs 的所有數(shù)據(jù)模型。我們已經(jīng)構(gòu)建了一個(gè)報(bào)告層框架來進(jìn)行聚合和連接，以創(chuàng)建可通過 BI 工具訪問的報(bào)告表。我們還在這些層中維護(hù)物化視圖。我們還在我們的數(shù)據(jù)模型中實(shí)現(xiàn)了 SCD type1 和 SCD type2，以捕捉數(shù)據(jù)集中的歷史變化。

(8) MWAA

MWAA 用于編排工作流程。

(9) Cloud Watch和EFK

Cloud Watch 和 EFK 相結(jié)合，構(gòu)建集中的日志記錄、監(jiān)控和警報(bào)系統(tǒng)。

(10) Dynamicdb

平臺(tái)中使用 Dynamodb 將失敗的事件存儲(chǔ)在控制表中發(fā)布。開發(fā)了一個(gè)再處理框架來處理失敗的事件并按預(yù)定的頻率將它們推送到控制表。

二、為什么選擇基于 CDC 的方法？

在 Halodoc，當(dāng)我們開始數(shù)據(jù)工程之旅時(shí)，我們采用了基于時(shí)間戳的數(shù)據(jù)遷移。我們依靠修改后的時(shí)間戳將數(shù)據(jù)從源遷移到目標(biāo)。我們幾乎用這個(gè)管道服務(wù)了 2 年。隨著業(yè)務(wù)的增長，我們的數(shù)據(jù)集呈指數(shù)級增長，這要求我們將遷移實(shí)例增加到更大的集群以支持大量數(shù)據(jù)。

問題如下：

• 由于源處生成的大量數(shù)據(jù)導(dǎo)致遷移集群大小增加，因此成本高。
• 由于某些后端問題，未更新已修改列時(shí)的數(shù)據(jù)質(zhì)量問題。
• 架構(gòu)更改很難在目標(biāo)中處理。
• 在基于 CDC 的情況下，我們通過在 MySQL 中啟用 binlog(二進(jìn)制日志)和在 Postgres 中啟用 WAL(預(yù)寫日志)來開始讀取事務(wù)數(shù)據(jù)。提取每個(gè)事件更改的新文件是一項(xiàng)昂貴的操作，因?yàn)闀?huì)有很多 S3 Put 操作。為了平衡成本，我們將 DMS 二進(jìn)制日志設(shè)置為每 60 秒讀取和拉取一次。每 1 分鐘，通過 DMS 插入新文件。基于 CDC 還解決了數(shù)據(jù)量大增長的問題，因?yàn)槲覀冮_始以最大分鐘間隔遷移，而不是每小時(shí)間隔數(shù)據(jù)。

三、使用Apache Hudi

HUDI 提供內(nèi)置功能來支持開放數(shù)據(jù)湖。在我們的平臺(tái)中加入或集成 HUDI 時(shí)，我們面臨以下一些挑戰(zhàn)并試圖解決它們。

1. 保留 HUDI 數(shù)據(jù)集中的最大提交

HUDI 根據(jù)配置集清理/刪除較舊的提交文件。默認(rèn)情況下，它已將保留的提交設(shè)置為 10。必須根據(jù)一個(gè)工作負(fù)載正確設(shè)置這些提交。由于我們在 5 分鐘內(nèi)運(yùn)行了大部分事務(wù)表遷移，因此我們將 hoodie.cleaner.commits.retained 設(shè)置為 15，以便我們有 75 分鐘的時(shí)間來完成 ETL 作業(yè)。甚至壓縮和集群添加到提交，因此必須分析和設(shè)置更清潔的策略，以使增量查詢不間斷地運(yùn)行。

2. 確定要分區(qū)的表

在數(shù)據(jù)湖中對數(shù)據(jù)進(jìn)行分區(qū)總是可以減少掃描的數(shù)據(jù)量并提高查詢性能。同樣，在湖中擁有大分區(qū)會(huì)降低讀取查詢性能，因?yàn)樗仨毢喜⒍鄠€(gè)文件來進(jìn)行數(shù)據(jù)處理。我們選擇我們的數(shù)據(jù)湖來進(jìn)行最小的每日分區(qū)，并計(jì)劃將歷史數(shù)據(jù)歸檔到其他存儲(chǔ)層，如 Glacier 或低成本的 S3 存儲(chǔ)層。

3. 選擇正確的存儲(chǔ)類型

HUDI 目前支持 2 種類型的存儲(chǔ)，即。MoR(讀取時(shí)合并)和 CoW(寫入時(shí)復(fù)制)。必須根據(jù)用例和工作負(fù)載精確選擇存儲(chǔ)類型。我們?yōu)榫哂休^低數(shù)據(jù)延遲訪問的表選擇了 MoR，為可能具有超過 2 小時(shí)數(shù)據(jù)延遲的表選擇了 CoW。

4. MoR 數(shù)據(jù)集的不同視圖

MoR 支持 _ro 和 _rt 視圖。_ro 代表讀取優(yōu)化視圖，_rt 代表實(shí)時(shí)視圖。根據(jù)用例，必須確定要查詢哪個(gè)表。我們?yōu)?ETL 工作負(fù)載選擇了 _ro 視圖，因?yàn)閿?shù)據(jù)模型中的數(shù)據(jù)延遲約為 1 小時(shí)。建立在數(shù)據(jù)湖之上的報(bào)告正在查詢 _rt 表以獲取數(shù)據(jù)集的最新視圖。

5. HUDI 中的索引

索引在 HUDI 中對于維護(hù) UPSERT 操作和讀取查詢性能非常有用。有全局索引和非全局索引。我們使用默認(rèn)的bloom索引并為索引選擇了一個(gè)靜態(tài)列，即非全局索引。我們依靠 HUDI 提交時(shí)間來獲取增量數(shù)據(jù)。這也有助于將遲到的數(shù)據(jù)處理到要處理的數(shù)據(jù)湖，而無需任何人工干預(yù)。

五、為什么框架驅(qū)動(dòng)

我們之前的大部分實(shí)施都是管道驅(qū)動(dòng)的，這意味著我們?yōu)槊總€(gè)數(shù)據(jù)源手動(dòng)構(gòu)建管道以服務(wù)于業(yè)務(wù)用例。在 Platform 2.0 中，我們對實(shí)現(xiàn)模型進(jìn)行了細(xì)微的更改，并采用了框架驅(qū)動(dòng)的管道。我們開始在每一層上構(gòu)建一個(gè)框架，例如數(shù)據(jù)攝取框架、數(shù)據(jù)處理框架和報(bào)告框架。每個(gè)框架都專用于使用預(yù)定義的輸入執(zhí)行某些任務(wù)。采用框架驅(qū)動(dòng)減少了冗余代碼，以維護(hù)和簡化數(shù)據(jù)湖中新表的載入過程。

1. 使用表格格式的控制平面的好處

在我們的平臺(tái)中，控制平面是一個(gè)關(guān)鍵組件，用于存儲(chǔ)元數(shù)據(jù)并幫助輕松載入數(shù)據(jù)湖和數(shù)據(jù)倉庫中的新表。它存儲(chǔ)啟用數(shù)據(jù)遷移所需的必要配置。對于構(gòu)建任何產(chǎn)品，元數(shù)據(jù)在自動(dòng)化和控制管道流程方面起著至關(guān)重要的作用。在 Yaml、DynamoDB 或 RDBMS 中，我們有不同的選項(xiàng)可供選擇。我們選擇 RDS 的原因如下：

• 輕松在元數(shù)據(jù)之上執(zhí)行任何分析，例如活動(dòng)管道的數(shù)量。
• 易于載入新表或數(shù)據(jù)模型。
• 借助 python flask API 輕松構(gòu)建 API 層。
• 審計(jì)可以很容易地完成。
• 數(shù)據(jù)安全

在醫(yī)療保健領(lǐng)域，安全一直是我們數(shù)據(jù)平臺(tái)中啟用的重中之重。我們在私有子網(wǎng)中托管了幾乎所有基礎(chǔ)設(shè)施，并啟用 Lake Formation 來管理對 Data Lake 的訪問。我們還對靜態(tài)數(shù)據(jù)使用 AWS 加密。這提供了數(shù)據(jù)湖和整體數(shù)據(jù)平臺(tái)的安全存儲(chǔ)。

2. 自動(dòng)化

自動(dòng)化總是有助于減少構(gòu)建和維護(hù)平臺(tái)的工程工作量。在 Platform 2.0 中，我們的大部分流水線都使用 Jenkins 和 API 實(shí)現(xiàn)自動(dòng)化。我們通過部署燒瓶服務(wù)器并使用 boto3 創(chuàng)建資源來自動(dòng)創(chuàng)建 DMS 資源。

我們幾乎所有的基礎(chǔ)設(shè)施/資源都是通過 Terraform 創(chuàng)建的。SRE 在建立我們的大部分?jǐn)?shù)據(jù)平臺(tái)基礎(chǔ)設(shè)施方面發(fā)揮了重要作用。

3. 記錄、監(jiān)控和警報(bào)

盡管我們的基礎(chǔ)設(shè)施是健壯的、容錯(cuò)的和高度可擴(kuò)展的，但有時(shí)會(huì)出現(xiàn)可能導(dǎo)致基礎(chǔ)設(shè)施停機(jī)的意外錯(cuò)誤。為了識(shí)別和解決這些問題，我們使用 Cloud watch 和 EFK(Elasticsearch、Fluentbit 和 Kibana)堆棧對我們數(shù)據(jù)平臺(tái)中涉及的每個(gè)組件啟用了監(jiān)控和警報(bào)。

4. 工作流程編排

任何數(shù)據(jù)平臺(tái)都需要調(diào)度能力來運(yùn)行批處理數(shù)據(jù)管道。由于我們已經(jīng)在之前的平臺(tái)中使用 Airflow 進(jìn)行工作流編排，因此我們繼續(xù)使用相同的編排工具。MWAA 已經(jīng)在減少維護(hù)工作量和節(jié)省成本方面發(fā)揮了很大作用。我們在之前的博客中解釋了我們在 MWAA 中評估的內(nèi)容。

五、概括

在這篇文章中，我們查看了 Lake House 架構(gòu)、構(gòu)建平臺(tái) 2.0 所涉及的所有組件，以及我們將 HUDI 用作數(shù)據(jù)湖的關(guān)鍵要點(diǎn)。由于我們現(xiàn)在已經(jīng)構(gòu)建了 Data Platform 2.0 的基礎(chǔ)部分，接下來我們計(jì)劃專注于平臺(tái)的以下方面：

• 數(shù)據(jù)質(zhì)量 -> 維護(hù)整個(gè)數(shù)據(jù)存儲(chǔ)的數(shù)據(jù)檢查和數(shù)據(jù)一致性。
• 數(shù)據(jù)血緣 -> 提供數(shù)據(jù)轉(zhuǎn)換的端到端步驟。
• BI 團(tuán)隊(duì)的自助服務(wù)平臺(tái) -> 減少對 DE 團(tuán)隊(duì)對入職報(bào)告表的依賴。
• 處理遲到的維度：保持我們的數(shù)據(jù)模型的一致性，并處理從湖到倉庫的遲到的維度鍵。