Part 01

Apache Paimon是什么？  

Apache Paimon是一種多功能的流數(shù)據(jù)湖平臺(tái)，支持高速數(shù)據(jù)攝取、變更數(shù)據(jù)跟蹤和實(shí)時(shí)分析，它為讀/寫操作提供靈活的架構(gòu)，并與各種計(jì)算引擎（如Apache Flink、Apache Hive、Apache Spark和Trino）集成。Paimon利用列式文件存儲(chǔ)和LSM樹結(jié)構(gòu)進(jìn)行高效的數(shù)據(jù)更新和查詢。它提供連接器，用于消息隊(duì)列、OLAP系統(tǒng)和批量存儲(chǔ)的統(tǒng)一存儲(chǔ)。Paimon的表抽象可以無縫的批處理和流處理執(zhí)行模式，用于數(shù)據(jù)處理。

圖1 Apache Paimon架構(gòu)圖

Part 02

Apache Paimon的特點(diǎn) 

Apache Paimon作為一個(gè)數(shù)據(jù)湖平臺(tái)，目前具有以下幾個(gè)主要特點(diǎn)：（1）大規(guī)模實(shí)時(shí)更新；（2）數(shù)據(jù)表局部更新；（3）流批一體讀寫。

- 大規(guī)模實(shí)時(shí)更新

Paimon 是一種新穎的數(shù)據(jù)存儲(chǔ)系統(tǒng)，它結(jié)合了湖存儲(chǔ)、LSM 和列式格式（如ORC、Parquet）等多種技術(shù)，為湖存儲(chǔ)帶來了大規(guī)模實(shí)時(shí)更新能力。其中，LSM 數(shù)據(jù)結(jié)構(gòu)的追加寫能力是 Paimon 實(shí)現(xiàn)高性能的關(guān)鍵。Paimon 的設(shè)計(jì)使得它可以在大規(guī)模的數(shù)據(jù)輸入場(chǎng)景中提供出色的性能表現(xiàn)，同時(shí)支持快速的查詢和分析操作。其特點(diǎn)如下：

• 高容錯(cuò)性：LSM 的多版本存儲(chǔ)機(jī)制，保障數(shù)據(jù)可靠性和恢復(fù)能力
• 高可擴(kuò)展性：LSM 的水平擴(kuò)展能力非常強(qiáng)，可以支持 PB 級(jí)別的數(shù)據(jù)規(guī)模
• 高靈活性：Paimon 支持多種列式格式，可以根據(jù)不同的業(yè)務(wù)需求選擇最適合的格式
• 高可定制性：Paimon 提供了豐富的配置選項(xiàng)，可以根據(jù)不同的場(chǎng)景進(jìn)行優(yōu)化和定制。

- 數(shù)據(jù)表局部更新

在數(shù)據(jù)倉庫的業(yè)務(wù)場(chǎng)景中，寬表數(shù)據(jù)模型是非常常見的。它是指將業(yè)務(wù)主體相關(guān)的指標(biāo)、維表和屬性關(guān)聯(lián)在一起的模型表，也可以泛指將多個(gè)事實(shí)表和多個(gè)維度表相關(guān)聯(lián)到一起形成的寬表。這種模型能夠幫助我們更好地理解業(yè)務(wù)數(shù)據(jù)，提高數(shù)據(jù)分析的效率。Paimon開發(fā)了一個(gè) Partial-Update 合并引擎。它可以根據(jù)相同的主鍵實(shí)時(shí)合并多條流，形成 Paimon 的一張大寬表。而且，借助 它里面的LSM 樹的延遲 Compaction 機(jī)制，我們可以用較低的成本完成合并，從而提高了數(shù)據(jù)處理的效率。舉例來說：當(dāng)收到主鍵為1的以下三條數(shù)據(jù)后，它最終會(huì)合并成合并的一條數(shù)據(jù)。

## 輸入
?<1, 23.0, 10, NULL>
?<1, NULL, NULL, 'This is a book'>
?<1, 25.2, NULL, NULL>
## 輸出
<1, 25.2, 10, 'This is a book'>

同時(shí)，合并后的表可以提供批讀和流讀：

• 批讀：在批讀時(shí)，讀時(shí)合并仍然可以完成 Projection Pushdown，提供高性能的查詢。
• 流讀：下游可以看到完整的、合并后的數(shù)據(jù)，而不是部分列。

- 流批一體讀寫

作為一個(gè)流批一體的數(shù)據(jù)湖存儲(chǔ)，Paimon提供了流寫流讀和批寫批讀的功能。我們可以利用這些特性來構(gòu)建Streaming Pipeline，并將數(shù)據(jù)沉淀到Paimon存儲(chǔ)中。在使用 Paimon進(jìn)行數(shù)據(jù)處理的過程中，它不僅可以實(shí)時(shí)更新Flink Streaming作業(yè)的數(shù)據(jù)，還能夠支持OLAP查詢各個(gè)Paimon表的歷史和實(shí)時(shí)數(shù)據(jù)。此外，還可以通過Batch SQL對(duì)之前的分區(qū)進(jìn)行回填，實(shí)現(xiàn)批讀批寫的功能，從而更加高效地進(jìn)行數(shù)據(jù)處理。

Part 03

Apache Paimon的數(shù)據(jù)結(jié)構(gòu) 

在Paimon中一張表的所有數(shù)據(jù)文件都存在一個(gè)層級(jí)的目錄中。其中第一層包含3個(gè)文件夾，分別是snapshot、manifest、schema和data。snapshot文件夾主要用于存儲(chǔ)這個(gè)表的快照,內(nèi)容包括為上一次提交產(chǎn)生的 manifest，加上本次提交產(chǎn)生的 manifest 作為增量。schema文件夾主要用于存儲(chǔ)這個(gè)表的元信息。manifest文件夾主要用于存儲(chǔ)這個(gè)一系列manifest文件，manifest記錄了每次經(jīng) checkpoint 觸發(fā)而提交的數(shù)據(jù)文件變更，包含新增和刪除的數(shù)據(jù)文件。Data文件夾按桶進(jìn)行劃分。每個(gè)桶文件夾包含一個(gè)LSM樹和changelog文件。

圖2 Apache Paimon文件層級(jí)圖

其文件的更新機(jī)制如下：在Apache Paimon中，會(huì)在Sink端維護(hù)一個(gè)Memory Table，用作數(shù)據(jù)合并，數(shù)據(jù)會(huì)寫入到File Store和 Log Store當(dāng)中，F(xiàn)ile Store中保存的就是經(jīng)過桶分區(qū)的LSM樹存儲(chǔ)結(jié)構(gòu)，Log Store則是保存了LSM中的 Write Ahead Log 信息。對(duì)于批讀，只需要去讀取File Store；而對(duì)于流讀，則需要混合的讀取，先讀取File Store 中的全量數(shù)據(jù)，再通過Log Store讀取變更的數(shù)據(jù)。

圖3 Apache Paimon讀寫機(jī)制

Part 04

Apache Paimon在威脅情報(bào)中的應(yīng)用 

中國移動(dòng)智慧家庭運(yùn)營中心威脅情報(bào)云平臺(tái)是基于中國移動(dòng)網(wǎng)絡(luò)和數(shù)據(jù)資源優(yōu)勢(shì)構(gòu)建的。通過應(yīng)用威脅情報(bào)挖掘技術(shù)和運(yùn)營，該平臺(tái)為安全產(chǎn)品和安全分析人員提供豐富的惡意IP/域名/樣本IOC、whois、PDNS等情報(bào)查詢服務(wù)，幫助企業(yè)以較低的成本享受專業(yè)的威脅情報(bào)服務(wù)，更好地了解和應(yīng)對(duì)網(wǎng)絡(luò)威脅，加強(qiáng)企業(yè)的安全防護(hù)能力。

在該平臺(tái)上，業(yè)務(wù)分析人員需要仔細(xì)分析表中的數(shù)據(jù)完成每周的報(bào)表統(tǒng)計(jì)。然而，原始業(yè)務(wù)數(shù)據(jù)存儲(chǔ)在mongoDB上，因此在大數(shù)據(jù)量的情況直接對(duì)mongoDB進(jìn)行分析操作必定會(huì)對(duì)業(yè)務(wù)產(chǎn)生影響。為了減少對(duì)業(yè)務(wù)的影響，業(yè)務(wù)需要將mongoDB表導(dǎo)入到大數(shù)據(jù)平臺(tái)進(jìn)行分析。考慮到情報(bào)數(shù)據(jù)的規(guī)模達(dá)到億級(jí)別，并且每天都會(huì)發(fā)生情報(bào)老化以及頻繁更新的情況，如果每天都定時(shí)進(jìn)行全量更新，那將會(huì)耗費(fèi)大量資源且效率低下。

因此，為了解決這個(gè)問題，我們采用了Flink CDC技術(shù)和Apache Paimon數(shù)據(jù)湖。通過Flink CDC采集mongoDB的oplog，我們能夠?qū)崿F(xiàn)數(shù)據(jù)的增量更新到Apache Paimon中，從而提高了同步效率并降低了資源消耗。這種方法使得數(shù)據(jù)更新更加高效且無需大量的資源投入。

下面是一個(gè)Flink SQL通過CDC同步mongoDB情報(bào)表到paimon的例子。首先創(chuàng)建一張dim_mongo_threaten_score的mongodb-cdc表，接著創(chuàng)建一張dim_fts_threaten_score的paimon表，最后把dim_mongo_threaten_score導(dǎo)入到dim_fts_threaten_score。

CREATE TABLE IF NOT EXISTS dim_mongo_threaten_score (
    _id STRING,
    threaten_type  STRING,
    credit_level STRING,
    threaten_score STRING,
    threaten_source STRING,
    created_time TIMESTAMP(3),
    updated_time TIMESTAMP(3),
    attack_time STRING,
    source STRING,
    ip STRING,
    domain STRING,
    device STRING,
    iphone STRING,
    PRIMARY KEY(_id) NOT ENFORCED
) WITH (
    'connector' = 'mongodb-cdc',
    'hosts' = '......',
    'username' = '......',
    'password' = '.....',
    'database' = '......',
    'collection' = '......',
    'heartbeat.interval.ms' = '......',
    'poll.max.batch.size' = '......'
);
CREATE CATALOG fts_catalog WITH (
    'type'='table-store',
    'warehouse'='hdfs:///......'
);
USE CATALOG fts_catalog;
CREATE TABLE IF NOT EXISTS dim_fts_threaten_score (
    _id STRING,
    threaten_type  STRING,
    credit_level STRING,
    threaten_score STRING,
    threaten_source STRING,
    created_time TIMESTAMP(3),
    updated_time TIMESTAMP(3),
    attack_time STRING,
    source STRING,
    ip STRING,
    domain STRING,
    device STRING,
    iphone STRING,
    PRIMARY KEY(_id) NOT ENFORCED
) with (
    'bucket' = '......',
    'snapshot.time-retained' = '......'
);
insert into fts_catalog.`default`.dim_fts_threaten_score 
select * from default_catalog.default_database.dim_mongo_threaten_score;

從上面的腳本可以看出，我們采用了非常簡(jiǎn)潔高效的方法來實(shí)現(xiàn)從mongoDB到Apache Paimon的數(shù)據(jù)增量同步。只需要建立兩張表并添加一個(gè)簡(jiǎn)單的insert語句，就可以完成整個(gè)同步過程。最后，從同步效率上來看，從原先的天級(jí)延遲到現(xiàn)在的秒級(jí)延遲，其提升顯著；從資源消耗上來看，CPU從原先的8核減少到現(xiàn)有的4核，其提升也非常明顯。

Part 05

總結(jié)展望  

Apache Paimon做為新一代數(shù)據(jù)湖，其支持高速數(shù)據(jù)攝取、變更數(shù)據(jù)跟蹤和實(shí)時(shí)分析，并為讀/寫操作提供靈活的架構(gòu)，并與各種計(jì)算引擎集成。由于其強(qiáng)大的性能，目前已完成了CDC同步業(yè)務(wù)數(shù)據(jù)庫數(shù)據(jù)到數(shù)據(jù)湖的場(chǎng)景，實(shí)現(xiàn)了自動(dòng)化數(shù)據(jù)集成。在未來，也可以基于PartialUpdate機(jī)制實(shí)現(xiàn)準(zhǔn)實(shí)時(shí)寬表，解決大寬表延遲高和資源浪費(fèi)的情況。同時(shí)，也可以基于AppendOnly機(jī)制來替換消息隊(duì)列，達(dá)到解耦和降本增效的效果。