流计算 Oceanus 数据库连接 JDBC

一、介绍

注意：CDW PostgreSQL 只能使用 PostgreSQL connector。

二、版本说明

三、使用范围

JDBC 支持用作数据源表（Source），用于按固定列扫描表和用于 JOIN 的右表（维表）；也支持用作数据目的表（sink），用于 Tuple 数据流表和用于 Upsert 数据流表（需要指定主键）。

若需要将 JDBC 数据库的变动记录作为流式源表消费，请使用内置的 CDC 数据源。若内置的 CDC 数据源无法满足需求，还可使用 Debezium、Canal 等，对 JDBC 数据库的变更进行捕获和订阅，然后 Oceanus 即可对这些变更事件进行进一步的处理。详情可参见 消息队列 Kafka。

四、用作按列扫描的数据源（Source）

CREATE TABLE `jdbc_source_table` ( `id` INT, `name` STRING) WITH ( — 指定数据库连接参数 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — 请替换为你的实际 MySQL 连接参数 'table-name' = 'my-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'scan.partition.column' = 'id', — 分区扫描列名 'scan.partition.num' = '2', — 分区数量 'scan.partition.lower-bound' = '0', — 首个分区的最小值 'scan.partition.upper-bound' = '100', — 最后一个分区的最大值 'scan.fetch-size' = '1' — 每次从数据库读取时，批量获取的行数);

五、用作维表数据源（Source）

CREATE TABLE `jdbc_dim_table` ( `id` INT, `name` STRING) WITH ( — 指定数据库连接参数 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — 请替换为你的实际 MySQL 连接参数 'table-name' = 'my-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'lookup.cache.max-rows' = '100', — 读缓存大小 'lookup.cache.ttl' = '5000' — 读缓存的 TTL);

六、用作数据目的（Tuple Sink）

CREATE TABLE `jdbc_sink_table` ( `id` INT, `name` STRING) WITH ( — 指定数据库连接参数 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — 请替换为你的实际 MySQL 连接参数 'table-name' = 'my-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'sink.buffer-flush.max-rows' = '200', — 批量输出的条数 'sink.buffer-flush.interval' = '2s' — 批量输出的间隔);

七、用作数据目的（Upsert Sink）

CREATE TABLE `jdbc_upsert_sink_table` ( `id` INT PRIMARY KEY NOT ENFORCED, `name` STRING) WITH ( — 指定数据库连接参数 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — 请替换为你的实际 MySQL 连接参数 'table-name' = 'my-upsert-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'sink.buffer-flush.max-rows' = '200', — 批量输出的条数 'sink.buffer-flush.interval' = '2s' — 批量输出的间隔);

注意：Upsert 需要定义 PRIMARY KEY。

八、WITH 参数

九、代码示例

CREATE TABLE `jdbc_source_table` ( `id` INT, `name` STRING) WITH ( — 指定数据库连接参数 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — 请替换为你的实际 MySQL 连接参数 'table-name' = 'my-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'scan.partition.column' = 'id', — 分区扫描列名 'scan.partition.num' = '2', — 分区数量 'scan.partition.lower-bound' = '0', — 首个分区的最小值 'scan.partition.upper-bound' = '100', — 最后一个分区的最大值 'scan.fetch-size' = '1' — 每次从数据库读取时，批量获取的行数);CREATE TABLE `jdbc_upsert_sink_table` ( `id` INT PRIMARY KEY NOT ENFORCED, `name` STRING) WITH ( — 指定数据库连接参数 'connector' = 'jdbc', 'url' = 'jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true&serverTimezone=Asia/Shanghai', — 请替换为你的实际 MySQL 连接参数 'table-name' = 'my-upsert-table', — 需要写入的数据表 'username' = 'admin', — 数据库访问的用户名（需要提供 INSERT 权限） 'password' = 'MyPa$$w0rd', — 数据库访问的密码 'sink.buffer-flush.max-rows' = '200', — 批量输出的条数 'sink.buffer-flush.interval' = '2s' — 批量输出的间隔);insert into jdbc_upsert_sink_table select * from jdbc_source_table;

十、主键说明

·对于 Append（Tuple）数据，作业 DDL 不需要设置主键，JDBC 数据库的表也不需要定义主键，也不建议定义主键（否则可能因为重复数据而造成写入失败）。

·对于 Upsert 数据，JDBC 数据库的表必须定义主键，并需要在 DDL 语句的 CREATE TABLE 中，将相应列名也加上 PRIMARY KEY NOT ENFORCED 约束。

十一、分区扫描

分区扫描（Partitioned Scan）可以加速多个并行度的 Source 算子读取 JDBC 数据表，每个子任务可以读取自己的专属分区。使用该功能时，所有四个 scan.partition 开头的参数都必须指定，否则会报错。

注意：这里的 scan.partition.upper-bound 指定的最大值和 scan.partition.lower-bound 指定的最小值，指的是每个分区的步长，不会影响最终读取的数据条数和精确性。

十二、读取缓存

Flink 提供了读取缓存（Lookup Cache）功能，可以提升维表读取的性能。目前该缓存的实现是同步的，默认未启用（每次请求都会读取数据库，吞吐量很低），必须手动设置 lookup.cache.max-rows 和 lookup.cache.ttl 两个参数来启用该功能。

注意：如果缓存的 TTL 太长，或者缓存的条数太多，可能会造成数据库中数据更新后，Flink 作业仍然读取的是缓存中的旧数据。因此对于数据库变动敏感的作业，请谨慎使用缓存功能。

十三、批量写入优化

通过设置 sink.buffer-flush 开头的两个参数，可以实现批量写入数据库。建议配合相应底层数据库的参数，以达到更好的批量写入效果，否则底层仍然会一条一条写入，效率不高。

·对于 MySQL，建议在 url 连接参数后加入 rewriteBatchedStatements=true 参数。

jdbc:mysql://10.1.28.93:3306/CDB?rewriteBatchedStatements=true

·对于 PostgreSQL，建议在 url 连接参数后加入 reWriteBatchedInserts=true 参数。

jdbc:postgresql://10.1.28.93:3306/PG?reWriteBatchedInserts=true&currentSchema=数据库的Schema

十四、监控指标说明

Oceanus 为 JDBC Connector 也增加了很多实用的统计指标。单击 Flink UI 的运行图中的 JDBC Sink 算子，即可搜索并查看指标：

numberOfInsertRecords：获取输出的 +I 消息数。

numberOfDeleteRecords：获取输出的 -D 消息数。

numberOfUpdateBeforeRecords：获取输出的 -U 消息数。

numberOfUpdateAfterRecords：获取输出的 +U 消息数。