Flink系列之：基于Flink CDC构建MySQL和Postgres的Streaming ETL

Flink系列之：基于Flink CDC构建MySQL和Postgres的Streaming ETL

一、Flink CDC架构

二、安装docker compose

docker系列之：Centos7安装docker compose

三、基于docker-compose准备所需组件

使用docker-compose快速安装mysql、postgres、elasticsearch、kibana

version:'2.1'services:postgres:image: debezium/example-postgres:1.1ports:-"5432:5432"environment:-POSTGRES_DB=postgres
      -POSTGRES_USER=postgres
      -POSTGRES_PASSWORD=postgres
  mysql:image: debezium/example-mysql:1.1ports:-"3306:3306"environment:-MYSQL_ROOT_PASSWORD=123456-MYSQL_USER=mysqluser
      -MYSQL_PASSWORD=mysqlpw
  elasticsearch:image: elastic/elasticsearch:7.6.0environment:- 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:-1hard:-1nofile:soft:65536hard:65536kibana:image: elastic/kibana:7.6.0ports:-"5601:5601"

该Docker Compose中包含的容器有：

• MySQL: 商品表 products 和 订单表 orders 将存储在该数据库中， 这两张表将和 Postgres 数据库中的物流表 shipments进行关联，得到一张包含更多信息的订单表 enriched_orders
• Postgres: 物流表 shipments 将存储在该数据库中
• Elasticsearch: 最终的订单表 enriched_orders 将写到 Elasticsearch
• Kibana: 用来可视化 ElasticSearch 的数据

在 docker-compose.yml 所在目录下执行下面的命令来启动本教程需要的组件：

docker-compose up -d

该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 docker ps 来观察上述的容器是否正常启动了，也可以通过访问 http://localhost:5601/ 来查看 Kibana 是否运行正常。

四、下载Flink和所需要的依赖包

下载 Flink 1.13.2 并将其解压至目录 flink-1.13.2
下载下面列出的依赖包，并将它们放到目录 flink-1.13.2/lib/ 下：

• flink-sql-connector-elasticsearch7_2.11-1.13.2.jar
• flink-sql-connector-mysql-cdc-2.3-SNAPSHOT.jar
• flink-sql-connector-postgres-cdc-2.3-SNAPSHOT.jar