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

FlinkSQL使用jdbc连接数据库参数配置及注意事项

时间:2023-06-15

Flink SQL 通过jdbc连接表的DDL语句

create table deal_mysql(tenant_id String,shop_id String,pay_amt decimal(19, 2),pay_qty double,sku_code String,category_code_b string,write_time timestamp,dt Date,primary key (tenant_id,sku_code) not enforced)with('connector'='jdbc','url'='jdbc:mysql://ip:3306/database_name','username'='root','password'='password','table-name'='table_name');

上述案例中涉及到的就是主要的连接参数,更多的参数设置还有

连接器参数 # 参数是否必填默认值类型描述

connector

必填(none)String指定使用什么类型的连接器,这里应该是'jdbc'。

url

必填(none)StringJDBC 数据库 url。

table-name

必填(none)String连接到 JDBC 表的名称。

driver

可选(none)String用于连接到此 URL 的 JDBC 驱动类名,如果不设置,将自动从 URL 中推导。

username

可选(none)StringJDBC 用户名。如果指定了 'username' 和 'password' 中的任一参数,则两者必须都被指定。

password

可选(none)StringJDBC 密码。

connection.max-retry-timeout

可选60sDuration最大重试超时时间,以秒为单位且不应该小于 1 秒。

scan.partition.column

可选(none)String用于将输入进行分区的列名。请参阅下面的分区扫描部分了解更多详情。

scan.partition.num

可选(none)Integer分区数。

scan.partition.lower-bound

可选(none)Integer第一个分区的最小值。

scan.partition.upper-bound

可选(none)Integer最后一个分区的最大值。

scan.fetch-size

可选0Integer每次循环读取时应该从数据库中获取的行数。如果指定的值为 '0',则该配置项会被忽略。

scan.auto-commit

可选trueBoolean在 JDBC 驱动程序上设置 auto-commit 标志, 它决定了每个语句是否在事务中自动提交。有些 JDBC 驱动程序,特别是 Postgres,可能需要将此设置为 false 以便流化结果。

lookup.cache.max-rows

可选(none)Integerlookup cache 的最大行数,若超过该值,则最老的行记录将会过期。 默认情况下,lookup cache 是未开启的。请参阅下面的 Lookup Cache 部分了解更多详情。

lookup.cache.ttl

可选(none)Durationlookup cache 中每一行记录的最大存活时间,若超过该时间,则最老的行记录将会过期。 默认情况下,lookup cache 是未开启的。请参阅下面的 Lookup Cache 部分了解更多详情。

lookup.cache.caching-missing-key

可选trueBoolean标记缓存丢失的键,默认为true

lookup.max-retries

可选3Integer查询数据库失败的最大重试时间。

sink.buffer-flush.max-rows

可选100Integerflush 前缓存记录的最大值,可以设置为 '0' 来禁用它。

sink.buffer-flush.interval

可选1sDurationflush 间隔时间,超过该时间后异步线程将 flush 数据。可以设置为 '0' 来禁用它。注意, 为了完全异步地处理缓存的 flush 事件,可以将 'sink.buffer-flush.max-rows' 设置为 '0' 并配置适当的 flush 时间间隔。

sink.max-retries

可选3Integer写入记录到数据库失败后的最大重试次数。

sink.parallelism

可选(none)Integer用于定义 JDBC sink 算子的并行度。默认情况下,并行度是由框架决定:使用与上游链式算子相同的并行度。

当然也可以在官网中查看更多的jdbc连接的参数配置和使用方法

这里最主要要说的注意点就是primary key的设置:

在上面的案例中设置的primary key是(sku_code,tenent_id)联合主键来保证数据唯一性,如果你jdbc连接的表示当做source表primary key和原表一样或者指定一个就可,但是如果是要作为sink表则需要注意,必须要保证同一批次写入的数据中primary key必须是唯一的,否则会因为主键的存在丢失数据。

在测试中只有一个tenant_id作为主键,当两条数据具有相同的tenant_id数据同时处理时就会出现数据丢失一条。

如官网中的解释

JDBC 连接器允许使用 JDBC 驱动向任意类型的关系型数据库读取或者写入数据。本文档描述了针对关系型数据库如何通过建立 JDBC 连接器来执行 SQL 查询。

如果在 DDL 中定义了主键,JDBC sink 将以 upsert 模式与外部系统交换 UPDATE/DELETE 消息;否则,它将以 append 模式与外部系统交换消息且不支持消费 UPDATE/DELETE 消息。

当写入数据到外部数据库时,Flink 会使用 DDL 中定义的主键。如果定义了主键,则连接器将以 upsert 模式工作,否则连接器将以 append 模式工作。

在 upsert 模式下,Flink 将根据主键判断插入新行或者更新已存在的行,这种方式可以确保幂等性。为了确保输出结果是符合预期的,推荐为表定义主键并且确保主键是底层数据库中表的唯一键或主键。在 append 模式下,Flink 会把所有记录解释为 INSERT 消息,如果违反了底层数据库中主键或者唯一约束,INSERT 插入可能会失败。

有关 PRIMARY KEY 语法的更多详细信息,请参见 CREATE TABLE DDL。

这里要注意的是如果你的数据来源是通过Flink CDC则必须要指定主键,否则会报如下错误

[ERROR] Could not execute SQL statement、Reason:java.lang.IllegalStateException: please declare primary key for sink table when query contains update/delete record.

如果是数据来源为kafka无更新和删除则可以不指定主键。

 

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

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