欢迎您访问365答案网,请分享给你的朋友!
生活常识 学习资料

PyFlink流处理基础实例MySQLCDC方式实时备份

时间:2023-04-18
01 JDBC SQL 连接器

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 从入门到精通

Copyright © 2016-2020 www.365daan.com All Rights Reserved. 365答案网 版权所有 备案号:

部分内容来自互联网,版权归原作者所有,如有冒犯请联系我们,我们将在三个工作时内妥善处理。