簡介： 為了消滅數據孤島，企業往往會把各個組織的數據都接入到數據湖以提供統一的查詢或分析。本文將介紹流利說當前數據接入的整個過程，期間遇到的挑戰，以及delta在數據接入中產生的價值。

?

背景

流利說目前的離線計算任務中，大部分數據源都是來自于業務 DB，業務DB數據接入的準確性、穩定性和及時性，決定著下游整個離線計算 pipeline 的準確性和及時性。同時，我們還有部分業務需求，需要對 DB 中的數據和 hive 中的數據做近實時的聯合查詢。
在引入阿里云 EMR Delta Lake 之前，我們通過封裝 DataX 來完成業務 DB 數據的接入，采用 Master-Slave 架構，Master 維護著每日要執行的 DataX 任務的元數據信息，Worker 節點通過不斷的以搶占的方式獲取狀態為 init 和 restryable 的 DataX 任務來執行，直到當天的所有的 DataX 任務全都執行完畢為止。

架構圖大致如下:

Worker 處理的過程如下:

對于近實時需求，我們是直接開一個從庫，配置 presto connector 去連接從庫，來實現業務 BD 中的數據和 hive 中的數據做近實時的聯合查詢需求。

這種架構方案的優點是簡單，易于實現。但是隨著數據量也來越多，缺點也就逐漸暴露出來了:
性能瓶頸: 隨著業務的增長，這種通過 SELECT 的方式接入數據的性能會越來越差，受 DB 性能瓶頸影響，無法通過增加 Worker 節點的方式來緩解。
規模大的表只能通過從庫來拉取，造成數據接入的成本越來越高。
無法業務滿足近實時的查詢需求，近實時查詢只能通過從庫的方式查詢，進一步加大了接入的成本。
為了解決這些問題，我們將目光聚焦到了 CDC實時接入的方案上。

技術方案選型

對于 CDC實時接入的方案，目前業內主要有以下幾種: CDC + Merge 方案、CDC + Hudi、CDC + Delta Lake 及 CDC + Iceberg 等幾種方案。其中，CDC + Merge 方案是在是在數據湖方案出現之前的做法，這種方案能節省DB從庫的成本，但是無法滿足業務近實時查詢的需求等功能，所以最開始就 pass 掉了，而 Iceberg 在我們選型之初，還不夠成熟，業界也沒有可參考的案列，所以也被 pass 掉了，最后我們是在 CDC + Hudi 和 CDC + Delta Lake 之間選擇。
在選型時，Hudi 和 Delta Lake 兩者的功能上都是大同小異的，所以我們主要是從這幾方案來考慮的: 穩定性、小文件合并、是否支持SQL、云廠商支持程度、語言支持程度等幾個方面來考慮。

基于以上指標，加上我們整個數據平臺都是基于阿里云 EMR 搭建的，選擇 Delta Lake 的話，會省掉大量的適配開發工作，所以我們最終選擇了 CDC + Delta Lake 的方案。
整體架構

?

總體架構圖

整體的架構如上圖所示。我們接入的數據會分為兩部分，存量歷史數據和新數據，存量歷史數據使用 DataX 從 MySQL 中導出，存入 OSS 中，新數據使用 Binlog 采集存入 Delta Lake 表中。每日凌晨跑 ETL 任務前，先對歷史數據和新數據做 Merge 操作，ETL 任務使用 Merge 之后的數據。

Delta Lake 數據接入
在 Binlog 實時采集方面，我們采用了開源的 Debezium ，負責從 MySQL 實時拉取 Binlog 并完成適當解析，每張表對應一個 Topic ，分庫分表合并為一個 Topic 分發到 Kafka 上供下游消費。Binlog 數據接入到 Kafka 之后，我們需要創建 Kafka Source 表指向對應的 Kafka Topic 中, 表的格式為:

CREATE TABLE kafka_{db_name}_{table_name} (key BINARY, value BINARY, topic STRING, partition INT, offset BIGINT, timestamp TIMESTAMP, timestampType INT)
USING kafka
OPTIONS (
kafka.sasl.mechanism 'PLAIN',
subscribe 'cdc-{db_name}-{table_name}',
serialization.format '1',
kafka.sasl.jaas.config '*****(redacted)',
kafka.bootstrap.servers '{bootstrap-servers}',
kafka.security.protocol 'SASL_PLAINTEXT'
)
我們主要用到的字段是 value 和 offset ，其中 value 的格式如下:

{
"payload": {
"before": {
db記錄變更前的schema及內容，op=c時，為null
},
"after": {
db記錄變更后的schema及內容，op=d時，為null
},
"source": {
ebezium配置信息
},
"op": "c",
"ts_ms":
}
}
同時創建 Delta Lake 表，Location 指向 HDFS 或者 OSS ，表結構為:

CREATE TABLE IF NOT EXISTS delta.delta_{dbname}{table_name}(
{row_key_info},
ts_ms bigint,
json_record string,
operation_type string,
offset bigint
)
USING delta
LOCATION '------/delta/{db_name}.db/{table_name}'
其中 row_key_info 為 Delta Lake 表的唯一索引字段，對于單庫單表而言，row_key_info 為 mysql 表的 primary key 字段 eg: id long，對于分庫分表及分實例分庫分表而言，row_key_info 為分庫分表的字段和單表里primary key 字段組成，eg: 以 user_id 為分表字段，每張表里以 id 為 primary key , 那么對應的 row_key_info 為 id long, user_id long。
StreamingSQL 處理 Kafka 中的數據，我們主要是提取 Kafka Source 表中的 offset、value 字段及 value 字段中的 CDC 信息如: op、ts_ms 及 payload 的 after 和 before 字段。StreamingSQL 中，我們采用 5min 一個 mini batch，主要是考慮到 mini batch 太小會產生很多小文件，處理速度會越來越慢，也會影響讀的性能，太大了又沒法滿足近實時查詢的要求。而 Delta Lake 表，我們不將 after 或者 before 字段解析出來，主要是考慮到我們業務表 的 schema 經常變更，業務表 schema 一變更就要去修復一遍數據，成本比較大。在 StreamingSQL 處理過程中，對于 op=’c’ 的數據我們會直接 insert 操作，json_record 取 after 字段。對于 op=’u’ 或者 op=’d’ 的數據，如果 Delta Lake 表中不存在，那么執行 insert 操作, 如果存在，那么執行 update 操作；json_record 的賦值值，op=’d’，json_record 取 before 字段，op=’u’，jsonrecord 取 after 字段。保留 op=’d’ 的字段，主要是考慮到刪除的數據可能在存量歷史表中，如果直接刪除的話，凌晨 merge 的數據中，存在存量歷史表中的數據就不會被刪除。
整個 StreamingSQL 的處理大致如下:
CREATE SCAN incremental{dbname}{tablename} on kafka{dbname}{table_name} USING STREAM
OPTIONS(
startingOffsets='earliest',
maxOffsetsPerTrigger='1000000',
failOnDataLoss=false
);
CREATE STREAM job
OPTIONS(
checkpointLocation='------/delta/{db_name}.db/{table_name}checkpoint',
triggerIntervalMs='300000'
)
MERGE INTO delta.delta{dbname}{table_name} as target
USING (
SELECT * FROM (
SELECT ts_ms, offset, operation_type, {key_column_sql}, coalesce(after_record, before_record) as after_record, row_number() OVER (PARTITION BY {key_column_partition_sql} ORDER BY ts_ms DESC, offset DESC) as rank
FROM (
SELECT ts_ms, offset, operation_type, before_record, after_record, {key_column_include_sql}
FROM ( SELECT get_json_object(string(value), '$.payload.op') as operation_type, get_json_object(string(value), '$.payload.before') as before_record,
get_json_object(string(value), '$.payload.after') as after_record, get_json_object(string(value), '$.payload.ts_ms') as tsms,
offset
FROM incremental{dbname}{table_name}
) binlog
) binlog_wo_init ) binlog_rank where rank = 1) as source
ON {key_column_condition_sql}
WHEN MATCHED AND (source.operation_type = 'u' or source.operation_type='d') THEN
UPDATE SET {set_key_column_sql}, ts_ms=source.ts_ms, json_record=source.after_record, operation_type=source.operation_type, offset=source.offset
WHEN NOT MATCHED AND (source.operation_type='c' or source.operation_type='u' or source.operation_type='d') THEN
INSERT ({inser_key_column_sql}, ts_ms, json_record, operation_type, offset) values ({insert_key_column_value_sql}, source.ts_ms, source.after_record, source.operation_type, source.offset);
執行完 StreamingSQL 之后，就會生成如下格式的數據：

其中 part-xxxx.snappy.parquet 保存的是 DeltaLake 表的數據文件，而 _deltalog 目錄下保存的是 DeltaLake 表的元數據，包括如下:
其中 xxxxxxxx 表示的是版本信息，xxxxxxxx.json 文件里保存的是有效的 parquet 文件信息，其中 add 類型的為有效的 parquet 文件， remove 為無效的 parquet 文件。
Delta Lake 是支持 Time travel 的，但是我們 CDC 數據接入的話，用不到數據回滾策略，如果多版本的數據一直保留會給我們的存儲帶來一定的影響，所以我們要定期刪除過期版本的數據，目前是僅保留2個小時內的版本數據。同時，Delta Lake 不支持自動合并小文件的功能，所以我們還需要定期合并小文件。目前我們的做法是，每小時通過 OPTIMIZE 和 VACCUM 來做一次合并小文件操作及清理過期數據文件操作:
optimize delta{dbname}{tablename};
set spark.databricks.delta.retentionDurationCheck.enabled = false;
VACUUM delta{dbname}{table_name} RETAIN 1 HOURS;
由于目前 Hive 和 Presto 無法直接讀取 Spark SQL 創建的 Delta Lake 表，但是監控及近實時查詢需求，需要查詢 Delta Lake 表，所以我們還創建了用于 Hive 和 Presto 表查詢的。
Delta Lake 數據與存量數據 Merge
由于 Delta Lake 的數據我們僅接入新數據，對于存量歷史數據我們是通過DataX 一次性導入的，加上 Delta Lake 表 Hive 無法直接查詢，所以每日凌晨我們需要對這兩部分數據做一次 merge 操作，寫入到新的表中便于 Spark SQL 和 Hive 統一使用。這一模塊的架構大致如下:
圖片

每日凌晨0點前，調用 DeltaService API ，根據 Delta Lake 任務的配置自動生成 merge任務 的 task 信息、spark-sql 腳本及 對應的 Airflow DAG 文件。
merge 任務的 task 信息主要包括如下信息:

自動生成 Merge 腳本，主要是從 Delta Lake 任務的配置中獲取 mysql 表的schema 信息，刪掉歷史的 Hive 表，再根據 schema 信息重新創建 Hive 外部表，再根據新的 schema 從Delta Lake表的 json_record 字段和歷史存量數據表中獲取對應的字段值做 union all 操作，缺失值采用mysql 的默認值, union 之后，再根據 row_key 進行分組，按 ts_ms 排序取第一條，同時取出operation_type=’d’ 的數據。整體如下:
CREATE DATABASE IF NOT EXISTS {db_name} LOCATION '------/delta/{db_name}.db';
DROP TABLE IF EXISTS {db_name}.{table_name};
CREATE TABLE IF NOT EXISTS {db_name}.{table_name}(
{table_column_infos}
)
STORED AS PARQUET
LOCATION '------/delta/{db_name}.db/{table_name}/data_date=${{data_date}}';
INSERT OVERWRITE TABLE {db_name}.{table_name}
SELECT {table_columns}
FROM ( SELECT {table_columns}, _operation_type, row_number() OVER (PARTITION BY {row_keys} ORDER BY ts_ms DESC) as ranknum
FROM (
SELECT {delta_columns}, operation_type as _operation_type, tsms
FROM delta{dbname}{table_name}
UNION ALL
SELECT {hive_columns}, 'c' as _operation_type, 0 as ts_ms
FROM {db_name}.{table_name}_delta_history
) union_rank
) ranked_data
WHERE ranknum=1
AND _operation_type <> 'd'
凌晨0點之后，Airflow 會根據 Airflow DAG 文件自動調度執行 merge 的Spark SQL 腳本，腳本執行成功后，更新 merge task 的狀態為 succeed ，Airflow 的 ETL DAG 會根據merge task 的狀態自動調度下游的 ETL 任務。
Delta Lake 數據監控
對于 Delta Lake 數據的監控，我們主要是為了兩個目的：監控數據是否延遲及監控數據是否丟失，主要是在 MySQL 與 Delta Lake 表之間及 CDC 接入過來的 Kafka Topic 與 Delta Lake 表之間。
CDC 接入過來的 Kafka Topic 和 Delta Lake 表之間的延遲監控：我們是每15分鐘從 Kafka 的 Topic 中獲取每個 Partition 的最大 offset 對應的 mysql 的 row_key 字段內容，放入監控的 MySQL 表 delta_kafka_monitor_info 中，再從 delta_kafka_monitor_info 中獲取上一周期的 row_key 字段內容，到 Delta Lake 表中查詢，如果查詢不到，說明數據有延遲或者丟失，發出告警。
MySQL 與 Delta Lake 之間的監控：我們有兩種，一種是探針方案，每15分鐘，從 MySQL 中獲取最大的 id，對于分庫分表，只監控一張表的，存入 delta_mysql_monitor_info 中，再從 delta_mysql_monitor_info 中獲取上一周期的最大 id，到 Delta Lake 表中查詢，如果查詢不到，說明數據有延遲或者丟失，發出告警。另一種是直接 count(id)，這種方案又分為單庫單表和分庫分表兩種，元數據保存在 mysql 表 id_based_mysql_delta_monitor_info 中，主要包含 min_id、max_id、mysql_count 三個字段，對于單庫單表，也是每隔5分鐘，從 Delta Lake 表中獲取 min_id 和 max_id 之間的 count 值，跟 mysql_count 對比，如果小于 mysql_count 值說明有數據丟失或者延遲，發出告警。再從 mysql 中獲取 max(id) 和 max_id 與 max(id) 之間的 count 值，更新到 id_based_mysql_delta_monitor_info 表中。對于分庫分表的情況，根據分庫分表規則，生成每一張表對應的 id_based_mysql_delta_monitor_info 信息，每半小時執行一遍監控，規則同單庫單表。

遇到的挑戰

業務表 schema 變更頻繁，Delta Lake 表如果直接解析 CDC 的字段信息的話，如果不能及時發現并修復數據的話，后期修復數據的成本會較大，目前我們是不解析字段，等到凌晨 merge 的時候再解析。
隨著數據量越來越大，StreamingSQL 任務的性能會越來越差。我們目前是 StreamingSQL 處理延遲，出現大量延遲告警后，將 Delta Lake 存量數據替換成昨日 merge 后的數據，再刪掉 Delta Lake 表，刪除 checkpoint 數據，從頭開始消費 KafkaSource 表的數據。降低 Delta Lake 表數據，從而緩解StreamingSQL 的壓力。
Hive 和 Presto 不能直接查詢 Spark SQL 創建的 Delta Lake 表，目前我們是創建支持 Hive 和 Presto 查詢的外部表來供 Hive 和 Presto 使用，但是這些表又無法通過 Spark SQL 查詢。所以上層 ETL 應用無法在不更改代碼的情況下，在 Hive 和 Spark SQL 及Presto 引擎之間自由切換。

帶來的收益

節省了 DB 從庫的成本，采用 CDC + Delta Lake 之后，我們的成本節省了近80%。
凌晨 DB 數據接入的時間成本大大降低，能夠確保所有非特殊要求的 DB 數據接入都能在1個小時內跑完。

后續規劃

StreamingSQL 任務隨著 Delta Lake 表數據量越來越大，性能越來越差問題跟進。
推動能否解決 Spark SQL 創建的 Delta Lake 表，無法直接使用 Hive 和 Presto 查詢的問題。

后續規劃

StreamingSQL 任務隨著 Delta Lake 表數據量越來越大，性能越來越差問題跟進。
推動能否解決 Spark SQL 創建的 Delta Lake 表，無法直接使用 Hive 和 Presto 查詢的問題。

原文鏈接?

總結

以上是生活随笔為你收集整理的阿里云 EMR Delta Lake 在流利说数据接入中的架构和实践的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。