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 類路徑。再通過代碼，來看看反序列化中的細節：
publicvoid 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); } elseif (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); } elseif (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 如下所示。
privateRowTypecreateJsonRowType(DataType databaseSchema) { // Maxwell JSON contains other information, e.g. "database", "ts"// but we don't need themreturn (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 中的實現。
PR 實現詳情連結： https://github.com/apache/flink/pull/13090
功能演示
給大家演示一下從 Kafka 中讀取 Maxwell 推送來的 maxwell json 格式數據，並將聚合後的數據再次寫入 Kafka 後，重新讀出來驗證數據是否正確。
Kafka 數據源表
CREATETABLE topic_products ( -- schema is totally the same to the MySQL "products" tableidBIGINT, nameSTRING, description STRING, weight DECIMAL(10, 2)) WITH ('connector' = 'kafka','topic' = 'maxwell','properties.bootstrap.servers' = 'localhost:9092','properties.group.id' = 'testGroup','format' = 'maxwell-json');
Kafka 數據結果表&數據源表
CREATETABLE topic_sink ( nameSTRING, 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 中的表CREATETABLE product (idINTEGERNOTNULL AUTO_INCREMENT PRIMARY KEY,nameVARCHAR(255),description VARCHAR(512),weight FLOAT);truncate product ;ALTERTABLE product AUTO_INCREMENT = 101;INSERTINTO productVALUES (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'WHEREid=106;UPDATE product SET weight='5.1'WHEREid=107;INSERTINTO product VALUES (default,"jacket","water resistent white wind breaker",0.2);INSERTINTO product VALUES (default,"scooter","Big 2-wheel scooter ",5.18);UPDATE product SET description='new water resistent white wind breaker', weight='0.5'WHEREid=110;UPDATE product SET weight='5.17'WHEREid=111;DELETEFROM product WHEREid=111;UPDATE product SET weight='5.17'WHEREid=102orid = 101;DELETEFROM product WHEREid=102orid = 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 數據！