You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by shuai xu <xu...@gmail.com> on 2024/01/10 13:23:14 UTC
[DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Hi devs,
I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
A new option is added to control the operator to avoid influencing existing jobs.
Please find more details in the FLIP wiki document [1]. Looking
forward to your feedback.
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
Best,
Xu Shuai
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by shuai xu <xu...@gmail.com>.
Hi Benchao,
I think your suggestion is very reasonable. For most users, having compaction enabled by default if mini-batch enabled is a more beneficial approach. However, I think this is an another thing we could discuss in the future about compaction within minibatch, which is an orthogonal topic to this discussion. Minibatch join itself would follow the option 'table.exec.mini-batch.enabled’, 'table.exec.mini-batch.allow-latency' and 'table.exec.mini-batch.size'.
> 2024年1月16日 18:52，Benchao Li <li...@apache.org> 写道：
>
> shuai,
>
> Thanks for the explanations, I understand the scenario you described
> now. IIUC, this will be a rather rare case that need to disable
> "compaction" when mini-batch is enabled, so I won't be against
> introducing it. However, I would suggest to enable the "compaction" by
> default (if mini-batch enabled), which will benefit most of use cases.
> For others that have special requirements about the changelog semantic
> (no compaction), they can disable compaction by themselves. WDYT?
>
>> This is a relatively large optimization that may pose a significant
>> risk of bugs, so I like to keep it from being enabled by default for
>> now.
> @Jingsong has raised an interesting point that for large optimization
> or new features, we want to have an option for it and disable it by
> default in case of the risk of bugs. I agree with it, mostly.
> Currently there is no standard about whether a change is major or not,
> which means we may run into a situation debating whether a change is
> major or not. Anyway, it's an orthogonal topic to this discussion.
>
> shuai xu <xu...@gmail.com> 于2024年1月16日周二 13:14写道：
>>
>> Hi Benchao,
>>
>> Do you have any other questions about this issue? Also, I would appreciate your thoughts on the proposal to introduce the new option 'table.exec.mini-batch.compact-changes-enabled'. I’m looking forward your feedback.
>>
>>> 2024年1月12日 15:01，shuai xu <xu...@gmail.com> 写道：
>>>
>>> Suppose we currently have a job that joins two CDC sources after de-duplicating them and the output is available for audit analysis, and the user turns off the parameter "table.exec.deduplicate.mini-batch.compact-changes-enabled" to ensure that it does not lose update details. If we don't introduce this parameter, after the user upgrades the version, some update details may be lost due to the mini-batch connection being enabled by default, resulting in distorted audit results.
>>>
>>>> 2024年1月11日 16:19，Benchao Li <li...@apache.org> 写道：
>>>>
>>>>> the change might not be supposed for the downstream of the job which requires details of changelog
>>>>
>>>> Could you elaborate on this a bit? I've never met such kinds of
>>>> requirements before, I'm curious what is the scenario that requires
>>>> this.
>>>>
>>>> shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
>>>>>
>>>>> Thanks for your response, Benchao.
>>>>>
>>>>> Here is my thought on the newly added option.
>>>>> Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
>>>>>
>>>>> As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
>>>>>
>>>>> Best,
>>>>> Xu Shuai
>>>>>
>>>>>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
>>>>>>
>>>>>> Thanks shuai for driving this, mini-batch Join is a very useful
>>>>>> optimization, +1 for the general idea.
>>>>>>
>>>>>> Regarding the configuration
>>>>>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
>>>>>> necessary. The semantic of changelog emitted by the Join operator is
>>>>>> eventual consistency, so there is no much difference between original
>>>>>> Join and mini-batch Join from this aspect. Besides, introducing more
>>>>>> options would make it more complex for users, harder to understand and
>>>>>> maintain, which we should be careful about.
>>>>>>
>>>>>> One thing about the implementation, could you make the new operator
>>>>>> share the same state definition with the original one?
>>>>>>
>>>>>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>>>>>>>
>>>>>>> Hi devs,
>>>>>>>
>>>>>>> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
>>>>>>>
>>>>>>> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
>>>>>>>
>>>>>>> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
>>>>>>> A new option is added to control the operator to avoid influencing existing jobs.
>>>>>>>
>>>>>>> Please find more details in the FLIP wiki document [1]. Looking
>>>>>>> forward to your feedback.
>>>>>>>
>>>>>>> [1]
>>>>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>>>>>>>
>>>>>>> Best,
>>>>>>> Xu Shuai
>>>>>>
>>>>>>
>>>>>>
>>>>>> --
>>>>>>
>>>>>> Best,
>>>>>> Benchao Li
>>>>>
>>>>
>>>>
>>>> --
>>>>
>>>> Best,
>>>> Benchao Li
>>>
>>> Best,
>>> Xu Shuai
>>
>>
>> Best,
>> Xu Shuai
>
>
>
> --
>
> Best,
> Benchao Li
Best,
Xu Shuai
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Benchao Li <li...@apache.org>.
shuai,
Thanks for the explanations, I understand the scenario you described
now. IIUC, this will be a rather rare case that need to disable
"compaction" when mini-batch is enabled, so I won't be against
introducing it. However, I would suggest to enable the "compaction" by
default (if mini-batch enabled), which will benefit most of use cases.
For others that have special requirements about the changelog semantic
(no compaction), they can disable compaction by themselves. WDYT?
> This is a relatively large optimization that may pose a significant
> risk of bugs, so I like to keep it from being enabled by default for
> now.
@Jingsong has raised an interesting point that for large optimization
or new features, we want to have an option for it and disable it by
default in case of the risk of bugs. I agree with it, mostly.
Currently there is no standard about whether a change is major or not,
which means we may run into a situation debating whether a change is
major or not. Anyway, it's an orthogonal topic to this discussion.
shuai xu <xu...@gmail.com> 于2024年1月16日周二 13:14写道：
>
> Hi Benchao,
>
> Do you have any other questions about this issue? Also, I would appreciate your thoughts on the proposal to introduce the new option 'table.exec.mini-batch.compact-changes-enabled'. I’m looking forward your feedback.
>
> > 2024年1月12日 15:01，shuai xu <xu...@gmail.com> 写道：
> >
> > Suppose we currently have a job that joins two CDC sources after de-duplicating them and the output is available for audit analysis, and the user turns off the parameter "table.exec.deduplicate.mini-batch.compact-changes-enabled" to ensure that it does not lose update details. If we don't introduce this parameter, after the user upgrades the version, some update details may be lost due to the mini-batch connection being enabled by default, resulting in distorted audit results.
> >
> >> 2024年1月11日 16:19，Benchao Li <li...@apache.org> 写道：
> >>
> >>> the change might not be supposed for the downstream of the job which requires details of changelog
> >>
> >> Could you elaborate on this a bit? I've never met such kinds of
> >> requirements before, I'm curious what is the scenario that requires
> >> this.
> >>
> >> shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
> >>>
> >>> Thanks for your response, Benchao.
> >>>
> >>> Here is my thought on the newly added option.
> >>> Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
> >>>
> >>> As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
> >>>
> >>> Best,
> >>> Xu Shuai
> >>>
> >>>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
> >>>>
> >>>> Thanks shuai for driving this, mini-batch Join is a very useful
> >>>> optimization, +1 for the general idea.
> >>>>
> >>>> Regarding the configuration
> >>>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
> >>>> necessary. The semantic of changelog emitted by the Join operator is
> >>>> eventual consistency, so there is no much difference between original
> >>>> Join and mini-batch Join from this aspect. Besides, introducing more
> >>>> options would make it more complex for users, harder to understand and
> >>>> maintain, which we should be careful about.
> >>>>
> >>>> One thing about the implementation, could you make the new operator
> >>>> share the same state definition with the original one?
> >>>>
> >>>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
> >>>>>
> >>>>> Hi devs,
> >>>>>
> >>>>> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
> >>>>>
> >>>>> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
> >>>>>
> >>>>> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
> >>>>> A new option is added to control the operator to avoid influencing existing jobs.
> >>>>>
> >>>>> Please find more details in the FLIP wiki document [1]. Looking
> >>>>> forward to your feedback.
> >>>>>
> >>>>> [1]
> >>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
> >>>>>
> >>>>> Best,
> >>>>> Xu Shuai
> >>>>
> >>>>
> >>>>
> >>>> --
> >>>>
> >>>> Best,
> >>>> Benchao Li
> >>>
> >>
> >>
> >> --
> >>
> >> Best,
> >> Benchao Li
> >
> > Best,
> > Xu Shuai
>
>
> Best,
> Xu Shuai
--
Best,
Benchao Li
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by shuai xu <xu...@gmail.com>.
Hi Benchao,
Do you have any other questions about this issue? Also, I would appreciate your thoughts on the proposal to introduce the new option 'table.exec.mini-batch.compact-changes-enabled'. I’m looking forward your feedback.
> 2024年1月12日 15:01，shuai xu <xu...@gmail.com> 写道：
>
> Suppose we currently have a job that joins two CDC sources after de-duplicating them and the output is available for audit analysis, and the user turns off the parameter "table.exec.deduplicate.mini-batch.compact-changes-enabled" to ensure that it does not lose update details. If we don't introduce this parameter, after the user upgrades the version, some update details may be lost due to the mini-batch connection being enabled by default, resulting in distorted audit results.
>
>> 2024年1月11日 16:19，Benchao Li <li...@apache.org> 写道：
>>
>>> the change might not be supposed for the downstream of the job which requires details of changelog
>>
>> Could you elaborate on this a bit? I've never met such kinds of
>> requirements before, I'm curious what is the scenario that requires
>> this.
>>
>> shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
>>>
>>> Thanks for your response, Benchao.
>>>
>>> Here is my thought on the newly added option.
>>> Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
>>>
>>> As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
>>>
>>> Best,
>>> Xu Shuai
>>>
>>>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
>>>>
>>>> Thanks shuai for driving this, mini-batch Join is a very useful
>>>> optimization, +1 for the general idea.
>>>>
>>>> Regarding the configuration
>>>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
>>>> necessary. The semantic of changelog emitted by the Join operator is
>>>> eventual consistency, so there is no much difference between original
>>>> Join and mini-batch Join from this aspect. Besides, introducing more
>>>> options would make it more complex for users, harder to understand and
>>>> maintain, which we should be careful about.
>>>>
>>>> One thing about the implementation, could you make the new operator
>>>> share the same state definition with the original one?
>>>>
>>>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>>>>>
>>>>> Hi devs,
>>>>>
>>>>> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
>>>>>
>>>>> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
>>>>>
>>>>> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
>>>>> A new option is added to control the operator to avoid influencing existing jobs.
>>>>>
>>>>> Please find more details in the FLIP wiki document [1]. Looking
>>>>> forward to your feedback.
>>>>>
>>>>> [1]
>>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>>>>>
>>>>> Best,
>>>>> Xu Shuai
>>>>
>>>>
>>>>
>>>> --
>>>>
>>>> Best,
>>>> Benchao Li
>>>
>>
>>
>> --
>>
>> Best,
>> Benchao Li
>
> Best,
> Xu Shuai
Best,
Xu Shuai
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Jingsong Li <ji...@gmail.com>.
Hi all,
This is a relatively large optimization that may pose a significant
risk of bugs, so I like to keep it from being enabled by default for
now.
Best,
Jingsong
On Fri, Jan 12, 2024 at 3:01 PM shuai xu <xu...@gmail.com> wrote:
>
> Suppose we currently have a job that joins two CDC sources after de-duplicating them and the output is available for audit analysis, and the user turns off the parameter "table.exec.deduplicate.mini-batch.compact-changes-enabled" to ensure that it does not lose update details. If we don't introduce this parameter, after the user upgrades the version, some update details may be lost due to the mini-batch connection being enabled by default, resulting in distorted audit results.
>
> > 2024年1月11日 16:19，Benchao Li <li...@apache.org> 写道：
> >
> >> the change might not be supposed for the downstream of the job which requires details of changelog
> >
> > Could you elaborate on this a bit? I've never met such kinds of
> > requirements before, I'm curious what is the scenario that requires
> > this.
> >
> > shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
> >>
> >> Thanks for your response, Benchao.
> >>
> >> Here is my thought on the newly added option.
> >> Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
> >>
> >> As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
> >>
> >> Best,
> >> Xu Shuai
> >>
> >>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
> >>>
> >>> Thanks shuai for driving this, mini-batch Join is a very useful
> >>> optimization, +1 for the general idea.
> >>>
> >>> Regarding the configuration
> >>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
> >>> necessary. The semantic of changelog emitted by the Join operator is
> >>> eventual consistency, so there is no much difference between original
> >>> Join and mini-batch Join from this aspect. Besides, introducing more
> >>> options would make it more complex for users, harder to understand and
> >>> maintain, which we should be careful about.
> >>>
> >>> One thing about the implementation, could you make the new operator
> >>> share the same state definition with the original one?
> >>>
> >>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
> >>>>
> >>>> Hi devs,
> >>>>
> >>>> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
> >>>>
> >>>> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
> >>>>
> >>>> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
> >>>> A new option is added to control the operator to avoid influencing existing jobs.
> >>>>
> >>>> Please find more details in the FLIP wiki document [1]. Looking
> >>>> forward to your feedback.
> >>>>
> >>>> [1]
> >>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
> >>>>
> >>>> Best,
> >>>> Xu Shuai
> >>>
> >>>
> >>>
> >>> --
> >>>
> >>> Best,
> >>> Benchao Li
> >>
> >
> >
> > --
> >
> > Best,
> > Benchao Li
>
> Best,
> Xu Shuai
>
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by shuai xu <xu...@gmail.com>.
Suppose we currently have a job that joins two CDC sources after de-duplicating them and the output is available for audit analysis, and the user turns off the parameter "table.exec.deduplicate.mini-batch.compact-changes-enabled" to ensure that it does not lose update details. If we don't introduce this parameter, after the user upgrades the version, some update details may be lost due to the mini-batch connection being enabled by default, resulting in distorted audit results.
> 2024年1月11日 16:19，Benchao Li <li...@apache.org> 写道：
>
>> the change might not be supposed for the downstream of the job which requires details of changelog
>
> Could you elaborate on this a bit? I've never met such kinds of
> requirements before, I'm curious what is the scenario that requires
> this.
>
> shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
>>
>> Thanks for your response, Benchao.
>>
>> Here is my thought on the newly added option.
>> Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
>>
>> As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
>>
>> Best,
>> Xu Shuai
>>
>>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
>>>
>>> Thanks shuai for driving this, mini-batch Join is a very useful
>>> optimization, +1 for the general idea.
>>>
>>> Regarding the configuration
>>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
>>> necessary. The semantic of changelog emitted by the Join operator is
>>> eventual consistency, so there is no much difference between original
>>> Join and mini-batch Join from this aspect. Besides, introducing more
>>> options would make it more complex for users, harder to understand and
>>> maintain, which we should be careful about.
>>>
>>> One thing about the implementation, could you make the new operator
>>> share the same state definition with the original one?
>>>
>>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>>>>
>>>> Hi devs,
>>>>
>>>> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
>>>>
>>>> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
>>>>
>>>> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
>>>> A new option is added to control the operator to avoid influencing existing jobs.
>>>>
>>>> Please find more details in the FLIP wiki document [1]. Looking
>>>> forward to your feedback.
>>>>
>>>> [1]
>>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>>>>
>>>> Best,
>>>> Xu Shuai
>>>
>>>
>>>
>>> --
>>>
>>> Best,
>>> Benchao Li
>>
>
>
> --
>
> Best,
> Benchao Li
Best,
Xu Shuai
Re: Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Jane Chan <qi...@gmail.com>.
Hi shuai,
Thanks for your clarification.
The internal behavior of minibatch processing is not well-defined now.
I think you're right on this point. If you change the goal of the newly
introduced configuration to address this issue, then I'm ok with it.
Best,
Jane
On Mon, Jan 15, 2024 at 2:27 PM Xuyang <xy...@163.com> wrote:
> Hi, shuai.
> Thanks for this explaination. This scenario sounds reasonable to me. I
> agree that we need to split the behavior
> in minibatch into two types of options: 1. Whether to open minibatch to
> save batch data; 2. Whether to compress
> the changelog data while saving the batch, and merge the data with the
> same upsert key. For the latter the new option
> ''table.exec.mini-batch.compact-changes-enabled'' looks pretty good.
>
>
>
>
> --
>
> Best！
> Xuyang
>
>
>
>
>
> At 2024-01-12 18:13:12, "shuai xu" <xu...@gmail.com> wrote:
> >Hi all.
> >
> >The point I want to highlight is that minibatch join could potentially
> yield incomplete changelog which existing jobs are not supposed to be. For
> example, the scenario that joins two CDC sources after de-duplicating them
> and the output would be used for audit analysis could not accept incomplete
> changelog. While the minibatch processing itself would not introduce any
> problem.
> >
> >The internal behavior of minibatch processing is not well-defined now. I
> don't think reusing the minibatch option for minibatch join is problematic,
> but precise control is necessary to mitigate the risk of generating
> incomplete changelog within minibatch.
> >
> >Controlling the behavior on changelog within minibatch should be a global
> option. Therefore, I propose introducing a new option
> 'table.exec.mini-batch.compact-changes-enabled' to precisely control
> changelog compaction within minibatch. Then we deprecate the option
> 'table.exec.deduplicate.mini-batch.compact-changes-enabled' . The
> deduplicate operator would fall back to follow the newly introduced option
> and the minibatch join would follow it as well.
> >
> >
> >> 2024年1月12日 16:30，Jane Chan <qi...@gmail.com> 写道：
> >>
> >> Hi shuai,
> >>
> >> Thanks for the update! Regarding the newly introduced configuration, I
> hold
> >> the same concern with Benchao and Xuyang.
> >>
> >> First of all, in most cases, the fact that users choose to enable
> >> mini-batch configuration indicates they are aware of the trade-off
> between
> >> throughput and completeness of the changelog.
> >> And if we finally adopt this configuration solely to avoid state
> >> incompatibility, does it mean that we will need to introduce a new
> >> configuration for every future operator's mini-batch optimization,
> similar
> >> to what we did today?
> >>
> >> Best,
> >> Jane
> >>
> >> On Fri, Jan 12, 2024 at 1:45 PM Xuyang <xy...@163.com> wrote:
> >>
> >>> Hi, Xu Shuai. Thanks for driving this flip.
> >>>
> >>>
> >>> The CDC message amplification of cascade join has always been a problem
> >>> for users. Judging from the
> >>> nexmark results, this optimization is very meaningful. I just have the
> >>> same doubts as Benchao, why can't we
> >>> use minibatch join as the default behavior when the user turns on
> >>> minibatch?
> >>>
> >>>
> >>>> Although the semantic of changelog emitted by the Join operator is
> >>> eventual consistency, the change might
> >>> not be supposed for the downstream of the job which requires details of
> >>> changelog.
> >>>
> >>>
> >>> I think if the user adds the minibatch options to his job to enable
> >>> minibatch, he should know that flink will reduce
> >>> the amount of data sent to downstream by folding CDC messages as much
> as
> >>> possible. In scenarios where all
> >>> details of CDC records need to be retained, such as just synchronizing
> >>> data with jobs from one db to another db,
> >>> users have no reason to enable minibatch.
> >>>
> >>>
> >>> The only scenario I can think of that requires adding this independent
> >>> minibatch join option is to ensure that the state
> >>> is compatible between multiple versions, but we have not promised users
> >>> state compatibility during cross-version upgrades.
> >>>
> >>>
> >>> Maybe we need to figure it out why does the
> >>> 'table.exec.deduplicate.mini-batch.compact-changes-enabled' option
> need to
> >>> be added to deduplicate operator? I think this is the same reason as
> >>> adding a separate parameter to join to control CDC message folding.
> >>>
> >>>
> >>>
> >>>
> >>> --
> >>>
> >>> Best！
> >>> Xuyang
> >>>
> >>>
> >>>
> >>>
> >>>
> >>> 在 2024-01-11 16:19:30，"Benchao Li" <li...@apache.org> 写道：
> >>>>> the change might not be supposed for the downstream of the job which
> >>> requires details of changelog
> >>>>
> >>>> Could you elaborate on this a bit? I've never met such kinds of
> >>>> requirements before, I'm curious what is the scenario that requires
> >>>> this.
> >>>>
> >>>> shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
> >>>>>
> >>>>> Thanks for your response, Benchao.
> >>>>>
> >>>>> Here is my thought on the newly added option.
> >>>>> Users' current jobs are running on a version without minibatch join.
> If
> >>> the existing option to enable minibatch join is utilized, then when
> users'
> >>> jobs are migrated to the new version, the internal behavior of the join
> >>> operation within the jobs will change. Although the semantic of
> changelog
> >>> emitted by the Join operator is eventual consistency, the change might
> not
> >>> be supposed for the downstream of the job which requires details of
> >>> changelog. This newly added option also refers to
> >>> 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
> >>>>>
> >>>>> As for the implementation，The new operator shares the state of the
> >>> original operator and it merely has an additional minibatch for storing
> >>> records to do some optimization. The storage remains consistent, and
> there
> >>> is minor modification to the computational logic.
> >>>>>
> >>>>> Best,
> >>>>> Xu Shuai
> >>>>>
> >>>>>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
> >>>>>>
> >>>>>> Thanks shuai for driving this, mini-batch Join is a very useful
> >>>>>> optimization, +1 for the general idea.
> >>>>>>
> >>>>>> Regarding the configuration
> >>>>>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's
> really
> >>>>>> necessary. The semantic of changelog emitted by the Join operator is
> >>>>>> eventual consistency, so there is no much difference between
> original
> >>>>>> Join and mini-batch Join from this aspect. Besides, introducing more
> >>>>>> options would make it more complex for users, harder to understand
> and
> >>>>>> maintain, which we should be careful about.
> >>>>>>
> >>>>>> One thing about the implementation, could you make the new operator
> >>>>>> share the same state definition with the original one?
> >>>>>>
> >>>>>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
> >>>>>>>
> >>>>>>> Hi devs,
> >>>>>>>
> >>>>>>> I’d like to start a discussion on FLIP-415: Introduce a new join
> >>> operator to support minibatch[1].
> >>>>>>>
> >>>>>>> Currently, when performing cascading connections in Flink, there is
> >>> a pain point of record amplification. Every record join operator
> receives
> >>> would trigger join process. However, if records of +I and -D matches ,
> they
> >>> could be folded to reduce two times of join process. Besides, records
> of
> >>> -U +U might output 4 records in which two records are redundant when
> >>> encountering outer join .
> >>>>>>>
> >>>>>>> To address this issue, this FLIP introduces a new
> >>> MiniBatchStreamingJoinOperator to achieve batch processing which could
> >>> reduce number of outputting redundant messages and avoid unnecessary
> join
> >>> processes.
> >>>>>>> A new option is added to control the operator to avoid influencing
> >>> existing jobs.
> >>>>>>>
> >>>>>>> Please find more details in the FLIP wiki document [1]. Looking
> >>>>>>> forward to your feedback.
> >>>>>>>
> >>>>>>> [1]
> >>>>>>>
> >>>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
> >>>>>>>
> >>>>>>> Best,
> >>>>>>> Xu Shuai
> >>>>>>
> >>>>>>
> >>>>>>
> >>>>>> --
> >>>>>>
> >>>>>> Best,
> >>>>>> Benchao Li
> >>>>>
> >>>>
> >>>>
> >>>> --
> >>>>
> >>>> Best,
> >>>> Benchao Li
> >>>
> >
> >Best,
> >Xu Shuai
> >
>
Re:Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Xuyang <xy...@163.com>.
Hi, shuai.
Thanks for this explaination. This scenario sounds reasonable to me. I agree that we need to split the behavior
in minibatch into two types of options: 1. Whether to open minibatch to save batch data; 2. Whether to compress
the changelog data while saving the batch, and merge the data with the same upsert key. For the latter the new option
''table.exec.mini-batch.compact-changes-enabled'' looks pretty good.
--
Best！
Xuyang
At 2024-01-12 18:13:12, "shuai xu" <xu...@gmail.com> wrote:
>Hi all.
>
>The point I want to highlight is that minibatch join could potentially yield incomplete changelog which existing jobs are not supposed to be. For example, the scenario that joins two CDC sources after de-duplicating them and the output would be used for audit analysis could not accept incomplete changelog. While the minibatch processing itself would not introduce any problem.
>
>The internal behavior of minibatch processing is not well-defined now. I don't think reusing the minibatch option for minibatch join is problematic, but precise control is necessary to mitigate the risk of generating incomplete changelog within minibatch.
>
>Controlling the behavior on changelog within minibatch should be a global option. Therefore, I propose introducing a new option 'table.exec.mini-batch.compact-changes-enabled' to precisely control changelog compaction within minibatch. Then we deprecate the option 'table.exec.deduplicate.mini-batch.compact-changes-enabled' . The deduplicate operator would fall back to follow the newly introduced option and the minibatch join would follow it as well.
>
>
>> 2024年1月12日 16:30，Jane Chan <qi...@gmail.com> 写道：
>>
>> Hi shuai,
>>
>> Thanks for the update! Regarding the newly introduced configuration, I hold
>> the same concern with Benchao and Xuyang.
>>
>> First of all, in most cases, the fact that users choose to enable
>> mini-batch configuration indicates they are aware of the trade-off between
>> throughput and completeness of the changelog.
>> And if we finally adopt this configuration solely to avoid state
>> incompatibility, does it mean that we will need to introduce a new
>> configuration for every future operator's mini-batch optimization, similar
>> to what we did today?
>>
>> Best,
>> Jane
>>
>> On Fri, Jan 12, 2024 at 1:45 PM Xuyang <xy...@163.com> wrote:
>>
>>> Hi, Xu Shuai. Thanks for driving this flip.
>>>
>>>
>>> The CDC message amplification of cascade join has always been a problem
>>> for users. Judging from the
>>> nexmark results, this optimization is very meaningful. I just have the
>>> same doubts as Benchao, why can't we
>>> use minibatch join as the default behavior when the user turns on
>>> minibatch?
>>>
>>>
>>>> Although the semantic of changelog emitted by the Join operator is
>>> eventual consistency, the change might
>>> not be supposed for the downstream of the job which requires details of
>>> changelog.
>>>
>>>
>>> I think if the user adds the minibatch options to his job to enable
>>> minibatch, he should know that flink will reduce
>>> the amount of data sent to downstream by folding CDC messages as much as
>>> possible. In scenarios where all
>>> details of CDC records need to be retained, such as just synchronizing
>>> data with jobs from one db to another db,
>>> users have no reason to enable minibatch.
>>>
>>>
>>> The only scenario I can think of that requires adding this independent
>>> minibatch join option is to ensure that the state
>>> is compatible between multiple versions, but we have not promised users
>>> state compatibility during cross-version upgrades.
>>>
>>>
>>> Maybe we need to figure it out why does the
>>> 'table.exec.deduplicate.mini-batch.compact-changes-enabled' option need to
>>> be added to deduplicate operator? I think this is the same reason as
>>> adding a separate parameter to join to control CDC message folding.
>>>
>>>
>>>
>>>
>>> --
>>>
>>> Best！
>>> Xuyang
>>>
>>>
>>>
>>>
>>>
>>> 在 2024-01-11 16:19:30，"Benchao Li" <li...@apache.org> 写道：
>>>>> the change might not be supposed for the downstream of the job which
>>> requires details of changelog
>>>>
>>>> Could you elaborate on this a bit? I've never met such kinds of
>>>> requirements before, I'm curious what is the scenario that requires
>>>> this.
>>>>
>>>> shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
>>>>>
>>>>> Thanks for your response, Benchao.
>>>>>
>>>>> Here is my thought on the newly added option.
>>>>> Users' current jobs are running on a version without minibatch join. If
>>> the existing option to enable minibatch join is utilized, then when users'
>>> jobs are migrated to the new version, the internal behavior of the join
>>> operation within the jobs will change. Although the semantic of changelog
>>> emitted by the Join operator is eventual consistency, the change might not
>>> be supposed for the downstream of the job which requires details of
>>> changelog. This newly added option also refers to
>>> 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
>>>>>
>>>>> As for the implementation，The new operator shares the state of the
>>> original operator and it merely has an additional minibatch for storing
>>> records to do some optimization. The storage remains consistent, and there
>>> is minor modification to the computational logic.
>>>>>
>>>>> Best,
>>>>> Xu Shuai
>>>>>
>>>>>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
>>>>>>
>>>>>> Thanks shuai for driving this, mini-batch Join is a very useful
>>>>>> optimization, +1 for the general idea.
>>>>>>
>>>>>> Regarding the configuration
>>>>>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
>>>>>> necessary. The semantic of changelog emitted by the Join operator is
>>>>>> eventual consistency, so there is no much difference between original
>>>>>> Join and mini-batch Join from this aspect. Besides, introducing more
>>>>>> options would make it more complex for users, harder to understand and
>>>>>> maintain, which we should be careful about.
>>>>>>
>>>>>> One thing about the implementation, could you make the new operator
>>>>>> share the same state definition with the original one?
>>>>>>
>>>>>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>>>>>>>
>>>>>>> Hi devs,
>>>>>>>
>>>>>>> I’d like to start a discussion on FLIP-415: Introduce a new join
>>> operator to support minibatch[1].
>>>>>>>
>>>>>>> Currently, when performing cascading connections in Flink, there is
>>> a pain point of record amplification. Every record join operator receives
>>> would trigger join process. However, if records of +I and -D matches , they
>>> could be folded to reduce two times of join process. Besides, records of
>>> -U +U might output 4 records in which two records are redundant when
>>> encountering outer join .
>>>>>>>
>>>>>>> To address this issue, this FLIP introduces a new
>>> MiniBatchStreamingJoinOperator to achieve batch processing which could
>>> reduce number of outputting redundant messages and avoid unnecessary join
>>> processes.
>>>>>>> A new option is added to control the operator to avoid influencing
>>> existing jobs.
>>>>>>>
>>>>>>> Please find more details in the FLIP wiki document [1]. Looking
>>>>>>> forward to your feedback.
>>>>>>>
>>>>>>> [1]
>>>>>>>
>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>>>>>>>
>>>>>>> Best,
>>>>>>> Xu Shuai
>>>>>>
>>>>>>
>>>>>>
>>>>>> --
>>>>>>
>>>>>> Best,
>>>>>> Benchao Li
>>>>>
>>>>
>>>>
>>>> --
>>>>
>>>> Best,
>>>> Benchao Li
>>>
>
>Best,
>Xu Shuai
>
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by shuai xu <xu...@gmail.com>.
Hi all.
The point I want to highlight is that minibatch join could potentially yield incomplete changelog which existing jobs are not supposed to be. For example, the scenario that joins two CDC sources after de-duplicating them and the output would be used for audit analysis could not accept incomplete changelog. While the minibatch processing itself would not introduce any problem.
The internal behavior of minibatch processing is not well-defined now. I don't think reusing the minibatch option for minibatch join is problematic, but precise control is necessary to mitigate the risk of generating incomplete changelog within minibatch.
Controlling the behavior on changelog within minibatch should be a global option. Therefore, I propose introducing a new option 'table.exec.mini-batch.compact-changes-enabled' to precisely control changelog compaction within minibatch. Then we deprecate the option 'table.exec.deduplicate.mini-batch.compact-changes-enabled' . The deduplicate operator would fall back to follow the newly introduced option and the minibatch join would follow it as well.
> 2024年1月12日 16:30，Jane Chan <qi...@gmail.com> 写道：
>
> Hi shuai,
>
> Thanks for the update! Regarding the newly introduced configuration, I hold
> the same concern with Benchao and Xuyang.
>
> First of all, in most cases, the fact that users choose to enable
> mini-batch configuration indicates they are aware of the trade-off between
> throughput and completeness of the changelog.
> And if we finally adopt this configuration solely to avoid state
> incompatibility, does it mean that we will need to introduce a new
> configuration for every future operator's mini-batch optimization, similar
> to what we did today?
>
> Best,
> Jane
>
> On Fri, Jan 12, 2024 at 1:45 PM Xuyang <xy...@163.com> wrote:
>
>> Hi, Xu Shuai. Thanks for driving this flip.
>>
>>
>> The CDC message amplification of cascade join has always been a problem
>> for users. Judging from the
>> nexmark results, this optimization is very meaningful. I just have the
>> same doubts as Benchao, why can't we
>> use minibatch join as the default behavior when the user turns on
>> minibatch?
>>
>>
>>> Although the semantic of changelog emitted by the Join operator is
>> eventual consistency, the change might
>> not be supposed for the downstream of the job which requires details of
>> changelog.
>>
>>
>> I think if the user adds the minibatch options to his job to enable
>> minibatch, he should know that flink will reduce
>> the amount of data sent to downstream by folding CDC messages as much as
>> possible. In scenarios where all
>> details of CDC records need to be retained, such as just synchronizing
>> data with jobs from one db to another db,
>> users have no reason to enable minibatch.
>>
>>
>> The only scenario I can think of that requires adding this independent
>> minibatch join option is to ensure that the state
>> is compatible between multiple versions, but we have not promised users
>> state compatibility during cross-version upgrades.
>>
>>
>> Maybe we need to figure it out why does the
>> 'table.exec.deduplicate.mini-batch.compact-changes-enabled' option need to
>> be added to deduplicate operator? I think this is the same reason as
>> adding a separate parameter to join to control CDC message folding.
>>
>>
>>
>>
>> --
>>
>> Best！
>> Xuyang
>>
>>
>>
>>
>>
>> 在 2024-01-11 16:19:30，"Benchao Li" <li...@apache.org> 写道：
>>>> the change might not be supposed for the downstream of the job which
>> requires details of changelog
>>>
>>> Could you elaborate on this a bit? I've never met such kinds of
>>> requirements before, I'm curious what is the scenario that requires
>>> this.
>>>
>>> shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
>>>>
>>>> Thanks for your response, Benchao.
>>>>
>>>> Here is my thought on the newly added option.
>>>> Users' current jobs are running on a version without minibatch join. If
>> the existing option to enable minibatch join is utilized, then when users'
>> jobs are migrated to the new version, the internal behavior of the join
>> operation within the jobs will change. Although the semantic of changelog
>> emitted by the Join operator is eventual consistency, the change might not
>> be supposed for the downstream of the job which requires details of
>> changelog. This newly added option also refers to
>> 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
>>>>
>>>> As for the implementation，The new operator shares the state of the
>> original operator and it merely has an additional minibatch for storing
>> records to do some optimization. The storage remains consistent, and there
>> is minor modification to the computational logic.
>>>>
>>>> Best,
>>>> Xu Shuai
>>>>
>>>>> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
>>>>>
>>>>> Thanks shuai for driving this, mini-batch Join is a very useful
>>>>> optimization, +1 for the general idea.
>>>>>
>>>>> Regarding the configuration
>>>>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
>>>>> necessary. The semantic of changelog emitted by the Join operator is
>>>>> eventual consistency, so there is no much difference between original
>>>>> Join and mini-batch Join from this aspect. Besides, introducing more
>>>>> options would make it more complex for users, harder to understand and
>>>>> maintain, which we should be careful about.
>>>>>
>>>>> One thing about the implementation, could you make the new operator
>>>>> share the same state definition with the original one?
>>>>>
>>>>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>>>>>>
>>>>>> Hi devs,
>>>>>>
>>>>>> I’d like to start a discussion on FLIP-415: Introduce a new join
>> operator to support minibatch[1].
>>>>>>
>>>>>> Currently, when performing cascading connections in Flink, there is
>> a pain point of record amplification. Every record join operator receives
>> would trigger join process. However, if records of +I and -D matches , they
>> could be folded to reduce two times of join process. Besides, records of
>> -U +U might output 4 records in which two records are redundant when
>> encountering outer join .
>>>>>>
>>>>>> To address this issue, this FLIP introduces a new
>> MiniBatchStreamingJoinOperator to achieve batch processing which could
>> reduce number of outputting redundant messages and avoid unnecessary join
>> processes.
>>>>>> A new option is added to control the operator to avoid influencing
>> existing jobs.
>>>>>>
>>>>>> Please find more details in the FLIP wiki document [1]. Looking
>>>>>> forward to your feedback.
>>>>>>
>>>>>> [1]
>>>>>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>>>>>>
>>>>>> Best,
>>>>>> Xu Shuai
>>>>>
>>>>>
>>>>>
>>>>> --
>>>>>
>>>>> Best,
>>>>> Benchao Li
>>>>
>>>
>>>
>>> --
>>>
>>> Best,
>>> Benchao Li
>>
Best,
Xu Shuai
Re: Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Jane Chan <qi...@gmail.com>.
Hi shuai,
Thanks for the update! Regarding the newly introduced configuration, I hold
the same concern with Benchao and Xuyang.
First of all, in most cases, the fact that users choose to enable
mini-batch configuration indicates they are aware of the trade-off between
throughput and completeness of the changelog.
And if we finally adopt this configuration solely to avoid state
incompatibility, does it mean that we will need to introduce a new
configuration for every future operator's mini-batch optimization, similar
to what we did today?
Best,
Jane
On Fri, Jan 12, 2024 at 1:45 PM Xuyang <xy...@163.com> wrote:
> Hi, Xu Shuai. Thanks for driving this flip.
>
>
> The CDC message amplification of cascade join has always been a problem
> for users. Judging from the
> nexmark results, this optimization is very meaningful. I just have the
> same doubts as Benchao, why can't we
> use minibatch join as the default behavior when the user turns on
> minibatch?
>
>
> > Although the semantic of changelog emitted by the Join operator is
> eventual consistency, the change might
> not be supposed for the downstream of the job which requires details of
> changelog.
>
>
> I think if the user adds the minibatch options to his job to enable
> minibatch, he should know that flink will reduce
> the amount of data sent to downstream by folding CDC messages as much as
> possible. In scenarios where all
> details of CDC records need to be retained, such as just synchronizing
> data with jobs from one db to another db,
> users have no reason to enable minibatch.
>
>
> The only scenario I can think of that requires adding this independent
> minibatch join option is to ensure that the state
> is compatible between multiple versions, but we have not promised users
> state compatibility during cross-version upgrades.
>
>
> Maybe we need to figure it out why does the
> 'table.exec.deduplicate.mini-batch.compact-changes-enabled' option need to
> be added to deduplicate operator? I think this is the same reason as
> adding a separate parameter to join to control CDC message folding.
>
>
>
>
> --
>
> Best！
> Xuyang
>
>
>
>
>
> 在 2024-01-11 16:19:30，"Benchao Li" <li...@apache.org> 写道：
> >> the change might not be supposed for the downstream of the job which
> requires details of changelog
> >
> >Could you elaborate on this a bit? I've never met such kinds of
> >requirements before, I'm curious what is the scenario that requires
> >this.
> >
> >shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
> >>
> >> Thanks for your response, Benchao.
> >>
> >> Here is my thought on the newly added option.
> >> Users' current jobs are running on a version without minibatch join. If
> the existing option to enable minibatch join is utilized, then when users'
> jobs are migrated to the new version, the internal behavior of the join
> operation within the jobs will change. Although the semantic of changelog
> emitted by the Join operator is eventual consistency, the change might not
> be supposed for the downstream of the job which requires details of
> changelog. This newly added option also refers to
> 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
> >>
> >> As for the implementation，The new operator shares the state of the
> original operator and it merely has an additional minibatch for storing
> records to do some optimization. The storage remains consistent, and there
> is minor modification to the computational logic.
> >>
> >> Best,
> >> Xu Shuai
> >>
> >> > 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
> >> >
> >> > Thanks shuai for driving this, mini-batch Join is a very useful
> >> > optimization, +1 for the general idea.
> >> >
> >> > Regarding the configuration
> >> > "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
> >> > necessary. The semantic of changelog emitted by the Join operator is
> >> > eventual consistency, so there is no much difference between original
> >> > Join and mini-batch Join from this aspect. Besides, introducing more
> >> > options would make it more complex for users, harder to understand and
> >> > maintain, which we should be careful about.
> >> >
> >> > One thing about the implementation, could you make the new operator
> >> > share the same state definition with the original one?
> >> >
> >> > shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
> >> >>
> >> >> Hi devs,
> >> >>
> >> >> I’d like to start a discussion on FLIP-415: Introduce a new join
> operator to support minibatch[1].
> >> >>
> >> >> Currently, when performing cascading connections in Flink, there is
> a pain point of record amplification. Every record join operator receives
> would trigger join process. However, if records of +I and -D matches , they
> could be folded to reduce two times of join process. Besides, records of
> -U +U might output 4 records in which two records are redundant when
> encountering outer join .
> >> >>
> >> >> To address this issue, this FLIP introduces a new
> MiniBatchStreamingJoinOperator to achieve batch processing which could
> reduce number of outputting redundant messages and avoid unnecessary join
> processes.
> >> >> A new option is added to control the operator to avoid influencing
> existing jobs.
> >> >>
> >> >> Please find more details in the FLIP wiki document [1]. Looking
> >> >> forward to your feedback.
> >> >>
> >> >> [1]
> >> >>
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
> >> >>
> >> >> Best,
> >> >> Xu Shuai
> >> >
> >> >
> >> >
> >> > --
> >> >
> >> > Best,
> >> > Benchao Li
> >>
> >
> >
> >--
> >
> >Best,
> >Benchao Li
>
Re:Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Xuyang <xy...@163.com>.
Hi, Xu Shuai. Thanks for driving this flip.
The CDC message amplification of cascade join has always been a problem for users. Judging from the
nexmark results, this optimization is very meaningful. I just have the same doubts as Benchao, why can't we
use minibatch join as the default behavior when the user turns on minibatch?
> Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might
not be supposed for the downstream of the job which requires details of changelog.
I think if the user adds the minibatch options to his job to enable minibatch, he should know that flink will reduce
the amount of data sent to downstream by folding CDC messages as much as possible. In scenarios where all
details of CDC records need to be retained, such as just synchronizing data with jobs from one db to another db,
users have no reason to enable minibatch.
The only scenario I can think of that requires adding this independent minibatch join option is to ensure that the state
is compatible between multiple versions, but we have not promised users state compatibility during cross-version upgrades.
Maybe we need to figure it out why does the 'table.exec.deduplicate.mini-batch.compact-changes-enabled' option need to
be added to deduplicate operator? I think this is the same reason as adding a separate parameter to join to control CDC message folding.
--
Best！
Xuyang
在 2024-01-11 16:19:30，"Benchao Li" <li...@apache.org> 写道：
>> the change might not be supposed for the downstream of the job which requires details of changelog
>
>Could you elaborate on this a bit? I've never met such kinds of
>requirements before, I'm curious what is the scenario that requires
>this.
>
>shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
>>
>> Thanks for your response, Benchao.
>>
>> Here is my thought on the newly added option.
>> Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
>>
>> As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
>>
>> Best,
>> Xu Shuai
>>
>> > 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
>> >
>> > Thanks shuai for driving this, mini-batch Join is a very useful
>> > optimization, +1 for the general idea.
>> >
>> > Regarding the configuration
>> > "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
>> > necessary. The semantic of changelog emitted by the Join operator is
>> > eventual consistency, so there is no much difference between original
>> > Join and mini-batch Join from this aspect. Besides, introducing more
>> > options would make it more complex for users, harder to understand and
>> > maintain, which we should be careful about.
>> >
>> > One thing about the implementation, could you make the new operator
>> > share the same state definition with the original one?
>> >
>> > shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>> >>
>> >> Hi devs,
>> >>
>> >> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
>> >>
>> >> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
>> >>
>> >> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
>> >> A new option is added to control the operator to avoid influencing existing jobs.
>> >>
>> >> Please find more details in the FLIP wiki document [1]. Looking
>> >> forward to your feedback.
>> >>
>> >> [1]
>> >> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>> >>
>> >> Best,
>> >> Xu Shuai
>> >
>> >
>> >
>> > --
>> >
>> > Best,
>> > Benchao Li
>>
>
>
>--
>
>Best,
>Benchao Li
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Benchao Li <li...@apache.org>.
> the change might not be supposed for the downstream of the job which requires details of changelog
Could you elaborate on this a bit? I've never met such kinds of
requirements before, I'm curious what is the scenario that requires
this.
shuai xu <xu...@gmail.com> 于2024年1月11日周四 13:08写道：
>
> Thanks for your response, Benchao.
>
> Here is my thought on the newly added option.
> Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
>
> As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
>
> Best,
> Xu Shuai
>
> > 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
> >
> > Thanks shuai for driving this, mini-batch Join is a very useful
> > optimization, +1 for the general idea.
> >
> > Regarding the configuration
> > "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
> > necessary. The semantic of changelog emitted by the Join operator is
> > eventual consistency, so there is no much difference between original
> > Join and mini-batch Join from this aspect. Besides, introducing more
> > options would make it more complex for users, harder to understand and
> > maintain, which we should be careful about.
> >
> > One thing about the implementation, could you make the new operator
> > share the same state definition with the original one?
> >
> > shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
> >>
> >> Hi devs,
> >>
> >> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
> >>
> >> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
> >>
> >> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
> >> A new option is added to control the operator to avoid influencing existing jobs.
> >>
> >> Please find more details in the FLIP wiki document [1]. Looking
> >> forward to your feedback.
> >>
> >> [1]
> >> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
> >>
> >> Best,
> >> Xu Shuai
> >
> >
> >
> > --
> >
> > Best,
> > Benchao Li
>
--
Best,
Benchao Li
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by shuai xu <xu...@gmail.com>.
Thanks for your response, Benchao.
Here is my thought on the newly added option.
Users' current jobs are running on a version without minibatch join. If the existing option to enable minibatch join is utilized, then when users' jobs are migrated to the new version, the internal behavior of the join operation within the jobs will change. Although the semantic of changelog emitted by the Join operator is eventual consistency, the change might not be supposed for the downstream of the job which requires details of changelog. This newly added option also refers to 'table.exec.deduplicate.mini-batch.compact-changes-enabled'.
As for the implementation，The new operator shares the state of the original operator and it merely has an additional minibatch for storing records to do some optimization. The storage remains consistent, and there is minor modification to the computational logic.
Best,
Xu Shuai
> 2024年1月10日 22:56，Benchao Li <li...@apache.org> 写道：
>
> Thanks shuai for driving this, mini-batch Join is a very useful
> optimization, +1 for the general idea.
>
> Regarding the configuration
> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
> necessary. The semantic of changelog emitted by the Join operator is
> eventual consistency, so there is no much difference between original
> Join and mini-batch Join from this aspect. Besides, introducing more
> options would make it more complex for users, harder to understand and
> maintain, which we should be careful about.
>
> One thing about the implementation, could you make the new operator
> share the same state definition with the original one?
>
> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>>
>> Hi devs,
>>
>> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
>>
>> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
>>
>> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
>> A new option is added to control the operator to avoid influencing existing jobs.
>>
>> Please find more details in the FLIP wiki document [1]. Looking
>> forward to your feedback.
>>
>> [1]
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>>
>> Best,
>> Xu Shuai
>
>
>
> --
>
> Best,
> Benchao Li
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by shuai xu <xu...@gmail.com>.
Hi Jane,
Thanks for your reminder! I missed this.
I updated the FLIP with the UML of MiniBatchStreamingJoinOperator and linking my POC implementation as reference.
They are placed in the part of Proposed Changes.
Best,
Xu Shuai
> 2024年1月11日 11:18，Jane Chan <qi...@gmail.com> 写道：
>
> Hi shuai,
>
> Thanks for initiating the discussion. The mini-batch join optimization is
> very helpful, particularly for optimizing outer join conditions in CDC
> sources and handling cascade joins. And +1 for the proposal.
>
> However, I don't see any details on the proposed
> "MiniBatchStreamingJoinOperator", would you mind elaborating more about it?
>
> Best,
> Jane
>
>
> On Wed, Jan 10, 2024 at 10:56 PM Benchao Li <li...@apache.org> wrote:
>
>> Thanks shuai for driving this, mini-batch Join is a very useful
>> optimization, +1 for the general idea.
>>
>> Regarding the configuration
>> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
>> necessary. The semantic of changelog emitted by the Join operator is
>> eventual consistency, so there is no much difference between original
>> Join and mini-batch Join from this aspect. Besides, introducing more
>> options would make it more complex for users, harder to understand and
>> maintain, which we should be careful about.
>>
>> One thing about the implementation, could you make the new operator
>> share the same state definition with the original one?
>>
>> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>>>
>>> Hi devs,
>>>
>>> I’d like to start a discussion on FLIP-415: Introduce a new join
>> operator to support minibatch[1].
>>>
>>> Currently, when performing cascading connections in Flink, there is a
>> pain point of record amplification. Every record join operator receives
>> would trigger join process. However, if records of +I and -D matches , they
>> could be folded to reduce two times of join process. Besides, records of
>> -U +U might output 4 records in which two records are redundant when
>> encountering outer join .
>>>
>>> To address this issue, this FLIP introduces a new
>> MiniBatchStreamingJoinOperator to achieve batch processing which could
>> reduce number of outputting redundant messages and avoid unnecessary join
>> processes.
>>> A new option is added to control the operator to avoid influencing
>> existing jobs.
>>>
>>> Please find more details in the FLIP wiki document [1]. Looking
>>> forward to your feedback.
>>>
>>> [1]
>>>
>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>>>
>>> Best,
>>> Xu Shuai
>>
>>
>>
>> --
>>
>> Best,
>> Benchao Li
>>
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Jane Chan <qi...@gmail.com>.
Hi shuai,
Thanks for initiating the discussion. The mini-batch join optimization is
very helpful, particularly for optimizing outer join conditions in CDC
sources and handling cascade joins. And +1 for the proposal.
However, I don't see any details on the proposed
"MiniBatchStreamingJoinOperator", would you mind elaborating more about it?
Best,
Jane
On Wed, Jan 10, 2024 at 10:56 PM Benchao Li <li...@apache.org> wrote:
> Thanks shuai for driving this, mini-batch Join is a very useful
> optimization, +1 for the general idea.
>
> Regarding the configuration
> "table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
> necessary. The semantic of changelog emitted by the Join operator is
> eventual consistency, so there is no much difference between original
> Join and mini-batch Join from this aspect. Besides, introducing more
> options would make it more complex for users, harder to understand and
> maintain, which we should be careful about.
>
> One thing about the implementation, could you make the new operator
> share the same state definition with the original one?
>
> shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
> >
> > Hi devs,
> >
> > I’d like to start a discussion on FLIP-415: Introduce a new join
> operator to support minibatch[1].
> >
> > Currently, when performing cascading connections in Flink, there is a
> pain point of record amplification. Every record join operator receives
> would trigger join process. However, if records of +I and -D matches , they
> could be folded to reduce two times of join process. Besides, records of
> -U +U might output 4 records in which two records are redundant when
> encountering outer join .
> >
> > To address this issue, this FLIP introduces a new
> MiniBatchStreamingJoinOperator to achieve batch processing which could
> reduce number of outputting redundant messages and avoid unnecessary join
> processes.
> > A new option is added to control the operator to avoid influencing
> existing jobs.
> >
> > Please find more details in the FLIP wiki document [1]. Looking
> > forward to your feedback.
> >
> > [1]
> >
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
> >
> > Best,
> > Xu Shuai
>
>
>
> --
>
> Best,
> Benchao Li
>
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by Benchao Li <li...@apache.org>.
Thanks shuai for driving this, mini-batch Join is a very useful
optimization, +1 for the general idea.
Regarding the configuration
"table.exec.stream.join.mini-batch-enabled", I'm not sure it's really
necessary. The semantic of changelog emitted by the Join operator is
eventual consistency, so there is no much difference between original
Join and mini-batch Join from this aspect. Besides, introducing more
options would make it more complex for users, harder to understand and
maintain, which we should be careful about.
One thing about the implementation, could you make the new operator
share the same state definition with the original one?
shuai xu <xu...@gmail.com> 于2024年1月10日周三 21:23写道：
>
> Hi devs,
>
> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
>
> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
>
> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
> A new option is added to control the operator to avoid influencing existing jobs.
>
> Please find more details in the FLIP wiki document [1]. Looking
> forward to your feedback.
>
> [1]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>
> Best,
> Xu Shuai
--
Best,
Benchao Li
Re: [DISCUSS] FLIP-415: Introduce a new join operator to support minibatch
Posted by shuai xu <xu...@gmail.com>.
Hi all,
Thank you for the valuable input.
Based on the current discussion, the minibatch join is prepared to follow the existing three options of 'table.exec.mini-batch.enabled’, 'table.exec.mini-batch.allow-latency' and 'table.exec.mini-batch.size’. As for the compaction within the minibatch that was mentioned in the discussion, it could be discussed in a future FLIP.
Do any of you have further questions regarding this FLIP? If there are no more comments, I would like to open a voting thread at 12 a.m. UTC+8 on January 19th.
> 2024年1月10日 21:23，shuai xu <xu...@gmail.com> 写道：
>
> Hi devs,
>
> I’d like to start a discussion on FLIP-415: Introduce a new join operator to support minibatch[1].
>
> Currently, when performing cascading connections in Flink, there is a pain point of record amplification. Every record join operator receives would trigger join process. However, if records of +I and -D matches , they could be folded to reduce two times of join process. Besides, records of -U +U might output 4 records in which two records are redundant when encountering outer join .
>
> To address this issue, this FLIP introduces a new MiniBatchStreamingJoinOperator to achieve batch processing which could reduce number of outputting redundant messages and avoid unnecessary join processes.
> A new option is added to control the operator to avoid influencing existing jobs.
>
> Please find more details in the FLIP wiki document [1]. Looking
> forward to your feedback.
>
> [1]
> https://cwiki.apache.org/confluence/display/FLINK/FLIP-415%3A+Introduce+a+new+join+operator+to+support+minibatch
>
> Best,
> Xu Shuai
Best,
Xu Shuai