Re: kafka相关问题

方盛凯 Wed, 10 Jun 2020 00:18:49 -0700

我认为创建一张表并不会触发从source读取的操作，除非你使用select * from table_ddl这类的操作。
至于表的增大，显然是的。 我觉得你可能关注的是否会flush数据进入存储系统，保证不会out-of-memory这类的问题吗？
我个人猜可能有两种方案：
1.考虑到kafka是一种已经实现持久化的数据源，因此我们并非需要将所有的数据缓存到保存到本地，可以只记录读取到的数据位置，下次读取从记录的位置开始；
2.定期向文件系统写入数据。



小学生 <201782...@qq.com> 于2020年6月10日周三 下午2:48写道：

> 各位大佬好，请教一个问题：
> 利用kafka作为流式源，注册内部的flink的动态表，如下面的table_ddl&nbsp;，是否由于'update-mode' =
> 'append'，所以随着kafka消息逐步的推送，会导致table_ddl这个表的数据不断在增大，但是我想要的是kafka每推送一条数据，table_ddl就会相应的是这一条数据，请问这个如何实现呢？（或者有没有啥办法，可以直接从table_ddl中取最新的kafka数据呢）。
>
>
> table_ddl = """
> CREATE TABLE table_ddl&nbsp;(
> &nbsp;trck_id VARCHAR
> ) WITH (
> &nbsp;'connector.type' = 'kafka',
> &nbsp;'connector.version' = 'universal',&nbsp; &nbsp;&nbsp;
> &nbsp;'connector.topic' = 'w',&nbsp; &nbsp;&nbsp;
> &nbsp;'connector.startup-mode' = 'group-offsets',
> &nbsp;'connector.properties.group.id' = 'trck_w',&nbsp;
> &nbsp;'update-mode' = 'append',
> &nbsp;'connector.properties.zookeeper.connect' = '*',
> &nbsp;'connector.properties.bootstrap.servers' = '%#',
> &nbsp;'format.type' = 'json',&nbsp; &nbsp; &nbsp;
> &nbsp;'format.derive-schema' = 'true'&nbsp;
> )
> """

Re: kafka相关问题

回复