Flink环境的部署

系列教程采用本地模式安装，本系列开发环境为Windows + WSL。

1. 检查Java环境

   java -version
   

Flink需要依赖java环境，部署Flink前需要检查本地是否已经具备Java环境（没有Java环境的请自行查阅资料完成）。

2. 下载Flink并解压

Flink下载地址：https://flink.apache.org/zh/downloads/

选择最新版本的Flink，点击进入下载界面。

使用命令wget https://dlcdn.apache.org/flink/flink-1.19.0/flink-1.19.0-bin-scala_2.12.tgz完成下载。

使用命令tar -xzvf flink-1.19.0-bin-scala_2.12.tgz完成解压。

使用命令cd flink-1.19.0进入到项目根目录下。

3. 启动本地集群

Flink 附带了一个 bash 脚本，可以用于启动本地集群。

在Flink项目根目录下，使用命令./bin/start-cluster.sh启动本地集群。

4. 访问Flink的Web UI

在浏览器中访问地址：http://localhost:8081/，出现如上图界面，则代表我们完成了Flink环境的部署。

5. 停止集群

使用命令./bin/stop-cluster.sh可以用于停止集群。

至此，我们就完成了Flink环境的部署。

基于Docker Compose完成Mysql和Doris环境的部署

Docker环境

检查本地开发环境是否已经安装了docker

docker -v

检查是否安装docker-compose

docker-compose -v

没有docker和docker-compose环境的请自行安装。

由于Doris的运行需要内存映射支持，则需要先执行如下命令：

sysctl -w vm.max_map_count=2000000

MacOS 由于内部实现容器的方式不同，在部署时宿主机直接修改max_map_count值可能无法成功，需要先创建以下容器：

docker run -it --privileged --pid=host --name=change_count debian nsenter -t 1 -m -u -n -i sh

容器创建成功执行以下命令：

sysctl -w vm.max_map_count=2000000

然后exit退出。

编写docker-compose.yml文件

使用docker-compose部署需要先创建一个docker-compose.yml文件，内容如下：

version: '2.1'
services:
  doris:
    image: yagagagaga/doris-standalone
    ports:
      - "8030:8030"
      - "8040:8040"
      - "9030:9030"
  mysql:
    image: debezium/example-mysql:1.1
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=123456
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw

该 Docker Compose 中包含的容器有：

MySQL: 包含商品信息的数据库 app_db

Doris: 存储从 MySQL 中根据规则映射过来的结果表

启动容器

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

docker-compose up -d

该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。

验证容器是否正常启动：

docker ps

访问Doris Web UIhttp://localhost:8030/，查看Doris是否正常运行。

默认的用户名为 root，默认密码为空。

至此，我们就部署完成了Mysql和Doris环境。

Flink CDC环境的部署

Flink CDC下载并解压

访问Flink CDC下载页面https://www.apache.org/dyn/closer.lua/flink/flink-cdc-3.1.0/flink-cdc-3.1.0-bin.tar.gz

复制下载地址并下载

wget https://dlcdn.apache.org/flink/flink-cdc-3.1.0/flink-cdc-3.1.0-bin.tar.gz

解压文件

tar -zxvf flink-cdc-3.1.0-bin.tar.gz

查看文件目录结构

cd flink-cdc-3.1.0/
ls

bin目录

ls bin/

flink-cdc.sh用于提交任务到Flink环境中。

需要注意的是，在使用flink-cdc.sh时需要先指定flink的环境环境变量

解决该问题有两个方法：

1. 使用export设置环境变量

export FLINK_HOME=/home/liuyq/flink-1.19.0

再次运行

./bin/flink-cdc.sh

2. 在提交Pipeline任务时的命令参数里指定

./bin/flink-cdc.sh --flink-home=/home/liuyq/flink-1.19.0

此时运行会报错，需要同时指定Pipeline文件的路径

正确的写法为

./bin/flink-cdc.sh --flink-home=/home/liuyq/flink-1.19.0 mysql-to-doris.yaml

至此，Flink CDC的部署就完成了。最后我们也提到了Pipeline文件，后面我们将继续学习Pipeline文件的编写。

Pipeline的YAML文件编写

编写一个YAML文件

以下为Mysql整库同步到Doris的YAML文件，主要包含三个部分：

• source

用于指定数据从哪里来

• sink

用于指定数据到哪里去

• pipeline

用于任务的名称和并行度，在Flink任务管理页面可以看到对应的任务名称

source:
  type: mysql # 定义数据源的数据库类型
  hostname: localhost # 数据库连接地址
  port: 3306 # 数据库端口号
  username: root # 数据库账户名
  password: 123456 # 数据库账户密码
  tables: app_db.\.* # 指定需要同步的数据库以及表，此处采用正则表达式的方式实现整库同步配置,多个规则可以用逗号分隔，如：adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.*
  server-id: 5400-5404 # 服务id，详细用法可以参考官方文档

sink:
  type: doris # 定义目标数据源的类型
  fenodes: 127.0.0.1:8030 # Doris集群FE的Http地址
  username: root # Doris集群的用户名
  password: "" # Doris集群的密码
  table.create.properties.light_schema_change: true # Doris表属性，是否使用light_schema_change优化
  table.create.properties.replication_num: 1 # Doris表属性，副本数

pipeline:
  name: Sync MySQL Database to Doris # 数据管道的名称
  parallelism: 1 # 数据管道的并发数

有了Pipeline文件，我们就可以提交任务了。

Flink CDC提交任务并验证

数据准备

在提交任务之前，我们还需要准备好测试库和测试数据。

在第二篇中我们已经介绍了如何部署Mysql和Doris的环境，本文不再赘述。此处确保容器已经处于可用状态。

在Mysql中准备数据

1. 进入Mysql容器

使用命令：

docker-compose exec mysql mysql -uroot -p123456

进入到容器内部

另外也可以用数据库管理工具连接数据库

2. 创建数据库app_db和表orders, products, shipments，并插入数据

依次执行下面的SQL语句，完成数据库的创建和测试数据的插入

-- 创建数据库
CREATE DATABASE app_db;

USE app_db;

-- 创建 orders 表
CREATE TABLE `orders` (
`id` INT NOT NULL,
`price` DECIMAL(10,2) NOT NULL,
PRIMARY KEY (`id`)
);

-- 插入数据
INSERT INTO `orders` (`id`, `price`) VALUES (1, 4.00);
INSERT INTO `orders` (`id`, `price`) VALUES (2, 100.00);

-- 创建 shipments 表
CREATE TABLE `shipments` (
`id` INT NOT NULL,
`city` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);

-- 插入数据
INSERT INTO `shipments` (`id`, `city`) VALUES (1, 'beijing');
INSERT INTO `shipments` (`id`, `city`) VALUES (2, 'xian');

-- 创建 products 表
CREATE TABLE `products` (
`id` INT NOT NULL,
`product` VARCHAR(255) NOT NULL,
PRIMARY KEY (`id`)
);

-- 插入数据
INSERT INTO `products` (`id`, `product`) VALUES (1, 'Beer');
INSERT INTO `products` (`id`, `product`) VALUES (2, 'Cap');
INSERT INTO `products` (`id`, `product`) VALUES (3, 'Peanut');

至此，Mysql的准备工作就完成了。

在Doris中创建数据库

Doris 暂时不支持自动创建数据库，需要先创建写入表对应的数据库。

1. 进入 Doris Web UI。

访问http://localhost:8030/，默认的用户名为 root，默认密码为空。

2. 通过 Web UI 创建 app_db 数据库

create database app_db;

至此，Doris的准备工作就完成了。

通过Flink CDC CLI提交任务

同步任务依赖连接器，提交任务前确保flink-cdc-3.1.0/lib目录下有如下连接器：

• flink-cdc-pipeline-connector-mysql-3.1.0.jar
• flink-cdc-pipeline-connector-doris-3.1.0.jar

如果没有则下载，下载地址为：

# 下载mysql连接器
wget https://repo1.maven.org/maven2/org/apache/flink/fli
nk-cdc-pipeline-connector-mysql/3.1.0/flink-cdc-pipeline-connector-mysql-3.1.0.jar

# 下载doris连接器
wget https://repo1.maven.org/maven2/org/apache/flink/flink-cdc-pipeline-connector-doris/3.1.0/flink-cdc-pipeline-connector-doris-3.1.0.jar

另外，还需要确保Flink的flink-1.19.0/lib下包含MySQL驱动，

如果没有则需要下载

# 下载MySQL驱动
wget https://repo1.maven.org/maven2/mysql/mysql-connector-java/8.0.27/mysql-connector-java-8.0.27.jar

提交任务

./bin/flink-cdc.sh mysql-to-doris.yaml

直接执行提交任务的命令，你是否又一次碰到了这个错误。别着急，你只是忘记了告诉Flink CDC环境你的Flink环境在哪里。

明确了问题，调整以下命令

./bin/flink-cdc.sh mysql-to-doris.yaml --flink-home /home/liuyq/flink-1.19.0

如果提交任务出现了如下图错误

请想一想，确保你的Flink服务已经启动了

验证

访问Flink Web UIhttp://localhost:8081/，如果Running Jobs中出现下图中任务，则说明我们的实时数据同步任务已经成功上线了。

接下来，你就可以畅玩了，试一试在Mysql的表中插入一些数据，然后再以非常快的速度在Doris中查一下数据吧，看看是你的动作快，还是数据同步的快。