# JDBC #### 功能描述 JDBC连接器是Flink内置的Connector，提供了对MySQL、PostgreSQL等常见数据库的读写支持。表类型支持源表、结果表和维表。 表1支持类别 | 类别 | 详情 | |:---|:---| | 支持表类型 | 源表、维表、结果表 | #### 前提条件 - 要与实例建立增强型跨源连接，且用户可以根据实际所需设置相应安全组规则。 - 如何建立增强型跨源连接，请参考《数据湖探索用户指南》中[增强型跨源连接](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)章节。 - 如何设置安全组规则，请参见[《虚拟私有云用户指南》](https://support.huaweicloud.com/usermanual-vpc/zh-cn_topic_0073379079.html)中"安全组"章节。 #### 注意事项 - JDBC结果表如果定义了主键，将以upsert模式与外部系统交换UPDATE/DELETE消息；否则，它将以append模式与外部系统交换消息，不支持消费UPDATE/DELETE消息。 - 创建Flink OpenSource SQL作业时，在作业编辑界面的"运行参数"处，"Flink版本"需要选择"1.15"，勾选"保存作业日志"并设置保存作业日志的OBS桶，方便后续查看作业日志。 - 认证用的username和password等硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。[Flink Opensource SQL使用DEW管理访问凭据](https://support.huaweicloud.com/devg-dli/dli_09_0210.html) #### 语法格式 ``` 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' = '' ); ``` #### 参数说明 表2参数说明 | 参数 | 是否必选 | 默认值 | 类型 | 说明 | |:---|:---|:---|:---|:---| | connector | 是 | 无 | String | 指定要使用的连接器，当前固定为'jdbc'。 | | url | 是 | 无 | String | 数据库的URL。 - 连接MySQL数据库时，格式为：jdbc:mysql://***MySQLAddress**:* ***MySQLPort*** /**dbName** 。 - 连接PostgreSQL数据库时，格式为：jdbc:postgresql://***PostgreSQLAddress**:* ***PostgreSQLPort*** /**dbName**。 | | table-name | 是 | 无 | String | 读取数据库中的数据所在的表名。 | | driver | 否 | 无 | String | 连接数据库所需要的驱动。如果未配置，则会自动通过URL提取。 - MySQL数据库默认驱动为com.mysql.jdbc.Driver。 - PostgreSQL数据库默认驱动为org.postgresql.Driver。 | | username | 否 | 无 | String | 数据库认证用户名，需要和'password'一起配置。 | | password | 否 | 无 | String | 数据库认证密码，需要和'username'一起配置。 | | connection.max-retry-timeout | 否 | 60s | Duration | 尝试连接数据库服务器最大重试超时时间，不应小于1s。 | | scan.partition.column | 否 | 无 | String | 用于对输入进行分区的列名。分区扫描参数，具体请参考[分区扫描功能介绍]。 | | scan.partition.num | 否 | 无 | Integer | 分区的个数。分区扫描参数，具体请参考[分区扫描功能介绍]。 | | scan.partition.lower-bound | 否 | 无 | Integer | 第一个分区的最小值。分区扫描参数，具体请参考[分区扫描功能介绍]。 | | scan.partition.upper-bound | 否 | 无 | Integer | 最后一个分区的最大值。分区扫描参数，具体请参考[分区扫描功能介绍]。 | | scan.fetch-size | 否 | 0 | Integer | 每次从数据库拉取数据的行数。如果指定为0，则会忽略sql hint。 | | scan.auto-commit | 否 | true | Boolean | 是否设置自动提交，以确定事务中的每个statement是否自动提交 | | lookup.cache.max-rows | 否 | 无 | Integer | lookup cache的最大行数，如果超过该值，缓存中最先添加的条目将被标记为过期。 默认情况下，lookup cache是未开启的。具体请参考[Lookup Cache功能介绍]。 | | lookup.cache.ttl | 否 | 无 | Duration | lookup cache中每一行记录的最大存活时间，如果超过该时间，缓存中最先添加的条目将被标记为过期。 默认情况下，lookup cache是未开启的。具体请参考[Lookup Cache功能介绍]。 | | lookup.cache.caching-missing-key | 否 | true | Boolean | 是否缓存空查询结果，默认为true。具体请参考[Lookup Cache功能介绍]。 | | lookup.max-retries | 否 | 3 | Integer | 查询数据库失败的最大重试次数。 | | sink.buffer-flush.max-rows | 否 | 100 | Integer | flush前缓存记录的最大值，可以设置为 '0' 来禁用它。 | | sink.buffer-flush.interval | 否 | 1s | Duration | flush间隔时间，超过该时间后异步线程将flush数据。可以设置为 '0' 来禁用它。如果想完全异步地处理缓存的flush事件，可以将 'sink.buffer-flush.max-rows' 设置为 '0' ，并配置适当的flush时间间隔。 | | sink.max-retries | 否 | 3 | Integer | 写入到数据库失败后的最大重试次数。 | | sink.parallelism | 否 | 无 | Integer | 用于定义JDBC sink算子的并行度。默认情况下，并行度是由框架决定，即与上游并行度一致。 | #### 分区扫描功能介绍 为了加速Source任务实例中的数据读取，Flink为JDBC表提供了分区扫描功能。以下参数定义了从多个任务并行读取时如何对表进行分区。 - scan.partition.column：用于对输入进行分区的列名，该列的数据类型必须是数字，日期或时间戳。 - scan.partition.num: 分区数。 - scan.partition.lower-bound：第一个分区的最小值。 - scan.partition.upper-bound：最后一个分区的最大值。 ![](https://support.huaweicloud.com/sqlref-flink-dli/public_sys-resources/note_3.0-zh-cn.png) - **建表时以上扫描分区参数必须同时存在或者同时不存在。** - 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来切换行为。 #### 数据类型映射 表3数据类型映射 | MySQL类型 | PostgreSQL类型 | Flink SQL类型 | |:---|:---|:---| | TINYINT | - | TINYINT | | SMALLINT TINYINT UNSIGNED | SMALLINT INT2 SMALLSERIAL SERIAL2 | SMALLINT | | INT MEDIUMINT SMALLINT UNSIGNED | INTEGER SERIAL | INT | | BIGINT INT UNSIGNED | BIGINT BIGSERIAL | BIGINT | | BIGINT UNSIGNED | - | DECIMAL(20, 0) | | BIGINT | BIGINT | BIGINT | | FLOAT | REAL FLOAT4 | FLOAT | | DOUBLE DOUBLE PRECISION | FLOAT8 DOUBLE PRECISION | DOUBLE | | NUMERIC(p, s) DECIMAL(p, s) | NUMERIC(p, s) DECIMAL(p, s) | DECIMAL(p, s) | | BOOLEAN TINYINT(1) | BOOLEAN | BOOLEAN | | DATE | DATE | DATE | | TIME \[(p)\] | TIME \[(p)\] \[WITHOUT TIMEZONE\] | TIME \[(p)\] \[WITHOUT TIMEZONE\] | | DATETIME \[(p)\] | TIMESTAMP \[(p)\] \[WITHOUT TIMEZONE\] | TIMESTAMP \[(p)\] \[WITHOUT TIMEZONE\] | | CHAR(n) VARCHAR(n) TEXT | CHAR(n) CHARACTER(n) VARCHAR(n) CHARACTER VARYING(n) TEXT | STRING | | BINARY VARBINARY BLOB | BYTEA | BYTES | | - | ARRAY | ARRAY | #### 示例 - **示例1：使用JDBC作为数据源，Print作为结果表，从RDS MySQL数据库中读取数据，并写入到Print结果表中。** 1. 参考[增强型跨源连接](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 2. 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考[测试地址连通性](https://support.huaweicloud.com/usermanual-dli/dli_01_0489.html)根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 3. 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考[创建RDS数据库](https://support.huaweicloud.com/usermanual-rds-mysql/rds_05_0019.html)。 在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作业](https://support.huaweicloud.com/usermanual-dli/dli_01_0498.html)，创建flink opensource sql作业，输入以下作业运行脚本，提交运行作业。 注意：创建作业时，在作业编辑界面的"运行参数"处，"Flink版本"选择"1.15"，勾选"保存作业日志"并设置保存作业日志的OBS桶，方便后续查看作业日志。**如下脚本中的加粗参数请根据实际环境修改**。 认证用的username和password硬编码到代码中或者明文存储都有很大的安全风险，建议使用DEW管理凭证。配置文件或者环境变量中密文存放，使用时解密，确保安全。[Flink Opensource SQL使用DEW管理访问凭据](https://support.huaweicloud.com/devg-dli/dli_09_0210.html) ``` 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. 参考[增强型跨源连接](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 2. 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考[测试地址连通性](https://support.huaweicloud.com/usermanual-dli/dli_01_0489.html)根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 3. 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考[创建RDS数据库](https://support.huaweicloud.com/usermanual-rds-mysql/rds_05_0019.html)。 在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作业](https://support.huaweicloud.com/usermanual-dli/dli_01_0498.html)，创建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. 参考[增强型跨源连接](https://support.huaweicloud.com/usermanual-dli/dli_01_0426.html)，根据RDS MySQL所在的虚拟私有云和子网创建相应的增强型跨源，并绑定所要使用的Flink弹性资源池。 2. 设置RDS MySQL的安全组，添加入向规则使其对Flink的队列网段放通。参考[测试地址连通性](https://support.huaweicloud.com/usermanual-dli/dli_01_0489.html)根RDS的地址测试队列连通性。如果能连通，则表示跨源已经绑定成功，否则表示未成功。 3. 登录RDS MySQL，并使用下述命令在flink库下创建orders表，并插入数据。创建数据库的操作可以参考[创建RDS数据库](https://support.huaweicloud.com/usermanual-rds-mysql/rds_05_0019.html)。 在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作业](https://support.huaweicloud.com/usermanual-dli/dli_01_0498.html)，创建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) ``` #### 常见问题 无