> 你说的这个参数我看了默认值不是auto吗？需要我显式地指定为force?
Sink upsert materialize would be applied in the following circumstances:
1. `TABLE_EXEC_SINK_UPSERT_MATERIALIZE` set to FORCE and sink's primary key
nonempty.
2. `TABLE_EXEC_SINK_UPSERT_MATERIALIZE` set to AUTO and sink's primary key
doesn't contain upsert
你说的这个参数我看了默认值不是auto吗？需要我显式地指定为force?
Because of the disorder of ChangeLog data caused by Shuffle in distributed
system, the data received by Sink may not be the order of global upsert. So add
upsert materialize operator before upsert sink. It receives the upstream
changelog records and
Hi,
Re *"如何确保同一主键变更记录(上游接的是cdc数据源)的时间先后顺序性呢？", *checking out
ExecutionConfigOptions.TABLE_EXEC_SINK_UPSERT_MATERIALIZE for details about
solution of disordering problems in KeyBy shuffling.
Best,
Shuo
On Wed, Feb 22, 2023 at 10:23 AM casel.chen wrote:
>
>
如果想保证每次写入 mysql 的事件是最新的，需要在 Flink 内部针对事件时间排序取 TOP 1, 可以参考[1]。 但是需要注意这需要使用
state，你可以需要指定合适的 TTL[2] 来保证 state 不会过大
[1]
https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/queries/topn/
[2]
我理解如果sink 表上带有 pk，就会主动 keyby 的[1]。
Best,
Shengkai
[1]
https://github.com/apache/flink/blob/master/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/common/CommonExecSink.java#L188
Shammon FY 于2023年2月20日周一 08:41写道：
> Hi
>
>
Hi
如果join计算的关联字段里没有主键，在将join结果直接写入sink表时，像上面RS提到的，需要自己再增加一次shuffle操作
Best,
Shammon
On Sun, Feb 19, 2023 at 1:43 PM RS wrote:
> Hi，
> connector里面配置的主键控制是写入存储的，有些存储在写入数据的时候，能根据主键自动更新去重
> 所以我感觉，你这里的应该是想在计算的时候shuffle（写入之前），你应该需要先执行一个 group by 主键，然后再执行insert into
>
>
> Thanks
>
>
>
> 在 2023-02-17
Hi，
connector里面配置的主键控制是写入存储的，有些存储在写入数据的时候，能根据主键自动更新去重
所以我感觉，你这里的应该是想在计算的时候shuffle（写入之前），你应该需要先执行一个 group by 主键，然后再执行insert into
Thanks
在 2023-02-17 15:56:51，"casel.chen" 写道：
>作业场景是kafka cdc数据源关联几张redis维表再和其他流表进行双流regular inner
>join，最后将打宽表写入mongodb。使用的是flink 1.13.2 sql模式。开了debug日志。
