flink-user-zh mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Luan Cooper <gc.su...@gmail.com>
Subject Re: Streaming SQL 的 Source/Sink 在 Append Mode/Upsert Mode 上的疑问
Date Fri, 08 May 2020 07:32:53 GMT
圆满,感谢

On Fri, May 8, 2020 at 10:19 AM Jark Wu <imjark@gmail.com> wrote:

> Hi,
>
> FLIP-95 和 FLIP-105 后,上述的 query 就可以原生支持了。
> FLIP-95 和 FLIP-105 的核心工作就是识别 binlog 中的 update/delete/insert 消息,而不是全当成
> append 消息。
> 预计 1.11 能见到这些功能。
>
> Best,
> Jark
>
> On Thu, 7 May 2020 at 21:34, oliver <zzhbwh@foxmail.com> wrote:
>
> > > 有其他办法吗
> >
> > 可以尝试group by id并配合UDF:LAST_VALUE,
> > SQL示例如下:
> > insert into sink_es
> > select id,LAST_VALUE(column_name)
> > from binlog group by id;
> >
> >
> > Best,
> > Oliver yunchang
> >
> > > 2020年5月7日 下午8:55,Luan Cooper <gc.suprs@gmail.com> 写道:
> > >
> > > 自定义 sink 有这么几个疑问
> > > 1. 自带的 sink 得都改成 upsert 比如 jdbc/es
> > > 2. 这样 append/upsert 代码有大量重复
> > > 3. 和 flink 对 append/upsert 流的定义有冲突,有额外 hack 的解释成本
> > > 4. 得有地方另外指定 update key
> > >
> > > 这么做感觉会挖坑
> > >
> > > 有其他办法吗
> > >
> > > lec ssmi <shicheng31604@gmail.com>于2020年5月7日 周四20:42写道:
> > >
> > >> 使用自定义的Table Sink就可以了啊.
> > >>
> > >> Luan Cooper <gc.suprs@gmail.com> 于2020年5月7日周四 下午8:39写道:
> > >>
> > >>> Hi
> > >>>
> > >>> 有这么个场景,需要将 MySQL 的对一张表的更改 同步到
ElasticSearch 中,两个地方表的 PRIMARY KEY 都是
> > >> ID,SQL
> > >>> 如下
> > >>>
> > >>> INSERT INTO sink_es // 将更改同步 upsert 到 ES
> > >>> SELECT *
> > >>> FROM binlog // mysql 表的 binlog
> > >>>
> > >>> 假设对于 MySQL 中 id = 1 的变更有 10 条,需要在 ES 上都更新
id = 1 的一条记录
> > >>> 但是上面的 SQL 是做不到的,只会一直 Insert
> > >>>
> > >>> 如果想 Upsert ES 的话,就得让 Source 表变成 Upsert Mode,但是现在好像不支持?
> > >>>
> > >>> 社区的 FLIP-87
> > >>>
> > >>>
> > >>
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP+87%3A+Primary+key+constraints+in+Table+API
> > >>> 可以解决这个问题吗?
> > >>>
> > >>> 感谢
> > >>>
> > >>
> >
> >
>
Mime
  • Unnamed multipart/alternative (inline, None, 0 bytes)
View raw message