Hey Guys, I have flink-sql application that just perform simple insert into enrich table by joining multiple tables.

create table T1 (id int, client_id string ... primary key (id) ) WITH ( 'connector' = 'upsert-kafka','topic' = 'T1', ...)
create table T2 (id int, client_id string, t1_id int, ... primary key (id) WITH ( 'connector' = 'upsert-kafka','topic' = 'T2', ...)
create table enrich (...) WITH ( 'connector' = 'upsert-kafka','topic' = 'enrich', ...)

insert into enrich 
select ... from T1 inner join T2 ... T2.t1_id = t1.id and t2.client_id=t1.client_id

Note: • Data is is evenly distributed in partitions of raw topics (T1,T2..) • kafka partitions are 8 on raw topics and parallelism is set 8 on flink job and taskSlot=1 Question1: Where can i see key groups assignment to tasks (i have 1 taskSlot) Question2: Where can i see keys(data) to keygroups mapping in my running application? Question3: Columns used in join(like

t1.client_id,t1.id

) become the key and shuffle to land to same keygroup ?? or columns defined in

primary key

also used and trigger shuffle in source operators ?? reference: https://www.slideshare.net/FlinkForward/evening-out-the-uneven-dealing-with-skew-in-flink-252485368

The skew will be apparent in various metrics. E.g., you can look at numRecordsIn and numRecordsInPerSecond for each of the relevant instances. Comparing the checkpoint sizes for different instances can also be helpful.

Thanks David. i was wondering if KeyGroup with in TaskSlot could cause backpressure/performance impact internally. In above example, there are 16(128/8) KeyGroups for each TaskSlot. If one of the key group(with in same taskSlot that we don't have visibility) is hot but net number of records in TaskSlot (that appears under flink UI as subtask) is balanced does it impact performance or cause back-pressure ??

In the scenario you've outlined above, the situation is pretty well balanced, and there isn't enough skew to worry about. There's no performance impact resulting from having all of the records in the slot come from the same keyGroup (e.g., slot 2 above). However, if you were to scale up from 8 slots the situation would become more and more unbalanced as the number of keyGroups per slot declines. In general you want to keep the number of keygroups per slot high enough so that you avoid significant imbalances.

As usual great insight David. I was under impression keygroup not only used for

key > keygroup >taskSlot

assignment but also used internally to partition the data in rocksdb. Imagining the number that i presented is in million, i was wondering if it cause bottleneck somewhere at backend hence was looking if is there any metrics that can help me find hot key group. Besides, In order to find the optimal parallelism/max parallelism that is future proof.. i have about 3000 clients data (about 100-300 billions records) that eventually will be coming on to the stream with gradual onboarding. which means, parallelism and max parallelism will change over period of time. When it changes, i believe job will take longer to start as it will establish the state from checkpoint and distribute the data as per new numbers?? Does it helps if increase the parallelism and max parallelism both to keep the same ratio for example (parallelism 8-> 16, max parallelism 128 -> 256) On the other side i was thinking, since i have all the raw data in snowflake, i create a function that wraps the keyGroup logic and returns keygroupId for each record that i can group to find data that each group and slot should expect to receive. In this approach, as my keys are composite, should i use

Objects.hash(key1,key2..)

in the function to find the hash of the key or should i use hashCode implementation from Tuple2.java

Object.hash()

internally using

Arrays.hashCode(values)

that is initializing

result=1 instead 0

and makes the difference.

public static int hashCode(Object a[]) {
        if (a == null)
            return 0;

        int result = 1; // (this is the difference that results different hash)

        for (Object element : a)
            result = 31 * result + (element == null ? 0 : element.hashCode());

        return result;
    }

Changing the max parallelism (or in other words, changing the number of keygroups) is potentially painful, since this will break any checkpoints or savepoints you may have. It's generally best to establish a large enough max parallelism from the very beginning. With RocksDB, each state descriptor describes a column family. The RocksDB key is the serialized bytes of <Keygroup, Key, Namespace>. This allows for iterating separately over each keygroup for checkpointing, using native RocksDB iterators. The default max parallelism is 128 because increasing this to 32768 (the upper limit that Flink allows) causes a 10-15% performance degradation for the heap-based state backend. I'm not aware of any reason not to go ahead and set the max parallelism to 32768 with RocksDB.

..........................0.