upsert-kafka source表在sql client中执行sql没有输出

[高速缓存 <gu...@foxmail.com>]

大佬们好：
  我想使用sql-client方式验证版本表的特性，source使用upsert-kafka，表定义如下： 
  
  CREATE TABLE exange_rate (
  id bigint,
  currency string,
  rate decimal(10,3),
  `timestamp` TIMESTAMP(3),
  WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '5' SECOND,
  PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'flink-sql-test-topic1',
  'properties.bootstrap.servers' = '192.168.164.55:9093',
  'key.format' = 'json',
  'value.format' = 'json'
  );


  checkpoint相关配置：
  set execution.checkpointing.interval=10000;
  set state.backend.type = filesystem;
  set state.checkpoints.dir = file://opt/module/flink-states/



  kafka中的全部数据：
  {"id": 1, "currency": "USD", "rate": 0.200, "timestamp": "2024-11-06 01:05:00"}
  {"id": 1, "currency": "USD", "rate": 0.200, "timestamp": "2024-11-06 01:05:00"}
  {"id": 1, "currency": "USD", "rate": 0.212, "timestamp": "2024-11-06 01:35:15"}
  {"id": 1, "currency": "USD", "rate": 0.200, "timestamp": "2024-11-06 01:05:00"}
  {"id": 1, "currency": "USD", "rate": 0.212, "timestamp": "2024-11-06 01:35:15"}
  {"id": 1, "currency": "USD", "rate": 0.200, "timestamp": "2024-11-06 01:05:00"}
  {"id": 1, "currency": "USD", "rate": 0.205, "timestamp": "2024-11-06 01:12:20"}
  {"id": 4, "currency": "GBP", "rate": 0.110, "timestamp": "2024-11-06 01:42:25"}
  {"id": 1, "currency": "USD", "rate": 0.200, "timestamp": "2024-11-06 01:05:00"}
  {"id": 1, "currency": "USD", "rate": 0.200, "timestamp": "2024-11-06 01:05:00"}


  topic只有一个分区，flink使用的默认的并行度配置。


  最后执行如下sql：
  select * from default_catalog.default_database.exange_rate ;


  发现过了很久也没有任何数据输出，也没有任何报错信息。已经这个问题困扰好几天了，请各位帮忙看看问题出在哪里？多谢！

Re: upsert-kafka source表在sql client中执行sql没有输出

[Shawn Huang <hx...@gmail.com>]

你好，可以检查一下 kafka 消息的 key 是否正确，key 如果为空 upsert-kafka source 无法正确输出数据。


> 2024年11月12日 11:36，Yanquan Lv <de...@gmail.com> 写道：
> 
> 你好，你的 Kafka 的数据里有设置消息的键（key）值吗？
> Upsert-Kafka 依赖键进行更新、删除操作。
> 
> 
>> 2024年11月11日 22:30，高速缓存 <gu...@foxmail.com> 写道：
>> 
>> upsert-kafka
> 

Re: upsert-kafka source表在sql client中执行sql没有输出

[高速缓存 <gu...@foxmail.com>]

问题已解决，感谢各位的帮助！

Re: upsert-kafka source表在sql client中执行sql没有输出

[Shawn Huang <hx...@gmail.com>]

upsert-kafka 提供了 value.fields-include 配置项[1]，可以根据 value 中是否包含主键信息配置为 ALL 或 EXCEPT_KEY，默认为 ALL。

[1] https://nightlies.apache.org/flink/flink-docs-master/zh/docs/connectors/table/upsert-kafka/#value-fields-include


> 2024年11月12日 15:28，Yanquan Lv <de...@gmail.com> 写道：
> 
> 建议 value 中不添加主键信息了。
> 如果 value 中也添加了，就需要再配置 key.fields-prefix 来区分 id 这个信息是从 key 获取还是从 value 获取，参考文档[1]。
> 
> [1]
> https://nightlies.apache.org/flink/flink-docs-master/zh/docs/connectors/table/kafka/#%e6%b6%88%e6%81%af%e9%94%aekey%e4%b8%8e%e6%b6%88%e6%81%af%e4%bd%93value%e7%9a%84%e6%a0%bc%e5%bc%8f
> 
> 
> 
>> 2024年11月12日 13:58，高速缓存 <gu...@foxmail.com> 写道：
>> 
>> 你好，我确实没有设置kafka消息的key，我晚点加上key再实验一下，多谢！
>> 不过我有一些疑问，source table的主键信息已经写在了消息的key中，那是否还需要在value中包含主键的信息？就像我的kafka中的消息那样包含一个id字段？
> 
> 

Re: upsert-kafka source表在sql client中执行sql没有输出

[Yanquan Lv <de...@gmail.com>]

建议 value 中不添加主键信息了。
如果 value 中也添加了，就需要再配置 key.fields-prefix 来区分 id 这个信息是从 key 获取还是从 value 获取，参考文档[1]。

[1]https://nightlies.apache.org/flink/flink-docs-master/zh/docs/connectors/table/kafka/#%e6%b6%88%e6%81%af%e9%94%aekey%e4%b8%8e%e6%b6%88%e6%81%af%e4%bd%93value%e7%9a%84%e6%a0%bc%e5%bc%8f

> 2024年11月12日 13:58，高速缓存 <gu...@foxmail.com> 写道：
> 
> 你好，我确实没有设置kafka消息的key，我晚点加上key再实验一下，多谢！
> 不过我有一些疑问，source table的主键信息已经写在了消息的key中，那是否还需要在value中包含主键的信息？就像我的kafka中的消息那样包含一个id字段？

Re: upsert-kafka source表在sql client中执行sql没有输出

[高速缓存 <gu...@foxmail.com>]

你好，我确实没有设置kafka消息的key，我晚点加上key再实验一下，多谢！
不过我有一些疑问，source table的主键信息已经写在了消息的key中，那是否还需要在value中包含主键的信息？就像我的kafka中的消息那样包含一个id字段？

Re: upsert-kafka source表在sql client中执行sql没有输出

[Yanquan Lv <de...@gmail.com>]

你好，你的 Kafka 的数据里有设置消息的键（key）值吗？
Upsert-Kafka 依赖键进行更新、删除操作。


> 2024年11月11日 22:30，高速缓存 <gu...@foxmail.com> 写道：
> 
> upsert-kafka