Skewed Data when joining tables using Flink SQL

[Anne Lai <an...@gmail.com>]

Hi,

I have a Flink batch job that needs to join a large skewed table with a
smaller table, and because records are not evenly distributed to each
subtask, it always fails with a "too much data in partition" error.
I explored using DataStream API to broadcast the smaller tables as a
broadcast state, but it is not ideal since it is kept in-memory on all
downstream operators and even the smaller table is still too large to fit
into the memory. Using broadcast state would also introduce cold start
problems where the order of records processed in the connected stream can't
be guaranteed. As for coGroup transformation, it requires time windowing
but both tables are un-windowed.

Is there any workaround or ideas that I can try on? Thanks!

Best,
Anne

Re: Skewed Data when joining tables using Flink SQL

[Anne Lai <an...@gmail.com>]

Hi Yun,

Thanks for correcting the wrong assumption, I do plan to use BATCH
execution mode. It’s good to know that Flink is able to process the
broadcasted state first, although I’m still a bit concerned about the state
size.

Besides this, did I miss any approach or trick that can be applied in BATCH
mode? For example, I read about iterative broadcasting used in Apache
Spark, which basically broadcast a slice of dataset at a time, flush it and
repeat. I wonder if similar method can be done in Flink too.

Thanks,
Anne

On Fri, Jan 7, 2022 at 9:19 AM Yun Gao <yu...@aliyun.com> wrote:

> Hi Anne,
>
> For one thing, for the datastream and broadcast state method, May I have a
> double
> confirmation that are you using BATCH execution mode? I think with [1] for
> BATCH
> mode it should be able to first process the broadcast side before the
> non-broadcast side.
>
> Best,
> Yun
>
>
> [1] https://issues.apache.org/jira/browse/FLINK-20491
>
>
> ------------------Original Mail ------------------
> *Sender:*Anne Lai <an...@gmail.com>
> *Send Date:*Fri Jan 7 17:20:58 2022
> *Recipients:*User <us...@flink.apache.org>
> *Subject:*Skewed Data when joining tables using Flink SQL
>
> Hi,
>>
>> I have a Flink batch job that needs to join a large skewed table with a
>> smaller table, and because records are not evenly distributed to each
>> subtask, it always fails with a "too much data in partition" error.
>> I explored using DataStream API to broadcast the smaller tables as a
>> broadcast state, but it is not ideal since it is kept in-memory on all
>> downstream operators and even the smaller table is still too large to fit
>> into the memory. Using broadcast state would also introduce cold start
>> problems where the order of records processed in the connected stream can't
>> be guaranteed. As for coGroup transformation, it requires time windowing
>> but both tables are un-windowed.
>>
>> Is there any workaround or ideas that I can try on? Thanks!
>>
>> Best,
>> Anne
>>
>

Re: Skewed Data when joining tables using Flink SQL

[Yun Gao <yu...@aliyun.com>]

Hi Anne,

For one thing, for the datastream and broadcast state method, May I have a double 
confirmation that are you using BATCH execution mode? I think with [1] for BATCH
 mode it should be able to first process the broadcast side before the non-broadcast side.

Best,
Yun


[1] https://issues.apache.org/jira/browse/FLINK-20491



 ------------------Original Mail ------------------
Sender:Anne Lai <an...@gmail.com>
Send Date:Fri Jan 7 17:20:58 2022
Recipients:User <us...@flink.apache.org>
Subject:Skewed Data when joining tables using Flink SQL

Hi,

I have a Flink batch job that needs to join a large skewed table with a smaller table, and because records are not evenly distributed to each subtask, it always fails with a "too much data in partition" error. 
I explored using DataStream API to broadcast the smaller tables as a broadcast state, but it is not ideal since it is kept in-memory on all downstream operators and even the smaller table is still too large to fit into the memory. Using broadcast state would also introduce cold start problems where the order of records processed in the connected stream can't be guaranteed. As for coGroup transformation, it requires time windowing but both tables are un-windowed.

Is there any workaround or ideas that I can try on? Thanks!

Best,
Anne