JDBC 连接器允许使用 JDBC 驱动向任意类型的关系型数据库读取或者写入数据。
如果在 DDL 中定义了主键，JDBC sink 将以 upsert 模式与外部系统交换 UPDATe/DELETE 消息；否则，它将以 append 模式与外部系统交换消息且不支持消费 UPDATE/DELETE 消息。
1.1 下载依赖包针对关系型数据库实现 Flink 通过建立 JDBC 连接器来执行 SQL 查询，要下载 flink-connector-jdbc 依赖包，其下载地址为 flink-connector-jdbc_2.11-1.14.0.jar
在连接到具体数据库时，也需要对应的驱动依赖，MySql的驱动 jar 包下载地址为 mysql-connector-java
1.2 创建 JDBC 表作业中加入上述依赖包之后，使用 Table API 的 JDBC table 可以按如下定义：
-- 在 Flink SQL 中注册一张 MySQL 表 'users'CREATE TABLE MyUserTable ( id BIGINT, name STRING, age INT, status BOOLEAN, PRIMARY KEY (id) NOT ENFORCED) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://localhost:3306/mydatabase', 'table-name' = 'users');-- 从另一张表 "T" 将数据写入到 JDBC 表中INSERT INTO MyUserTableSELECT id, name, age, status FROM T;-- 查看 JDBC 表中的数据SELECt id, name, age, status FROM MyUserTable;-- JDBC 表在时态表关联中作为维表SELECt * FROM myTopicLEFT JOIN MyUserTable FOR SYSTEM_TIME AS OF myTopic.proctimeON myTopic.key = MyUserTable.id;
JDBC SQL 连接器的一些参数与含义如下：
connector：指定使用什么类型的连接器，这里应该是’jdbc’。url：JDBC 数据库 url。table-name：连接到 JDBC 表的名称。driver：用于连接到此 URL 的 JDBC 驱动类名，如果不设置，将自动从 URL 中推导。username：JDBC 用户名。如果指定了 ‘username’ 和 ‘password’ 中的任一参数，则两者必须都被指定。password：JDBC 密码。 02 JAR 依赖管理
如果作业中需要用到第三方的 jar 包，Python Table API 提供如下方式指明 jar 包
# 方式1：本地上传table_env.get_config().get_configuration().set_string("pipeline.jars", "file:///my/jar/path/connector.jar;file:///my/jar/path/udf.jar")# 方式2：运行时加载table_env.get_config().get_configuration().set_string("pipeline.classpaths", "file:///my/jar/path/connector.jar;file:///my/jar/path/udf.jar")
本地上传方式，通过将本地文件 URL 指向的 jar 包上传到 Flink 群集。在配置中通过 pipeline.jars 指定 jar包的 URL 列表，不同 jar 包之间用分号 ; 隔开。
运行时加载方式，指定确保 URL 在客户端和群集上都可访问的 jar 包路径规则，作业运行时根据路径加载指定 jar 包。
Python DataStream API 中提供了相似的方法指明 jar 包：本地上传 add_jars() 和运行时加载 add_classpaths()
stream_execution_environment.add_jars("file:///my/jar/path/connector1.jar", "file:///my/jar/path/connector2.jar")stream_execution_environment.add_classpaths("file:///my/jar/path/connector1.jar", "file:///my/jar/path/connector2.jar")
03 实时 MySql CDC数据备份实例本实例展示了如何用 Flink 进行 MySQL 数据库的实时同步。实例中通过监听 当前 MySQL 数据库的 binlog 数据变更，然后根据 binlog 日志将数据变更实时同步到另一个 MySQL 数据库中。
3.1 Flink CDC 连接器 CDC 的简介与用途CDC (change data capture) ，即变化数据捕捉，它是数据库进行备份的一种方式，常用于大量数据的备份工作。
CDC 分为入侵式的和非入侵式两种：
入侵式：如基于触发器备份、基于时间戳备份、基于快照备份。非入侵式：如基于日志的备份。
本实例采用非入侵式，基于 MySql 的 binlog 日志进行备份。
所以要实现基于日志的 CDC 数据备份， MySQL 就是要开启 binlog 选项。 MySQL 8.0.X 是默认开启 binlog 的，你可以通过执行下面的命令，查看 log_bin 变量的值是否为 ON来检查 MySQL 是否开启 binlog。
show variables like '%log_bin%';
使用 Flink CDC 连接器首先下载 MySql CDC 连接器 jar 包，下载地址为 flink-sql-connector-mysql-cdc
使用 Table API 可以创建如下 mysql cdc table， 通过 MySQL-CDC 连接器从 MySQL 的 binlog 里提取更改
-- creates a mysql cdc table sourceCREATE TABLE mysql_binlog ( id INT NOT NULL, name STRING, description STRING, weight DECIMAL(10,3)) WITH ( 'connector' = 'mysql-cdc', 'hostname' = 'localhost', 'port' = '3306', 'username' = 'flinkuser', 'password' = 'flinkpw', 'database-name' = 'inventory', 'table-name' = 'products');-- read snapshot and binlog data from mysql, and do some transformation, and show on the clientSELECt id, UPPER(name), description, weight FROM mysql_binlog;
3.2 实时 MySql CDC数据备份实例 （1）初始化流处理环境并指定依赖 jar 包为了在 Flink 中实现 MySQL 的 CDC，我们需要准备下面 3 个 jar 包：
flink-connector-jdbc_2.11-1.14.0.jar：通过 JDBC 连接器来从 MySQL 里读取或写入数据；用于创建 sink 表，定义连接器为 jdbcmysql-connector-java-8.0.16.jar：JDBC 连接器的驱动（ 帮助 java 连接 MySQL ）flink-sql-connector-mysql-cdc-2.0.2.jar：通过 MySQL-CDC 连接器从 MySQL 的 binlog 里提取更改；用于创建 source 表，定义连接器为 mysql-cdc JAR 名称 下载地址 flink-connector-jdbc_2.11-1.14.0.jar下载地址mysql-connector-java-8.0.16.jar下载地址flink-sql-connector-mysql-cdc-2.0.2.jar下载地址
初始化流处理环境并指定依赖 jar 包
# 初始化流处理环境 env_settings = EnvironmentSettings.in_streaming_mode()t_env = TableEnvironment.create(env_settings) # 编辑 jar 包 URLs jars = []for file in os.listdir(os.path.abspath(os.path.dirname(__file__))): if file.endswith('.jar'): jars.append(os.path.abspath(file))str_jars = ';'.join(['file://' + jar for jar in jars])# 指定 jar 依赖 t_env.get_config().get_configuration().set_string("pipeline.jars", str_jars)
（2）创建 MySql CDC 源表# 创建 MySql CDC 源表 mysql_cdc_ddl = """ CREATE TABLE source ( id INT, name STRING, PRIMARY KEY (id) NOT ENFORCED ) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '127.0.0.1', 'port' = '3307', 'database-name' = 'flink', 'table-name' = 'user', 'username' = 'root', 'password' = 'root' ) """ t_env.execute_sql(mysql_cdc_ddl)
（3）创建数据同步的 JDBC 输出表将数据写入外部数据库时，Flink 使用 DDL 中定义的主键。
如果定义了主键，则连接器将在 upsert 模式下运行，否则，连接器将在追加模式下运行。
在 upsert 模式下，Flink 将根据主键插入新行或更新现有行，Flink 可以通过这种方式确保幂等性。在追加模式下，Flink 将所有记录解释为 INSERT 消息，如果在基础数据库中发生主键或唯一约束冲突，则 INSERT 操作可能会失败。
# 创建数据同步的 JDBC 输出表 jdbc_sink_ddl = """ CREATE TABLE sink ( id INT, name STRING, PRIMARY KEY (id) NOT ENFORCED -- 需要定义主键，让连接器在 upsert 模式下运行 ) WITH ( 'connector' = 'jdbc', 'url' = 'jdbc:mysql://127.0.0.1:3308/flink', 'driver' = 'com.mysql.cj.jdbc.Driver', 'table-name' = 'user', 'username' = 'root', 'password' = 'root' ) """ t_env.execute_sql(jdbc_sink_ddl)
（4）运行实例指向 MySql CDC 实时数据备份作业之前，我们需要准备实验环境：两个 mysql 数据库。
我们用 docker 容器快速构建实验环境，除了两个 mysql 数据库，还创建一个 adminer 容器：
mysql1 容器作为待同步的数据源mysql2 容器作为备份的数仓adminer 容器允许我们在本地没有安装 mysql 客户端的情况下，使用网页来查看和操作 mysql 容器
2 个 mysql 容器 + 1 个 adminer 容器，编写容器编排 docker-compose.yml如下:
version: "3.5"services: mysql1: image: mysql:8.0.22 # 5.7 版本本地连接不上 command: [ '--default-authentication-plugin=mysql_native_password', '--character-set-server=utf8mb4', '--collation-server=utf8mb4_unicode_ci', '--log-bin=mysql-bin', ] ports: - 3307:3306 environment: MYSQL_ROOT_PASSWORD: root volumes: - ./examples/mysql:/docker-entrypoint-initdb.d container_name: mysql1 mysql2: image: mysql:8.0.22 # 5.7 版本本地连接不上 command: [ '--default-authentication-plugin=mysql_native_password', '--character-set-server=utf8mb4', '--collation-server=utf8mb4_unicode_ci' ] ports: - 3308:3306 # 第二个数据库的端口是 3307 environment: MYSQL_ROOT_PASSWORD: root volumes: - ./examples/mysql:/docker-entrypoint-initdb.d container_name: mysql2 adminer: image: adminer ports: - 8089:8080 container_name: adminer
使用 docker-compose up 命令启动容器环境，如果出现错误请检查容器端口映射是否冲突。
实验环境启动之后执行 MySql CDC 实时数据备份作业代码：
python flink_mysql_cdc.py
如果本地有 MySQL 客户端，可以同时连接 3307 和 3308 这两个端口（账密都是 root）。
在 3307 对应的实例上的 flink 数据库的 user 这张表里做相应的增删改，然后看看 3308 对应的同名的表里，是否已经实时同步了数据。
（5）实例完整代码 参考资料Flink 官方文档 JDBC SQL 连接器
Flink 官方文档 依赖管理
flink-cdc-connectors 官方文档
PyFlink 从入门到精通