Flink + Debezium CDC 實現原理及代碼實戰

一、Debezium 介紹Debezium 是一個分布式平臺，它將現有的資料庫轉換為事件流，應用程式消費事件流，就可以知道資料庫中的每一個行級更改，並立即做出響應。Debezium 構建在 Apache Kafka 之上，並提供 Kafka 連接器來監視特定的資料庫。在介紹 Debezium 之前，我們要先了解一下什麼是 Kafka Connect。二、Kafka Connect 介紹Kafka 相信大家都很熟悉，是一款分布式，高性能的消息隊列框架。一般情況下，讀寫 Kafka 數據，都是用 Consumer 和 Producer  Api 來完成，但是自己實現這些需要去考慮很多額外的東西，比如管理 Schema，容錯，並行化，數據延遲，監控等等問題。而在 0.9.0.0 版本之後，官方推出了 Kafka Connect ，大大減少了程式設計師的工作量，它有下面的特性：REST 接口，用來查看和管理Kafka connectors；自動化的offset管理，開發人員不必擔心錯誤處理的影響；Kafka Connect 有兩個核心的概念：Source 和 Sink，Source 負責導入數據到 Kafka，Sink 負責從 Kafka 導出數據，它們都被稱為是 Connector。如下圖，左邊的 Source 負責從源數據（RDBMS，File等）讀數據到 Kafka，右邊的 Sinks 負責從 Kafka 消費到其他系統。三、Debezium 架構和實現原理Debezium 有三種方式可以實現變化數據的捕獲以插件的形式，部署在 Kafka Connect 上在上圖中，中間的部分是 Kafka Broker，而 Kafka Connect 是單獨的服務，需要下載  debezium-connector-mysql  連接器，解壓到伺服器指定的地方，然後在 connect-distribute.properties 中指定連接器的根路徑，即可使用。這種模式中，需要配置不同的連接器，從源頭處捕獲數據的變化，序列化成指定的格式，發送到指定的系統中。內嵌模式，既不依賴 Kafka，也不依賴 Debezium Server，用戶可以在自己的應用程式中，依賴 Debezium 的 api 自行處理獲取到的數據，並同步到其他源上。四、使用 Docker 來安裝 Debezium Kafka Mysql這裡我們使用官網提供的 Docker 方式快速的搭建一個演示環境。Docker 的安裝和基本命令，可以參考我之前的文章或者在網上找相關的教程。1. 首先獲取一個 zk 的鏡像

docker pull debezium/zookeeper
以 daemo 的方式運行鏡像，並且暴露 2181，2888，3888 埠
docker run -d -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper
--rm 表示容器停止後刪除本地數據-d 表示在後臺運行容器2. 獲取一個 kafka 的鏡像
docker pull debezium/kafka
docker run -d -it --rm --name kafka -p 9092:9092 -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://192.168.56.10:9092 -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092 --link zookeeper:zookeeper debezium/kafka
--link 表示可以和 zookeeper 容器互相通信3. 拉取一個 mysql 的鏡像
docker pull debezium/example-mysql
在後臺執行 mysql 的鏡像
docker run -d -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql
4. 單獨開一個 shell 進入 mysql 的命令行中
docker run -it --rm --name mysqlterm --link mysql --rm debezium/example-mysql sh -c 'exec mysql -h"$MYSQL_PORT_3306_TCP_ADDR" -P"$MYSQL_PORT_3306_TCP_PORT" -uroot -p"$MYSQL_ENV_MYSQL_ROOT_PASSWORD"'
5 拉取一個 debezium/connect 的鏡像
docker pull debezium/connect
啟動 kafka connect 服務
docker run -d -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect
啟動之後，我們可以使用 rest api 來檢查 Kafka Connect 的服務狀態
curl -H "Accept:application/json" localhost:8083/
{"version":"2.6.0","commit":"62abe01bee039651","kafka_cluster_id":"vkx8c6lhT1emLtPSi-ge6w"}
使用 rest api 來查看有多少 connect 服務註冊到 Kafka Connect 上了
curl -H "Accept:application/json" localhost:8083/connectors/
現在會返回一個空數組，表示還沒有服務註冊上去。
6 註冊一個 Connector 去檢測 mysql 資料庫的變化註冊的話，需要往 Kafka Connect 的 rest api 發送一個 Post 請求，請求內容如下3 task 最大數量，應該配置成 1，因為 Mysql 的 Connector 會讀取 Mysql 的 binlog，使用單一的任務才能保證合理的順序；4 這裡配置的是 mysql，其實是一個 host，如果非 docker 環境，則要配置成 ip 地址或者可以解析的域名；5 唯一的 serverId，會被作為 Kafka Topic 的前綴；執行下面的命令發送一個 Post 請求，註冊到 Kafka Connect 上：
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "inventory-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "mysql", "database.port": "3306", "database.user": "debezium", "database.password": "dbz", "database.server.id": "184054", "database.server.name": "dbserver1", "database.include.list": "inventory", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.inventory" } }'
7 新開一個 shell，啟動一個 Kafka 的消費者，來查看 Debezium 發送過來的事件
docker run -it --rm --name watcher --link zookeeper:zookeeper --link kafka:kafka debezium/kafka watch-topic -a -k dbserver1.inventory.customers
8 在 mysql 的命令行窗口上，修改一條數據
use inventory;
UPDATE customers SET first_name='Anne211' WHERE id=1004;
9 觀察 kafka 消費者窗口的變化
發現會發送過來兩條 json，一條是更新的哪個主鍵，一條是具體的更新內容
五、Flink 集成 Debezium 同步數據下面我們使用 Flink 來消費 Debezium 產生的數據，把變更的數據都同步到另外一張表中。主要步驟有：
package com.hudsun.flink.cdc;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author wangkai
 * @Time 2020/12/22 23:11
 */
public class DebeziumCDC {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
                StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings streamSettings = EnvironmentSettings.newInstance()
                .inStreamingMode().useBlinkPlanner().build();
        StreamTableEnvironment tableEnvironment =
                StreamTableEnvironment.create(env, streamSettings);
        tableEnvironment.getConfig().setSqlDialect(SqlDialect.DEFAULT);
        // debezium 捕獲到變化的數據會寫入到這個 topic 中
        String topicName = "dbserver1.inventory.customers";
        String bootStrpServers = "192.168.56.10:9092";
        String groupID = "testGroup";

        // 目標資料庫地址
        String url = "jdbc:mysql://192.168.56.10:3306/inventory";
        String userName = "root";
        String password = "debezium";
        String mysqlSinkTable = "customers_copy";


        // 創建一個 Kafka 數據源的表
        tableEnvironment.executeSql("CREATE TABLE customers (\n" +
                " id int,\n" +
                " first_name STRING,\n" +
                " last_name STRING,\n" +
                " email STRING \n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = '" + topicName + "',\n" +
                " 'properties.bootstrap.servers' = '" + bootStrpServers + "',\n"
                +
                " 'debezium-json.schema-include' = 'true',\n" +
                " 'properties.group.id' = '" + groupID + "',\n" +
                " 'format' = 'debezium-json'\n" +
                ")");

        // 創建一個寫入數據的 sink 表
        tableEnvironment.executeSql("CREATE TABLE customers_copy (\n" +
                " id int,\n" +
                " first_name STRING,\n" +
                " last_name STRING,\n" +
                " email STRING, \n" +
                " PRIMARY KEY (id) NOT ENFORCED\n" +
                ") WITH (\n" +
                " 'connector' = 'jdbc',\n" +
                " 'url' = '" + url + "',\n" +
                " 'username' = '" + userName + "',\n" +
                " 'password' = '" + password + "',\n" +
                " 'table-name' = '" + mysqlSinkTable + "'\n" +
                ")");
        String updateSQL = "insert into customers_copy select * from customers";
        TableResult result = tableEnvironment.executeSql(updateSQL);

        env.execute("sync-flink-cdc");
    }
}

最後的最後，推薦進我的微信群，每天都有在更新乾貨，公眾號回覆：進群，即可。