Flink SQL 通过jdbc连接表的DDL语句
primary key (tenant_id,sku_code) not enforced'url'='jdbc:mysql://ip:3306/database_name','table-name'='table_name'上述案例中涉及到的就是主要的连接参数，更多的参数设置还有
连接器参数 #
|
参数 |
是否必填 |
默认值 |
类型 |
描述 |
|
connector |
必填 |
(none) |
String |
指定使用什么类型的连接器，这里应该是’jdbc’。 |
|
url |
必填 |
(none) |
String |
JDBC 数据库 url。 |
|
table-name |
必填 |
(none) |
String |
连接到 JDBC 表的名称。 |
|
driver |
可选 |
(none) |
String |
用于连接到此 URL 的 JDBC 驱动类名，如果不设置，将自动从 URL 中推导。 |
|
username |
可选 |
(none) |
String |
JDBC 用户名。如果指定了 ‘username’ 和 ‘password’ 中的任一参数，则两者必须都被指定。 |
|
password |
可选 |
(none) |
String |
JDBC 密码。 |
|
connection.max-retry-timeout |
可选 |
60s |
Duration |
最大重试超时时间，以秒为单位且不应该小于 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 |
可选 |
0 |
Integer |
每次循环读取时应该从数据库中获取的行数。如果指定的值为 ‘0’，则该配置项会被忽略。 |
|
scan.auto-commit |
可选 |
true |
Boolean |
在 JDBC 驱动程序上设置 auto-commit 标志， 它决定了每个语句是否在事务中自动提交。有些 JDBC 驱动程序，特别是 Postgres，可能需要将此设置为 false 以便流化结果。 |
|
lookup.cache.max-rows |
可选 |
(none) |
Integer |
lookup cache 的最大行数，若超过该值，则最老的行记录将会过期。 默认情况下，lookup cache 是未开启的。请参阅下面的 Lookup Cache 部分了解更多详情。 |
|
lookup.cache.ttl |
可选 |
(none) |
Duration |
lookup cache 中每一行记录的最大存活时间，若超过该时间，则最老的行记录将会过期。 默认情况下，lookup cache 是未开启的。请参阅下面的 Lookup Cache 部分了解更多详情。 |
|
lookup.cache.caching-missing-key |
可选 |
true |
Boolean |
标记缓存丢失的键，默认为true |
|
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 |
可选 |
(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无更新和删除则可以不指定主键。
免责声明：本站所有文章内容,图片，视频等均是来源于用户投稿和互联网及文摘转载整编而成，不代表本站观点，不承担相关法律责任。其著作权各归其原作者或其出版社所有。如发现本站有涉嫌抄袭侵权/违法违规的内容,侵犯到您的权益，请在线联系站长,一经查实,本站将立刻删除。 本文来自网络,若有侵权，请联系删除，如若转载，请注明出处：https://yundeesoft.com/5311.html
