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' );上述案例中涉及到的就是主要的连接参数,更多的参数设置还有
连接器参数?#当然也可以在官网中查看更多的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无更新和删除则可以不指定主键。
?
1.本站遵循行业规范,任何转载的稿件都会明确标注作者和来源;2.本站的原创文章,会注明原创字样,如未注明都非原创,如有侵权请联系删除!;3.作者投稿可能会经我们编辑修改或补充;4.本站不提供任何储存功能只提供收集或者投稿人的网盘链接。 |