Data Skew

Jan 01, 2020

Stream processing often encounters data skew. A common solution salts keys to redistribute records more evenly, then computes metrics through two-stage aggregation.

Translated from Chinese with AI · Read the original

Data Skew

Data skew is common in stream processing. A typical solution salts keys to spread data across more streams, then aggregates in two stages. Consider this real-time user-count query:

1
select city,count(distinct userId) from table group by city

If a city has too many users, the data becomes skewed. We typically redistribute it using a salt with 1024 possible values:

1
select
2
    city,sum(e1)
3
from
4
    (select city, userId % 1024, count(distinct userId) e1 from table group by city, userId % 1024) tbl
5
group by
6
    city

That handles skew in userId. What if we also need distinct product-SKU counts and both userId and SKU are skewed? The SQL would require several more layers.

How to Optimize

Blink adds SplitAggregateRule to Calcite’s rules, supporting this scenario more abstractly. Here we consider only that rule, ignoring other optimizations such as converting Count Distinct into Group and Count.

How does the rule optimize userId skew?

1
select
2
    city,sum(e1)
3
from
4
    (select city, userId % 1024, count(distinct userId) e1 from table group by city, userId % 1024) tbl
5
group by
6
    city

The corresponding logical plan is:

1
FlinkLogicalAggregate(group=[{0}], agg#0=[$SUM0($2)])
2
+- FlinkLogicalAggregate(group=[{0, 2}], agg#0=[COUNT(DISTINCT $1)])
3
   +- FlinkLogicalCalc(select=[city, userId, MOD(HASH_CODE(userId), 1024) AS $f2])
4
      +- FlinkLogicalCalc(select=[city, userId])
5
         +- FlinkLogicalNativeTableScan(table=[[builtin, default, _DataStreamTable_0]])

It adds MOD(HASH_CODE(userId), 1024) as a salt and computes count distinct in two stages. If both SKU and userId are skewed, two additional fields are needed:

salt UserId: to redistribute UserId
salt SKU: to redistribute SKU

1
+---------+--------+-----+-------------+----------+
2
| City    | UserId | SKU | Salt-UserId | Salt-SKU |
3
+---------+--------+-----+-------------+----------+
4
| Beijing | 1      | 1   | 1           | 1        |
5
| Jinan   | 2      | 3   | 2           | 3        |
6
+---------+--------+-----+-------------+----------+

After grouping by (City, Salt-UserId, Salt-SKU), we cannot derive both Count(Distinct UserId) and Count(Distinct SKU) in the second stage. To solve this, trade space for time by expanding one row into multiple rows. This is not a Blink-specific optimization, but I include it for clarity. Add a Group field:

1
+---------+--------+-----+-------------+----------+-------+
2
| City    | UserId | SKU | Salt-UserId | Salt-SKU | Group |
3
+---------+--------+-----+-------------+----------+-------+
4
| Beijing | 1      | 1   | 1           | null     | 1     |
5
| Beijing | 1      | 1   | null        | 1        | 2     |
6
| Jinan   | 2      | 3   | 2           | null     | 1     |
7
| Jinan   | 2      | 3   | null        | 3        | 2     |
8
+---------+--------+-----+-------------+----------+-------+

Group identifies the aggregation: 1 aggregates Salt-UserId only, while 2 aggregates Salt-SKU only.

After grouping by (City, Salt-UserId, Salt-SKU), filter Group=1 and group by City to obtain Count(Distinct UserId). Filter Group=2 and group by City to obtain Count(Distinct SKU).

The final logical plan looks like this:

1
FlinkLogicalAggregate(group=[{0}], agg#0=[$SUM0($3)], agg#1=[$SUM0($4)])
2
+- FlinkLogicalAggregate(group=[{0, 3, 4}], agg#0=[COUNT(DISTINCT $1) FILTER $5], agg#1=[COUNT(DISTINCT $2) FILTER $6])
3
   +- FlinkLogicalCalc(select=[city, userId, sku, $f3, $f4, =($e, 1) AS $g_1, =($e, 2) AS $g_2])
4
      +- FlinkLogicalExpand(projects=[{city=[$0], userId=[$1], sku=[$2], $f3=[$3], $f4=[null], $e=[1]}, {city=[$0], userId=[$1], sku=[$2], $f3=[null], $f4=[$4], $e=[2]}])
5
         +- FlinkLogicalCalc(select=[city, userId, sku, MOD(HASH_CODE(userId), 1024) AS $f3, MOD(HASH_CODE(sku), 1024) AS $f4])
6
            +- FlinkLogicalNativeTableScan(table=[[builtin, default, _DataStreamTable_0]])

FlinkLogicalExpand expresses the expansion of one row into multiple rows:

1
{city=[$0], userId=[$1], sku=[$2], $f3=[$3], $f4=[null], $e=[1]}
2
{city=[$0], userId=[$1], sku=[$2], $f3=[null], $f4=[$4], $e=[2]}

The first three fields of these two rows are identical; $e = [1] 表示 G r o u p = 1 ，$ e=[2] means Group=2.

Some Thoughts

This is useful only in certain scenarios. For small datasets it is unnecessary, especially because streaming data volumes can be orders of magnitude below batch volumes. Studying Blink’s rule also led me to explore Spark and Calcite’s Count Distinct implementations. This seemingly simple operator ranges from obvious map-based solutions to more advanced bitmaps and operator transformations, each with its strengths. I learned quite a lot.