Flink 原始碼 | 自定義 Format 消費 Maxwell CDC 資料
Flink 1.11 最重要的 Feature —— Hive Streaming 之前已經和大家分享過了，今天就和大家來聊一聊另一個特別重要的功能 —— CDC。
CDC概述
何為CDC？Change Data Capture，將資料庫中的’增’、’改’、’刪’操作記錄下來。在很早之前是通過觸發器來完成記錄，現在通過 binlog+同步中介軟體來實現。常用的 binlog 同步中介軟體有很多，比如 Alibaba 開源的 canal[1]，Red Hat 開源的debezium[2]，Zendesk 開源的 Maxwell[3] 等等。
這些中介軟體會負責 binlog 的解析，並同步到訊息中介軟體中，我們只需要消費對應的 Topic 即可。
回到 Flink 上，CDC 似乎和我們沒有太大的關聯？其實不然，讓我們更加抽象地來看這個世界。
當我們用 Flink 去消費資料比如 Kafka 時，我們就彷彿在讀一張表，什麼表？一張不斷有記錄被插入的表，我們將每一條被插入的資料取出來，完成我們的邏輯。
當插入的每條資料都沒有問題時，一切都很美好。關聯、聚合、輸出。
但當我們發現，某條已經被計算過的資料有問題時，麻煩大了。我們直接改最後的輸出值其實是沒有用的，這次改了，當再來資料觸發計算時，結果還是會被錯誤的資料覆蓋，因為中間計算結果沒有被修改，它仍然是一個錯誤的值。怎麼辦？撤回流似乎能解決這個問題，這也確實是解決這個問題的手段，但是問題來了，撤回流怎麼確定讀取的資料是要被撤回的？另外，怎麼去觸發一次撤回？
CDC 解決了這些：將訊息中介軟體的資料反序列化後，根據 Type 來識別資料是 Insert 還是 Delete；另外，如果大家看過 Flink 原始碼，會發現反序列化後的資料型別變了，從 Row 升級為 RowData，RowData 能夠將資料標記為撤回還是插入，這就意味著每個運算元能夠判斷出資料到底是需要下發還是撤回。
CDC 的重要性就先說這麼多，之後有機會的話，出一篇實時 DQC 的視訊，告訴大家 CDC 的出現，對於實時 DQC 的幫助有多大。下面讓我們回到正題。
既然有那麼多 CDC 同步中介軟體，那麼一定會有各種各樣的格式存放在訊息中介軟體中，我們必然需要去解析它們。於是 Flink 1.11 提供了 canal-json 和 debezium-json，但我們用的是 Maxwell 怎麼辦？只能等官方出或者說是等有人向社群貢獻嗎？那如果我們用的是自研的同步中介軟體怎麼辦？
所以就有了今天的分享：如何去自定義實現一個 Maxwell format。大家也可以基於此文的思路去實現其他 CDC format，比如 OGG, 或是自研 CDC 工具產生的資料格式。
如何實現
當我們提交任務之後，Flink 會通過 SPI 機制將 classpath 下注冊的所有工廠類載入進來，包括 DynamicTableFactory、DeserializationFormatFactory 等等。而對於 Format 來說，到底使用哪個 DeserializationFormatFactory，是根據 DDL 語句中的 Format 來決定的。通過將 Format 的值與工廠類的 factoryIdentifier() 方法的返回值進行匹配 來確定。
再通過 DeserializationFormatFactory 中的 createDecodingFormat(...) 方法，將反序列化物件提供給 DynamicTableSource。
通過圖來了解整個過程(僅從反序列化資料並消費的角度來看)：
想要實現 CDC Format 去解析某種 CDC 工具產生的資料其實很簡單，核心元件其實就三個：
- 工廠類（DeserializationFormatFactory）：負責編譯時根據 ‘format’ = ‘maxwell-json’建立對應的反序列化器。即 MaxwellJsonFormatFactory。
- 反序列化類（DeserializationSchema）：負責執行時的解析，根據固定格式將 CDC 資料轉換成 Flink 系統能認識的 INSERT/DELETE/UPDATE 訊息，如 RowData。即 MaxwellJsonDeserializationSchema。
- Service 註冊檔案：需要新增 Service 檔案 META-INF/services/org.apache.flink.table.factories.Factory ，並在其中增加一行我們實現的 MaxwellJsonFormatFactory 類路徑。
再通過程式碼，來看看反序列化中的細節：
public void deserialize(byte[] message, Collectorout) throws IOException {
try {
RowData row = jsonDeserializer.deserialize(message);
String type = row.getString(2).toString(); // "type" field
if (OP_INSERT.equals(type)) {
RowData insert = row.getRow(0, fieldCount);
insert.setRowKind(RowKind.INSERT);
out.collect(insert);
} else if (OP_UPDATE.equals(type)) {
GenericRowData after = (GenericRowData) row.getRow(0, fieldCount); // "data" field
GenericRowData before = (GenericRowData) row.getRow(1, fieldCount); // "old" field
for (int f = 0; f < fieldCount; f++) {
if (before.isNullAt(f)) {
before.setField(f, after.getField(f));
}
}
before.setRowKind(RowKind.UPDATE_BEFORE);
after.setRowKind(RowKind.UPDATE_AFTER);
out.collect(before);
out.collect(after);
} else if (OP_DELETE.equals(type)) {
RowData delete = row.getRow(0, fieldCount);
delete.setRowKind(RowKind.DELETE);
out.collect(delete);
} else {
if (!ignoreParseErrors) {
throw new IOException(format(
"Unknown \"type\" value \"%s\". The Maxwell JSON message is '%s'", type, new String(message)));
}
}
} catch (Throwable t) {
if (!ignoreParseErrors) {
throw new IOException(format(
"Corrupt Maxwell JSON message '%s'.", new String(message)), t);
}
}
}其實並不複雜：先通過 jsonDeserializer 將位元組陣列根據 [data: ROW, old: ROW, type: String] 的 schema 反序列化成 RowData，然後根據 “type” 列的值來判斷資料是什麼型別：增、改、刪；再根據資料型別取出 “data” 或者 “old” 區的資料，來組裝成 Flink 認識的 INSERT/DELETE/UPDATE 資料並下發。
物件 jsonDeserializer 即 JSON 格式的反序列化器，它可以通過指定的 RowType 型別，讀取 JSON 的位元組陣列中指定的欄位並反序列化成 RowData。在我們的場景中，我們需要去讀取如下 Maxwell 資料的 “data”, “old” 和 “type” 部分的資料。
{"database":"test","table":"product","type":"update","ts":1596684928,"xid":7291,"commit":true,"data":{"id":102,"name":"car battery","description":"12V car battery","weight":5.17},"old":{"weight":8.1}}因此 MaxwellJsonDeserializationSchema 中定義的 JSON 的 RowType 如下所示。
private RowType createJsonRowType(DataType databaseSchema) {
// Maxwell JSON contains other information, e.g. "database", "ts"
// but we don't need them
return (RowType) DataTypes.ROW(
DataTypes.FIELD("data", databaseSchema),
DataTypes.FIELD("old", databaseSchema),
DataTypes.FIELD("type", DataTypes.STRING())).getLogicalType();
}databaseSchema 是使用者通過 DDL 定義的 schema 資訊，也對應著資料庫中表的 schema。結合上面的 JSON 和程式碼，我們能夠得知 jsonDeserializer 只會取走 byte[] 中 data、old、type 這三個欄位對應的值，其中 data 和old 還是個巢狀JSON，它們的 schema 資訊和 databaseSchema 一致。由於 Maxwell 在同步資料時，“old”區不包含未被更新的欄位，所以 jsonDeserializer 返回後，我們會通過 “data” 區的 RowData 將 old 區的缺失欄位補齊。
得到 RowData 之後，會取出 type 欄位，然後根據對應的值，會有三種分支：
- insert：取出 data 中的值，也就是我們通過DDL定義的欄位對應的值，再將其標記為 RowKind.INSERT 型別資料，最後下發。
- update：分別取出 data 和 old 的值，然後迴圈 old 中每個欄位，欄位值如果為空說明是未修改的欄位，那就用 data 中對應位置欄位的值替代；之後將 old 標記為 RowKind.UPDATE_BEFORE 也就意味著 Flink 引擎需要將之前對應的值撤回，data 標記為 RowKind.UPDATE_AFTER 正常下發。
- delete：取出 data 中的值，標記為 RowKind.DELETE，代表需要撤回。
處理的過程中，如果丟擲異常，會根據 DDL 中maxwell-json.ignore-parse-errors的值來確定是忽視這條資料繼續處理下一條資料，還是讓任務報錯。
筆者在 maxwell-json 反序列化功能的基礎之上，還實現了序列化的功能，即能將 Flink 產生的 changelog 以 Maxwell 的 JSON 格式輸出到外部系統中。其實現思路與反序列化器的思路正好相反，更多細節可以參考 Pull Request 中的實現。
功能演示
給大家演示一下從 Kafka 中讀取 Maxwell 推送來的 maxwell json 格式資料，並將聚合後的資料再次寫入 Kafka 後，重新讀出來驗證資料是否正確。
Kafka 資料來源表
CREATE TABLE topic_products (
-- schema is totally the same to the MySQL "products" table
id BIGINT,
name STRING,
description STRING,
weight DECIMAL(10, 2)
) WITH (
'connector' = 'kafka',
'topic' = 'maxwell',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
'format' = 'maxwell-json');Kafka 資料結果表&資料來源表
CREATE TABLE topic_sink (
name STRING,
sum_weight DECIMAL(10, 2)
) WITH (
'connector' = 'kafka',
'topic' = 'maxwell-sink',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'testGroup',
'format' = 'maxwell-json'
);MySQL 表
-- 注意，這部分 SQL 在 MySQL 中執行，不是 Flink 中的表
CREATE TABLE product (
id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255),
description VARCHAR(512),
weight FLOAT
);
truncate product ;
ALTER TABLE product AUTO_INCREMENT = 101;
INSERT INTO product
VALUES (default,"scooter","Small 2-wheel scooter",3.14),
(default,"car battery","12V car battery",8.1),
(default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3",0.8),
(default,"hammer","12oz carpenter's hammer",0.75),
(default,"hammer","14oz carpenter's hammer",0.875),
(default,"hammer","16oz carpenter's hammer",1.0),
(default,"rocks","box of assorted rocks",5.3),
(default,"jacket","water resistent black wind breaker",0.1),
(default,"spare tire","24 inch spare tire",22.2);
UPDATE product SET description='18oz carpenter hammer' WHERE id=106;
UPDATE product SET weight='5.1' WHERE id=107;
INSERT INTO product VALUES (default,"jacket","water resistent white wind breaker",0.2);
INSERT INTO product VALUES (default,"scooter","Big 2-wheel scooter ",5.18);
UPDATE product SET description='new water resistent white wind breaker', weight='0.5' WHERE id=110;
UPDATE product SET weight='5.17' WHERE id=111;
DELETE FROM product WHERE id=111;
UPDATE product SET weight='5.17' WHERE id=102 or id = 101;
DELETE FROM product WHERE id=102 or id = 103;先看看能不能正常讀取 Kafka 中的 maxwell json 資料。
select * from topic_products;可以看到，所有欄位值都變成了 Update 之後的值，同時，被 Delete 的資料也沒有出現。
接著讓我們再將聚合資料寫入 Kafka。
insert into topic_sink select name,sum(weight) as sum_weight from topic_products group by name;在 Flink 叢集的 Web 頁面也能夠看到任務正確提交，接下來再讓我們把聚合資料查出來。
select * from topic_sink最後，讓我們查詢一下 MySQL 中的表，來驗證資料是否一致；因為在 Flink 中，我們將 weight 欄位定義成 Decimal(10,2)，所以我們在查詢 MySQL 的時候，需要將 weight 欄位進行型別轉換。
沒有問題，我們的 maxwell json 解析很成功。
寫在最後
根據筆者實現 maxwell-json format 的經驗，Flink 對於介面的定義、對於模組職責的劃分還是很清晰的，所以實現一個自定義 CDC format 非常簡單（核心程式碼只有200多行）。因此，如果你是用的 OGG，或是自研的同步中介軟體，可以通過本文的思路快速實現一個 CDC format，一起解放你的 CDC 資料！
原文連結
本文為阿里雲原創內容，未經允許不得轉載。