[
https://issues.apache.org/jira/browse/FLINK-34552?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17822172#comment-17822172
]
Martijn Visser commented on FLINK-34552:
----------------------------------------
A couple of points from my point of view:
{quote}produces messages with duplicate payload within PK{quote}
That basically conflicts with the entire premise of Flink where a primary key
constraint is a hint for Flink to leverage for optimizations.
{quote}The Deduplication window doesn't work well for my case because the
interval between duplicates is one day and I don't want my data to be delayed
if I use such a big window.{quote}
But how would it then work at the source? Aren't you just moving the problem
from somewhere else in the logic to the front of the logic, since then the
source would have to keep this interval there? You will still encounter a large
delay in that case.
{quote}Kafka topics a and b are CDC data streams and contain DELETE and REFRESH
messages. {quote}
Are you actually using the upsert-kafka source, and treat these input sources
as a changelog stream?
> Support message deduplication for input data sources
> ----------------------------------------------------
>
> Key: FLINK-34552
> URL: https://issues.apache.org/jira/browse/FLINK-34552
> Project: Flink
> Issue Type: New Feature
> Components: Table SQL / API, Table SQL / Runtime
> Reporter: Sergey Anokhovskiy
> Priority: Major
>
> My main proposal is to have duplicate message suppression logic as a part of
> Flink Table API to be able to suppress duplicates from the input sources. It
> might be a parameter provided by the user if they want to suppress duplicates
> from the input source or not. Below I provided more details about my use case
> and available approaches.
>
> I have a Flink job which reads from two keyed kafka topics and emits messages
> to the keyed kafka topic. The Flink job executes the join query:
> SELECT a.id, adata, bdata
> FROM a
> JOIN b
> ON a.id = b.id
>
> One of the input kafka topics produces messages with duplicate payload within
> PK in addition to meaningful data. That causes duplicates in the output topic
> and creates extra load to the downstream services.
>
> I was looking for a way to suppress duplicates and I found two strategies
> which doesn't seem to work for my use case:
> # Based on the deduplication window as a kafka[ sink
> buffer|https://github.com/apache/flink-connector-kafka/blob/main/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/ReducingUpsertSink.java#L46]
> for example. The Deduplication window doesn't work well for my case because
> the interval between duplicates is one day and I don't want my data to be
> delayed if I use such a big window.
>
> # Using
> [ROW_NUMBER|https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/sql/queries/deduplication/]
> . Unfortunately, this approach doesn't suit my use case either. Kafka topics
> a and b are CDC data streams and contain DELETE and REFRESH messages. If
> DELETE and REFRESH messages are coming with the same payload the job will
> suppress the last message which will lead to the incorrect output result. If
> I add message_type to the PARTITION key then the job will not be able to
> process messages sequences like this: DELETE->REFRESH->DELETE (with the same
> payload and PK), because the last message will be suppressed which will lead
> to the incorrect output result.
>
> Finally, I had to create a separate custom Flink service which reads the
> output topic of the initial job and suppresses duplicates keeping hashes of
> the last processed message for each PK in the Flink state.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
