JDBC

功能描述

JDBC连接器是Flink内置的Connector，提供了对MySQL、PostgreSQL等常见数据库的读写支持。表类型支持源表、结果表和维表。

前提条件

• 要与实例建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。
• 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中增强型跨源连接章节。
• 如何设置安全组规则，请参见《虚拟私有云用户指南》中“安全组”章节。

注意事项

• JDBC结果表如果定义了主键，将以upsert模式与外部系统交换UPDATE/DELETE消息；否则，它将以append模式与外部系统交换消息，不支持消费UPDATE/DELETE消息。
• 创建Flink OpenSource SQL作业时，在作业编辑界面的“运行参数”处，“Flink版本”需要选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。
• 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据

语法格式

create table jbdcTable (
  attr_name attr_type 
  (',' attr_name attr_type)* 
  (','PRIMARY KEY (attr_name, ...) NOT ENFORCED)
  (',' watermark for rowtime_column_name as watermark-strategy_expression)
) with (
  'connector' = 'jdbc',
  'url' = '',
  'table-name' = '',
  'username' = '',
  'password' = ''
);

参数说明

分区扫描功能介绍

为了加速Source任务实例中的数据读取，Flink为JDBC表提供了分区扫描功能。以下参数定义了从多个任务并行读取时如何对表进行分区。

• scan.partition.column：用于对输入进行分区的列名，该列的数据类型必须是数字，日期或时间戳。
• scan.partition.num: 分区数。
• scan.partition.lower-bound：第一个分区的最小值。
• scan.partition.upper-bound：最后一个分区的最大值。

• 建表时以上扫描分区参数必须同时存在或者同时不存在。
• scan.partition.lower-bound和scan.partition.upper-bound参数仅用于决定分区步长，而不是用于过滤表中的行，表中的所有行都会被分区并返回。

Lookup Cache功能介绍

JDBC连接器可以用在时态表关联中作为一个可lookup的维表，当前只支持同步的查找模式。

默认情况下，Lookup cache是未启用的，所有请求都会发送到外部数据库。您可以设置Lookup.cache.max-rows和Lookup.cache.ttl参数来启用。Lookup cache的主要目的是用于提高时态表关联JDBC连接器的性能。

当Lookup cache被启用时，每个进程（即TaskManager）将维护一个缓存。Flink将优先查找缓存，只有当缓存未查找到时才向外部数据库发送请求，并使用返回的数据更新缓存。当缓存命中最大缓存行Lookup.cache.max-rows或当行超过最大存活时间Lookup.cache.ttl时，缓存中最先添加的条目将被标记为过期。缓存中的记录可能不是最新的，用户可以将Lookup.cache.ttl设置为一个更小的值以获得更好的刷新数据，但这可能会增加发送到数据库的请求数。所以要做好吞吐量和正确性之间的平衡。

默认情况下，Flink会缓存主键的空查询结果，您可以通过将Lookup.cache.caching-missing-key设置为false来切换行为。

数据类型映射

示例

• 示例1：使用JDBC作为数据源，Print作为结果表，从RDS MySQL数据库中读取数据，并写入到Print结果表中。
  1. 参考增强型跨源连接，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。
  2. 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。
  3. 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考创建RDS数据库。

     在flink数据库库下创建orders表：

     CREATE TABLE `flink`.`orders` (
     	`order_id` VARCHAR(32) NOT NULL,
     	`order_channel` VARCHAR(32) NULL,
     	PRIMARY KEY (`order_id`)
     )	ENGINE = InnoDB
     	DEFAULT CHARACTER SET = utf8mb4
     	COLLATE = utf8mb4_general_ci;

     插入表数据：

     insert into orders(
       order_id,
       order_channel
     ) values
       ('1', 'webShop'),  
       ('2', 'miniAppShop');

  4. 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。

     注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。

     认证用的username和password硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。Flink Opensource SQL使用DEW管理访问凭据

     CREATE TABLE jdbcSource (
       order_id string,
       order_channel string
     ) WITH (
       'connector' = 'jdbc',
       'url' = 'jdbc:mysql://MySQLAddress:MySQLPort/flink',--flink为RDS MySQL创建的数据库名
       'table-name' = 'orders',
       'username' = 'MySQLUsername',
       'password' = 'MySQLPassword',
       'scan.fetch-size' = '10',
       'scan.auto-commit' = 'true'
     );
     
     CREATE TABLE printSink (
       order_id string,
       order_channel string
     ) WITH (
       'connector' = 'print'
     );
     
     insert into printSink select * from jdbcSource;

  5. 查看taskmanager.out文件中的数据结果，数据结果参考如下：

     +I(1,webShop)
     +I(2,miniAppShop)

• 示例2：使用DataGen源表发送数据，通过JDBC结果表将数据输出到MySQL数据库中。
  1. 参考增强型跨源连接，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。
  2. 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。
  3. 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考创建RDS数据库。
     在flink数据库库下创建orders表：

     CREATE TABLE `flink`.`orders` (
     	`order_id` VARCHAR(32) NOT NULL,
     	`order_channel` VARCHAR(32) NULL,
     	PRIMARY KEY (`order_id`)
     )	ENGINE = InnoDB
     	DEFAULT CHARACTER SET = utf8mb4
     	COLLATE = utf8mb4_general_ci;

  4. 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。

     注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。

     CREATE TABLE dataGenSource (
       order_id string,
       order_channel string
     ) WITH (
       'connector' = 'datagen',
       'fields.order_id.kind' = 'sequence',
       'fields.order_id.start' = '1',
       'fields.order_id.end' = '1000',
       'fields.order_channel.kind' = 'random',
       'fields.order_channel.length' = '5'
     );
     
     CREATE TABLE jdbcSink (
       order_id string,
       order_channel string,
       PRIMARY KEY(order_id) NOT ENFORCED
     ) WITH (
       'connector' = 'jdbc',
       'url' = 'jdbc:mysql://MySQLAddress:MySQLPort/flink',--其中url中的flink表示MySQL中orders表所在的数据库名
       'table-name' = 'orders',
       'username' = 'MySQLUsername',
       'password' = 'MySQLPassword',
       'sink.buffer-flush.max-rows' = '1'
     );
     
     insert into jdbcSink select * from dataGenSource;

  5. 查看表中数据，在MySQL中执行sql查询语句

     select * from orders；

• 示例3：从DataGen源表中读取数据，将JDBC表作为维表，并将二者生成的表信息写入Print结果表中。
  1. 参考增强型跨源连接，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。
  2. 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考测试地址连通性根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。
  3. 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考创建RDS数据库。

     在flink数据库库下创建orders表：

     CREATE TABLE `flink`.`orders` (
     	`order_id` VARCHAR(32) NOT NULL,
     	`order_channel` VARCHAR(32) NULL,
     	PRIMARY KEY (`order_id`)
     )	ENGINE = InnoDB
     	DEFAULT CHARACTER SET = utf8mb4
     	COLLATE = utf8mb4_general_ci;

     插入表数据：

     insert into orders(
       order_id,
       order_channel
     ) values
       ('1', 'webShop'),  
       ('2', 'miniAppShop');

  4. 参考创建Flink OpenSource作业，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。该作业脚本将DataGen为数据源，JDBC作为维表，数据写入到Print结果表。

     注意：创建作业时，在作业编辑界面的“运行参数”处，“Flink版本”选择“1.15”，勾选“保存作业日志”并设置保存作业日志的OBS桶，方便后续查看作业日志。如下脚本中的加粗参数请根据实际环境修改。

     CREATE TABLE dataGenSource (
       order_id string,
       order_time timestamp,
       proctime as Proctime()
     ) WITH (
       'connector' = 'datagen',
       'fields.order_id.kind' = 'sequence',
       'fields.order_id.start' = '1',
       'fields.order_id.end' = '2'
     );
     
     --创建维表
     CREATE TABLE jdbcTable (
       order_id string,
       order_channel string
     ) WITH (
       'connector' = 'jdbc',
       'url' = 'jdbc:mysql://JDBCAddress:JDBCPort/flink',--flink为RDS MySQL中orders表所在的数据库名
       'table-name' = 'orders',
       'username' = 'JDBCUserName',
       'password' = 'JDBCPassWord',
       'lookup.cache.max-rows' = '100',
       'lookup.cache.ttl' = '1000',
       'lookup.cache.caching-missing-key' = 'false',
       'lookup.max-retries' = '5'
     );
     
     CREATE TABLE printSink (
       order_id string,
       order_time timestamp,
       order_channel string
     ) WITH (
       'connector' = 'print'
     );
     
     insert into
       printSink
     SELECT
       dataGenSource.order_id, dataGenSource.order_time, jdbcTable.order_channel
     from
       dataGenSource
       left join jdbcTable for system_time as of dataGenSource.proctime on dataGenSource.order_id = jdbcTable.order_id;

  5. 查看taskmanager.out文件中的数据结果，数据结果参考如下：

     +I(1, xxx, webShop)
     +I(2, xxx, miniAppShop)

常见问题

无