基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL

這篇具有很好參考價值的文章主要介紹了基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL。希望對大家有所幫助。如果存在錯誤或未考慮完全的地方，請大家不吝賜教，您也可以點擊"舉報違法"按鈕提交疑問。

官方網(wǎng)址：https://ververica.github.io/flink-cdc-connectors/release-2.3/content/%E5%BF%AB%E9%80%9F%E4%B8%8A%E6%89%8B/mysql-postgres-tutorial-zh.html官方教程有些坑，經(jīng)過自己實測，記錄個筆記。

服務(wù)器環(huán)境：

VM虛擬機：CentOS7.9

docker版本：Docker version 24.0.5, build ced0996

docker compose 版本：2.19

jdk 1.8

虛擬機IP:192.168.122.131?

內(nèi)存：16G(一定要大于等于16G)

CPU：4g

磁盤：>= 60G

一、docker? compose安裝

DOCKER_CONFIG=${DOCKER_CONFIG:-/usr/local/lib/docker/cli-plugins}
mkdir -p $DOCKER_CONFIG/cli-plugins
curl -SL https://github.com/docker/compose/releases/download/v2.19.1/docker-compose-linux-x86_64 -o $DOCKER_CONFIG/cli-plugins/docker-compose

對文件應(yīng)用可執(zhí)行權(quán)限：

chmod +x $DOCKER_CONFIG/cli-plugins/docker-compose

測試安裝是否成功

docker compose version #之前的v1版本命令是docker-compose --version

參考：https://blog.csdn.net/qq_40099908/article/details/131611496

二、實戰(zhàn)

這篇教程將展示如何基于 Flink CDC 快速構(gòu)建 MySQL 和 Postgres 的流式 ETL。本教程的演示都將在 Flink SQL CLI 中進行，只涉及 SQL，無需一行 Java/Scala 代碼，也無需安裝 IDE。

假設(shè)我們正在經(jīng)營電子商務(wù)業(yè)務(wù)，商品和訂單的數(shù)據(jù)存儲在 MySQL 中，訂單對應(yīng)的物流信息存儲在 Postgres 中。對于訂單表，為了方便進行分析，我們希望讓它關(guān)聯(lián)上其對應(yīng)的商品和物流信息，構(gòu)成一張寬表，并且實時把它寫到 ElasticSearch 中。

接下來的內(nèi)容將介紹如何使用 Flink Mysql/Postgres CDC 來實現(xiàn)這個需求，系統(tǒng)的整體架構(gòu)如下圖所示：

基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL,flink,etl,大數(shù)據(jù)

1、準備教程所需要的組件

接下來的教程將以?docker-compose?的方式準備所需要的組件。

使用下面的內(nèi)容創(chuàng)建一個?docker-compose.yml?文件：

version: '2.1'
services:
  postgres:
    image: debezium/example-postgres:1.1
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_DB=postgres
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
  mysql:
    image: debezium/example-mysql:1.1
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=123456
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw
  elasticsearch:
    image: elastic/elasticsearch:7.6.0
    environment:
      - cluster.name=docker-cluster
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
      - discovery.type=single-node
    ports:
      - "9200:9200"
      - "9300:9300"
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
  kibana:
    image: elastic/kibana:7.6.0
    ports:
      - "5601:5601"

該 Docker Compose 中包含的容器有：

MySQL: 商品表?products?和訂單表?orders?將存儲在該數(shù)據(jù)庫中，這兩張表將和 Postgres 數(shù)據(jù)庫中的物流表?shipments進行關(guān)聯(lián)，得到一張包含更多信息的訂單表?enriched_orders
Postgres: 物流表?shipments?將存儲在該數(shù)據(jù)庫中
Elasticsearch: 最終的訂單表?enriched_orders?將寫到 Elasticsearch
Kibana: 用來可視化 ElasticSearch 的數(shù)據(jù)

在?docker-compose.yml?所在目錄下執(zhí)行下面的命令來啟動本教程需要的組件：

docker compose up -d

該命令將以 detached 模式自動啟動 Docker Compose 配置中定義的所有容器。你可以通過 docker ps 來觀察上述的容器是否正常啟動了，也可以通過訪問?http://192.168.122.131:5601來查看 Kibana 是否運行正常。

2、下載 Flink 和所需要的依賴包

下載?Flink 1.16.0?并將其解壓至目錄?flink-1.16.0? ，

下載下面列出的依賴包，并將它們放到目錄?flink-1.16.0/lib/?下：

下載鏈接只對已發(fā)布的版本有效, SNAPSHOT 版本需要本地編譯
- flink-sql-connector-elasticsearch7-1.16.0.jar
- flink-sql-connector-mysql-cdc-2.3.0.jar
- flink-sql-connector-postgres-cdc-2.3.0.jar

準備數(shù)據(jù)

在 MySQL 數(shù)據(jù)庫中準備數(shù)據(jù)

進入 MySQL 容器

docker compose exec mysql mysql -uroot -p123456

創(chuàng)建數(shù)據(jù)庫和表?products，orders，并插入數(shù)據(jù)

-- MySQL
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);

注意：mysql會遇到時區(qū)不對的情況。

在mysql容器調(diào)整時區(qū)：

set time_zone='+8:00';
SET GLOBAL time_zone = '+8:00';
flush privileges;
SELECT @@global.time_zone;
show variables like '%time_zone%';

在 Postgres 數(shù)據(jù)庫中準備數(shù)據(jù)

進入 Postgres 容器

docker compose exec postgres psql -h localhost -U postgres

創(chuàng)建表?shipments，并插入數(shù)據(jù)

-- PG
CREATE TABLE shipments (
  shipment_id SERIAL NOT NULL PRIMARY KEY,
  order_id SERIAL NOT NULL,
  origin VARCHAR(255) NOT NULL,
  destination VARCHAR(255) NOT NULL,
  is_arrived BOOLEAN NOT NULL
);
ALTER SEQUENCE public.shipments_shipment_id_seq RESTART WITH 1001;
ALTER TABLE public.shipments REPLICA IDENTITY FULL;
INSERT INTO shipments
VALUES (default,10001,'Beijing','Shanghai',false),
       (default,10002,'Hangzhou','Shanghai',false),
       (default,10003,'Shanghai','Hangzhou',false);

啟動 Flink 集群和 Flink SQL CLI

使用下面的命令跳轉(zhuǎn)至 Flink 目錄下

cd flink-1.16.0

使用下面的命令啟動 Flink 集群

./bin/start-cluster.sh

啟動成功的話，可以在?http://192.168.122.131:8081/?訪問到 Flink Web UI，如下所示：

基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL,flink,etl,大數(shù)據(jù)

注：若在VM之外的本地的電腦里無法訪問，則需要調(diào)整 /flink-1.16.0/conf/flink-conf.yaml文件，

將rest.bind-address值改為：0.0.0.0

開放單個端口（開放后需要要重啟防火墻才生效）；

firewall-cmd --zone=public --add-port=8081/tcp --permanent

重啟防火墻；?systemctl restart firewalld

? 另：還有個參數(shù)taskmanager.numberOfTaskSlots: 50，一般設(shè)置大一些的值，比如50。

使用下面的命令啟動 Flink SQL CLI

./bin/sql-client.sh

啟動成功后，可以看到如下的頁面：

基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL,flink,etl,大數(shù)據(jù)

在 Flink SQL CLI 中使用 Flink DDL 創(chuàng)建表

首先，開啟 checkpoint，每隔3秒做一次 checkpoint

-- Flink SQL                   
Flink SQL> SET execution.checkpointing.interval = 3s;

然后, 對于數(shù)據(jù)庫中的表?products,?orders,?shipments，使用 Flink SQL CLI 創(chuàng)建對應(yīng)的表，用于同步這些底層數(shù)據(jù)庫表的數(shù)據(jù)

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

Flink SQL> CREATE TABLE orders (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   PRIMARY KEY (order_id) NOT ENFORCED
 ) WITH (
   'connector' = 'mysql-cdc',
   'hostname' = 'localhost',
   'port' = '3306',
   'username' = 'root',
   'password' = '123456',
   'database-name' = 'mydb',
   'table-name' = 'orders'
 );

Flink SQL> CREATE TABLE shipments (
   shipment_id INT,
   order_id INT,
   origin STRING,
   destination STRING,
   is_arrived BOOLEAN,
   PRIMARY KEY (shipment_id) NOT ENFORCED
 ) WITH (
   'connector' = 'postgres-cdc',
   'hostname' = 'localhost',
   'port' = '5432',
   'username' = 'postgres',
   'password' = 'postgres',
   'database-name' = 'postgres',
   'schema-name' = 'public',
   'table-name' = 'shipments'
 );

最后，創(chuàng)建?enriched_orders?表，用來將關(guān)聯(lián)后的訂單數(shù)據(jù)寫入 Elasticsearch 中

-- Flink SQL
Flink SQL> CREATE TABLE enriched_orders (
   order_id INT,
   order_date TIMESTAMP(0),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   order_status BOOLEAN,
   product_name STRING,
   product_description STRING,
   shipment_id INT,
   origin STRING,
   destination STRING,
   is_arrived BOOLEAN,
   PRIMARY KEY (order_id) NOT ENFORCED
 ) WITH (
     'connector' = 'elasticsearch-7',
     'hosts' = 'http://localhost:9200',
     'index' = 'enriched_orders'
 );

關(guān)聯(lián)訂單數(shù)據(jù)并且將其寫入 Elasticsearch 中

使用 Flink SQL 將訂單表?order?與商品表?products，物流信息表?shipments?關(guān)聯(lián)，并將關(guān)聯(lián)后的訂單信息寫入 Elasticsearch 中

-- Flink SQL
Flink SQL> INSERT INTO enriched_orders
 SELECT o.*, p.name, p.description, s.shipment_id, s.origin, s.destination, s.is_arrived
 FROM orders AS o
 LEFT JOIN products AS p ON o.product_id = p.id
 LEFT JOIN shipments AS s ON o.order_id = s.order_id;

現(xiàn)在，就可以在 Kibana 中看到包含商品和物流信息的訂單數(shù)據(jù)。

首先訪問 http://192.168.122.131:5601/app/kibana#/management/kibana/index_pattern 創(chuàng)建 index pattern enriched_orders.

基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL,flink,etl,大數(shù)據(jù)

然后就可以在 http://192.168.122.131:5601/app/kibana#/discover 看到寫入的數(shù)據(jù)了.

基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL,flink,etl,大數(shù)據(jù)

接下來，修改 MySQL 和 Postgres 數(shù)據(jù)庫中表的數(shù)據(jù)，Kibana中顯示的訂單數(shù)據(jù)也將實時更新：

在 MySQL 的?orders?表中插入一條數(shù)據(jù)

--MySQL
INSERT INTO orders
VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);

在 Postgres 的?shipment?表中插入一條數(shù)據(jù)

--PG
INSERT INTO shipments
VALUES (default,10004,'Shanghai','Beijing',false);

在 MySQL 的?orders?表中更新訂單的狀態(tài)

--MySQL
UPDATE orders SET order_status = true WHERE order_id = 10004;

在 Postgres 的?shipment?表中更新物流的狀態(tài)

--PG
UPDATE shipments SET is_arrived = true WHERE shipment_id = 1004;

在 MYSQL 的?orders?表中刪除一條數(shù)據(jù)

--MySQL
DELETE FROM orders WHERE order_id = 10004;

每執(zhí)行一步就刷新一次 Kibana，可以看到 Kibana 中顯示的訂單數(shù)據(jù)將實時更新，如下所示：

基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL,flink,etl,大數(shù)據(jù)

環(huán)境清理

本教程結(jié)束后，在?docker-compose.yml?文件所在的目錄下執(zhí)行如下命令停止所有容器：

docker compose down

在 Flink 所在目錄?flink-1.16.0?下執(zhí)行如下命令停止 Flink 集群：

./bin/stop-cluster.sh

異常排查

若數(shù)據(jù)異常，在flink的網(wǎng)頁里看查看錯誤信息。

http://192.168.122.131:8081/#/job/running

基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL,flink,etl,大數(shù)據(jù) 文章來源地址http://www.zghlxwxcb.cn/news/detail-702940.html

到了這里，關(guān)于基于 Flink CDC 構(gòu)建 MySQL 和 Postgres 的 Streaming ETL的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！