You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Kai Fu <zz...@gmail.com> on 2021/08/06 01:45:45 UTC
Traffic amplification between operators of simple Flink job
Hi team,
We noticed a pattern that there is a traffic amplification for a simple
Flink job which reads from upsert-kafka and sinks to Elasticsearch. The
traffic of the sink operator is 3 times(almost exactly) of the source,
which we do not quite understand. Could anyone help to explain this and if
it's reasonable or not?
[image: image.png]
--
*Best wishes,*
*- Kai*
Re: Traffic amplification between operators of simple Flink job
Posted by Kai Fu <zz...@gmail.com>.
Hi Jing,
Thank you for your reply, times amplification is also what we expect
instead of 3 based on the UB/UA events generated for each update. We'll
deep dive more and try to find out the root cause.
On Fri, Aug 6, 2021 at 3:03 PM JING ZHANG <be...@gmail.com> wrote:
> Hi Kai,
> To be frankly, I think `ChangelogNormalize` would only lead to 2 times
> amplify at most instead of 3 times.
> For example, kafka upsert source schema is (word, cnt), word is primary
> key.
> If Input data of `ChangelogNormalize` is:
> 'Hi', 1
> 'Hi', 2
> 'Hi', 3
> ...
> Then output of `ChangelogNormalize` is :
> I ('Hi, 1')
> UB ('Hi, 1')
> UA ('Hi', 2)
> UB ('Hi', 2)
> UA('Hi', 3)
>
>
> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午11:29写道：
>
>> Hi Jing,
>>
>> We understand that the `changelogNormalize` operator could lead to
>> traffic amplification, but not sure why it's 3 times, could you help us to
>> understand more about the `changelogNormalize` about how it works and what
>> kind of duplicated records could be sent out, so that we can take necessary
>> action for dedup to reduce the traffic.
>>
>> On Fri, Aug 6, 2021 at 10:47 AM JING ZHANG <be...@gmail.com> wrote:
>>
>>> Hi Kai,
>>> There are a `changelogNormalize` operator indeed which would leads to
>>> traffic amplify. It's reasonable, because `changelogNormalize` is needed to
>>> generate a changelog stream without deduplication on the key.
>>>
>>> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午10:44写道：
>>>
>>>> Hi Jing,
>>>>
>>>> Thank you for the quick response, here is the topology graph of the
>>>> operators, the job is written in SQL Table API, with both the source/sink
>>>> described in DDL, and syncing with a simple insert/select DML.
>>>>
>>>> [image: image.png]
>>>>
>>>> On Fri, Aug 6, 2021 at 10:30 AM JING ZHANG <be...@gmail.com>
>>>> wrote:
>>>>
>>>>> Hi Kai,
>>>>> Would you please provide more information (code or graph in WEB UI),
>>>>> which is helpful to analyze the reason.
>>>>>
>>>>> AFAIK, in some cases, there exists traffic amplification, for example,
>>>>> Flink would generates a Normalize operator to materialize the input source
>>>>> if input source is an upsert source or if the input table source produces
>>>>> deduplicate change events. While the Normalize operator would amplify the
>>>>> input traffic, however I think it would only lead to 2 times amplify at
>>>>> most instead of 3 times.
>>>>>
>>>>> Best,
>>>>> JING ZHANG
>>>>>
>>>>> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午9:46写道：
>>>>>
>>>>>> Hi team,
>>>>>>
>>>>>> We noticed a pattern that there is a traffic amplification for a
>>>>>> simple Flink job which reads from upsert-kafka and sinks to Elasticsearch.
>>>>>> The traffic of the sink operator is 3 times(almost exactly) of the source,
>>>>>> which we do not quite understand. Could anyone help to explain this and if
>>>>>> it's reasonable or not?
>>>>>>
>>>>>> [image: image.png]
>>>>>>
>>>>>> --
>>>>>> *Best wishes,*
>>>>>> *- Kai*
>>>>>>
>>>>>
>>>>
>>>> --
>>>> *Best wishes,*
>>>> *- Kai*
>>>>
>>>
>>
>> --
>> *Best wishes,*
>> *- Kai*
>>
>
--
*Best wishes,*
*- Kai*
Re: Traffic amplification between operators of simple Flink job
Posted by JING ZHANG <be...@gmail.com>.
Hi Kai,
To be frankly, I think `ChangelogNormalize` would only lead to 2 times
amplify at most instead of 3 times.
For example, kafka upsert source schema is (word, cnt), word is primary key.
If Input data of `ChangelogNormalize` is:
'Hi', 1
'Hi', 2
'Hi', 3
...
Then output of `ChangelogNormalize` is :
I ('Hi, 1')
UB ('Hi, 1')
UA ('Hi', 2)
UB ('Hi', 2)
UA('Hi', 3)
Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午11:29写道：
> Hi Jing,
>
> We understand that the `changelogNormalize` operator could lead to traffic
> amplification, but not sure why it's 3 times, could you help us to
> understand more about the `changelogNormalize` about how it works and what
> kind of duplicated records could be sent out, so that we can take necessary
> action for dedup to reduce the traffic.
>
> On Fri, Aug 6, 2021 at 10:47 AM JING ZHANG <be...@gmail.com> wrote:
>
>> Hi Kai,
>> There are a `changelogNormalize` operator indeed which would leads to
>> traffic amplify. It's reasonable, because `changelogNormalize` is needed to
>> generate a changelog stream without deduplication on the key.
>>
>> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午10:44写道：
>>
>>> Hi Jing,
>>>
>>> Thank you for the quick response, here is the topology graph of the
>>> operators, the job is written in SQL Table API, with both the source/sink
>>> described in DDL, and syncing with a simple insert/select DML.
>>>
>>> [image: image.png]
>>>
>>> On Fri, Aug 6, 2021 at 10:30 AM JING ZHANG <be...@gmail.com> wrote:
>>>
>>>> Hi Kai,
>>>> Would you please provide more information (code or graph in WEB UI),
>>>> which is helpful to analyze the reason.
>>>>
>>>> AFAIK, in some cases, there exists traffic amplification, for example,
>>>> Flink would generates a Normalize operator to materialize the input source
>>>> if input source is an upsert source or if the input table source produces
>>>> deduplicate change events. While the Normalize operator would amplify the
>>>> input traffic, however I think it would only lead to 2 times amplify at
>>>> most instead of 3 times.
>>>>
>>>> Best,
>>>> JING ZHANG
>>>>
>>>> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午9:46写道：
>>>>
>>>>> Hi team,
>>>>>
>>>>> We noticed a pattern that there is a traffic amplification for a
>>>>> simple Flink job which reads from upsert-kafka and sinks to Elasticsearch.
>>>>> The traffic of the sink operator is 3 times(almost exactly) of the source,
>>>>> which we do not quite understand. Could anyone help to explain this and if
>>>>> it's reasonable or not?
>>>>>
>>>>> [image: image.png]
>>>>>
>>>>> --
>>>>> *Best wishes,*
>>>>> *- Kai*
>>>>>
>>>>
>>>
>>> --
>>> *Best wishes,*
>>> *- Kai*
>>>
>>
>
> --
> *Best wishes,*
> *- Kai*
>
Re: Traffic amplification between operators of simple Flink job
Posted by Kai Fu <zz...@gmail.com>.
Hi Jing,
We understand that the `changelogNormalize` operator could lead to traffic
amplification, but not sure why it's 3 times, could you help us to
understand more about the `changelogNormalize` about how it works and what
kind of duplicated records could be sent out, so that we can take necessary
action for dedup to reduce the traffic.
On Fri, Aug 6, 2021 at 10:47 AM JING ZHANG <be...@gmail.com> wrote:
> Hi Kai,
> There are a `changelogNormalize` operator indeed which would leads to
> traffic amplify. It's reasonable, because `changelogNormalize` is needed to
> generate a changelog stream without deduplication on the key.
>
> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午10:44写道：
>
>> Hi Jing,
>>
>> Thank you for the quick response, here is the topology graph of the
>> operators, the job is written in SQL Table API, with both the source/sink
>> described in DDL, and syncing with a simple insert/select DML.
>>
>> [image: image.png]
>>
>> On Fri, Aug 6, 2021 at 10:30 AM JING ZHANG <be...@gmail.com> wrote:
>>
>>> Hi Kai,
>>> Would you please provide more information (code or graph in WEB UI),
>>> which is helpful to analyze the reason.
>>>
>>> AFAIK, in some cases, there exists traffic amplification, for example,
>>> Flink would generates a Normalize operator to materialize the input source
>>> if input source is an upsert source or if the input table source produces
>>> deduplicate change events. While the Normalize operator would amplify the
>>> input traffic, however I think it would only lead to 2 times amplify at
>>> most instead of 3 times.
>>>
>>> Best,
>>> JING ZHANG
>>>
>>> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午9:46写道：
>>>
>>>> Hi team,
>>>>
>>>> We noticed a pattern that there is a traffic amplification for a simple
>>>> Flink job which reads from upsert-kafka and sinks to Elasticsearch. The
>>>> traffic of the sink operator is 3 times(almost exactly) of the source,
>>>> which we do not quite understand. Could anyone help to explain this and if
>>>> it's reasonable or not?
>>>>
>>>> [image: image.png]
>>>>
>>>> --
>>>> *Best wishes,*
>>>> *- Kai*
>>>>
>>>
>>
>> --
>> *Best wishes,*
>> *- Kai*
>>
>
--
*Best wishes,*
*- Kai*
Re: Traffic amplification between operators of simple Flink job
Posted by JING ZHANG <be...@gmail.com>.
Hi Kai,
There are a `changelogNormalize` operator indeed which would leads to
traffic amplify. It's reasonable, because `changelogNormalize` is needed to
generate a changelog stream without deduplication on the key.
Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午10:44写道：
> Hi Jing,
>
> Thank you for the quick response, here is the topology graph of the
> operators, the job is written in SQL Table API, with both the source/sink
> described in DDL, and syncing with a simple insert/select DML.
>
> [image: image.png]
>
> On Fri, Aug 6, 2021 at 10:30 AM JING ZHANG <be...@gmail.com> wrote:
>
>> Hi Kai,
>> Would you please provide more information (code or graph in WEB UI),
>> which is helpful to analyze the reason.
>>
>> AFAIK, in some cases, there exists traffic amplification, for example,
>> Flink would generates a Normalize operator to materialize the input source
>> if input source is an upsert source or if the input table source produces
>> deduplicate change events. While the Normalize operator would amplify the
>> input traffic, however I think it would only lead to 2 times amplify at
>> most instead of 3 times.
>>
>> Best,
>> JING ZHANG
>>
>> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午9:46写道：
>>
>>> Hi team,
>>>
>>> We noticed a pattern that there is a traffic amplification for a simple
>>> Flink job which reads from upsert-kafka and sinks to Elasticsearch. The
>>> traffic of the sink operator is 3 times(almost exactly) of the source,
>>> which we do not quite understand. Could anyone help to explain this and if
>>> it's reasonable or not?
>>>
>>> [image: image.png]
>>>
>>> --
>>> *Best wishes,*
>>> *- Kai*
>>>
>>
>
> --
> *Best wishes,*
> *- Kai*
>
Re: Traffic amplification between operators of simple Flink job
Posted by Kai Fu <zz...@gmail.com>.
Hi Jing,
Thank you for the quick response, here is the topology graph of the
operators, the job is written in SQL Table API, with both the source/sink
described in DDL, and syncing with a simple insert/select DML.
[image: image.png]
On Fri, Aug 6, 2021 at 10:30 AM JING ZHANG <be...@gmail.com> wrote:
> Hi Kai,
> Would you please provide more information (code or graph in WEB UI), which
> is helpful to analyze the reason.
>
> AFAIK, in some cases, there exists traffic amplification, for example,
> Flink would generates a Normalize operator to materialize the input source
> if input source is an upsert source or if the input table source produces
> deduplicate change events. While the Normalize operator would amplify the
> input traffic, however I think it would only lead to 2 times amplify at
> most instead of 3 times.
>
> Best,
> JING ZHANG
>
> Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午9:46写道：
>
>> Hi team,
>>
>> We noticed a pattern that there is a traffic amplification for a simple
>> Flink job which reads from upsert-kafka and sinks to Elasticsearch. The
>> traffic of the sink operator is 3 times(almost exactly) of the source,
>> which we do not quite understand. Could anyone help to explain this and if
>> it's reasonable or not?
>>
>> [image: image.png]
>>
>> --
>> *Best wishes,*
>> *- Kai*
>>
>
--
*Best wishes,*
*- Kai*
Re: Traffic amplification between operators of simple Flink job
Posted by JING ZHANG <be...@gmail.com>.
Hi Kai,
Would you please provide more information (code or graph in WEB UI), which
is helpful to analyze the reason.
AFAIK, in some cases, there exists traffic amplification, for example,
Flink would generates a Normalize operator to materialize the input source
if input source is an upsert source or if the input table source produces
deduplicate change events. While the Normalize operator would amplify the
input traffic, however I think it would only lead to 2 times amplify at
most instead of 3 times.
Best,
JING ZHANG
Kai Fu <zz...@gmail.com> 于2021年8月6日周五 上午9:46写道：
> Hi team,
>
> We noticed a pattern that there is a traffic amplification for a simple
> Flink job which reads from upsert-kafka and sinks to Elasticsearch. The
> traffic of the sink operator is 3 times(almost exactly) of the source,
> which we do not quite understand. Could anyone help to explain this and if
> it's reasonable or not?
>
> [image: image.png]
>
> --
> *Best wishes,*
> *- Kai*
>