快捷導(dǎo)航

使用docker部署mysql并開啟binlog的方法

更新時間：2024年10月10日 10:41:32 作者：leiline

本文介紹了如何使用Docker部署MySQL服務(wù)并配置開啟binlog,以便通過Flink CDC Connector實現(xiàn)對MySQL數(shù)據(jù)的實時同步,有需要的朋友可以借鑒參考下,希望能夠有所幫助

在驗證flink-cdc-mysql時，需要部署一個開啟了binlog的mysql服務(wù)。cdc文檔中有一個demo可以嘗試部署，但是我在驗證的時候發(fā)現(xiàn)可能存在一些問題，所以就嘗試自己部署一個mysql服務(wù)。與cdc demo中類似，使用docker部署是最快的方案。

網(wǎng)上整理了一些資料，發(fā)現(xiàn)思路大概是：

部署mysql docker服務(wù)
編輯mysql 配置文件
啟動mysql 服務(wù)。

首先根據(jù)需要，編寫docker-compose文件。

創(chuàng)建一個mysql目錄，然后執(zhí)行 vim docker-compose.yml 命令。

將下面內(nèi)容粘貼進(jìn)去，適當(dāng)進(jìn)行修改。

version: "3.7"
services:
  mysql:
    image: mysql:5.7.28
    container_name: mysql-binlog2
    command: --default-authentication-plugin=mysql_native_password
    restart: always
    environment:
      # root用戶密碼
      MYSQL_ROOT_PASSWORD: 123456
      TZ: Asia/Shanghai
    ports:
      - 3306:3306
    volumes:
      - /home/leiline/cdc/data/mysql/master/data:/var/lib/mysql
      - /home/leiline/cdc/data/mysql/master/log:/var/log/mysql
      - /home/leiline/cdc/data/mysql/master/conf:/etc/mysql

保存文件后，退出。

這里需要在服務(wù)器中創(chuàng)建目錄，分別用來保存mysql的數(shù)據(jù)，日志和配置信息。

mkdir -p /home/leiline/cdc/data/mysql/master/data
mkdir -p /home/leiline/cdc/data/mysql/master/log
mkdir -p /home/leiline/cdc/data/mysql/master/conf

其中，我們要把配置信息放在conf目錄下。mysql的配置信息保存在一個名稱為my.cnf的文件中，將下列信息粘貼到my.cnf文件中。

## 局域網(wǎng)唯一
server_id=1
## 指定不需要同步的數(shù)據(jù)庫名稱
binlog-ignore-db=master
## 開啟二進(jìn)制日志功能
log-bin=/var/lib/mysql/mysql-bin
## 設(shè)置二進(jìn)制日志使用內(nèi)存大?。ㄊ聞?wù)）
binlog_cache_size=1M
## 設(shè)置使用的二進(jìn)制日志格式（mixed,statement,row）
binlog_format=ROW
## 二進(jìn)制日志過期清理時間。默認(rèn)值為0，表示不自動清理。
expire_logs_days=7

我們在這里設(shè)置的Binlog二進(jìn)制格式為ROW。

最后，啟動docker-compose。執(zhí)行命令 sudo docker-compose up -d docker-compose 會自動加載docker-compose.yml 文件中的內(nèi)容，拉取 mysql:5.7.28 鏡像，最后將創(chuàng)建mysql-binlog2 容器。

執(zhí)行 docker ps 可以查看當(dāng)前運行中的容器有哪些，以及一些概要信息。

sudo docker ps 
CONTAINER ID   IMAGE                                               COMMAND                  CREATED       STATUS             PORTS                                                  NAMES
c6e03b8a0656   mysql:5.7.28                                        "docker-entrypoint.s…"   2 hours ago   Up About an hour   0.0.0.0:3306->3306/tcp, :::3306->3306/tcp, 33060/tcp   mysql-binlog2

進(jìn)入mysql中進(jìn)行操作

到目前為止，mysql服務(wù)已經(jīng)啟動，我們可以進(jìn)入mysql中查看相關(guān)配置信息，并進(jìn)行數(shù)據(jù)操作。執(zhí)行命令 sudo docker-compose exec mysql mysql -uroot -p123456 就可以進(jìn)入到mysql服務(wù)中。

執(zhí)行命令 show global variables like "%binlog%"; 和 show global variables like "%log_bin%"; 可以查詢相關(guān)配置信息。

由截圖可以看到，binlog_format的值為ROW，log_bin的值為ON。這樣表示binlog已經(jīng)開啟，且二進(jìn)制日志類型為ROW。

接下來我們就可以進(jìn)行數(shù)據(jù)操作了。

插入mysql數(shù)據(jù)

我們還是借助flink cdc demo中的例子，將建表語句和insert語句在mysql中執(zhí)行。

CREATE DATABASE mydb;
USE mydb;
CREATE TABLE products (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  name VARCHAR(255) NOT NULL,
  description VARCHAR(512)
);
ALTER TABLE products AUTO_INCREMENT = 101;

INSERT INTO products
VALUES (default,"scooter","Small 2-wheel scooter"),
       (default,"car battery","12V car battery"),
       (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
       (default,"hammer","12oz carpenter's hammer"),
       (default,"hammer","14oz carpenter's hammer"),
       (default,"hammer","16oz carpenter's hammer"),
       (default,"rocks","box of assorted rocks"),
       (default,"jacket","water resistent black wind breaker"),
       (default,"spare tire","24 inch spare tire");

CREATE TABLE orders (
  order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  order_date DATETIME NOT NULL,
  customer_name VARCHAR(255) NOT NULL,
  price DECIMAL(10, 5) NOT NULL,
  product_id INTEGER NOT NULL,
  order_status BOOLEAN NOT NULL -- Whether order has been placed
) AUTO_INCREMENT = 10001;

INSERT INTO orders
VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
       (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
       (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);

flink mysql cdc 抽數(shù)

最后，我們要借助flink mysql cdc 進(jìn)行binlog數(shù)據(jù)的抽取。

首先準(zhǔn)備好sql語句：

SET execution.checkpointing.interval = 3s;

CREATE TABLE products (
    id INT,
    name STRING,
    description STRING,
    PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'database-name' = 'mydb',
    'table-name' = 'products'
  );

CREATE TABLE printer (
    id INT,
    name STRING,
    description STRING
  ) WITH (
    'connector' = 'print'
  );

INSERT INTO printer SELECT * FROM products;

sql語句的內(nèi)容很簡單，就是將products表的日志數(shù)據(jù)抽取，然后打印出來。

因為我們只是一個demo測試，所以我們使用flink local模式執(zhí)行SQL。與flink cdc demo一樣，首先執(zhí)行./bin/start-cluster.sh 然后進(jìn)入到 sql-client中，./sql-client.sh embedded 這個時候我們進(jìn)入到了flink sql客戶端中，將上述SQL復(fù)制進(jìn)sql-client中，sql-client會自動將SQL進(jìn)行解析，執(zhí)行，提交到flink集群中運行。

如果一切正常的話，我們可以打開地址 http://localhost:8081/ 可以看到flink dashboard。在taskmanager 的out中，可以看到輸出內(nèi)容。

通過對Mysql數(shù)據(jù)的修改，我們測試了增刪改的操作，CDC都可以正常的進(jìn)行數(shù)據(jù)采集。而將表進(jìn)行truncate后，CDC是不會有任何結(jié)果的。