https://flink.apache.org/ logo
Join Slack
Powered by
Hello all. I need help diagnosing why a flink sql...
# troubleshooting
j
Hello all. I need help diagnosing why a flink sql job with only 3 tables has only read 30 million records out from the main kafka topic, which contains 170million records. As time is progressing, it is getting slower and slower. We have checkpointing on and are using rocksdb. Could I be pointed to documentation/blog post/anything that will shake my noodle a bit so I can help my team solve this problem? Much appreciated.
s
Do you have any stateful logic in your topology, joins, aggregations, etc.?
j
yes we do. Even some distincts. Our client is trying to replicate code from oracle/oracle-bi, many of which require some crazy amounts of joins and some minimum number of aggregations. It's really the ones that require joins that are slowing down the most. We are using ververica with HA turned on as well.
working with the client, i've already removed the need for a handful of joins through out the project, but there are still some that they think they still need. I'm having them go back and validate that 100% they are needed. Either way, 1 usecase is 3 tables inner joining. There must be something I can do to allow this thing to do more than 6k records per min, running with parallization of 12. My guess is that it has something to do with shuffeling as well. I just don't know enough to know where to look to diagnose for sure.
s
If these joins are not temporal (e.g. not interval), the state they use will keep growing infinitely. Quick explainer with some links: https://docs.immerok.cloud/docs/how-to-guides/development/joining-and-deduplicating-data/#deduplicating-and-joining-the-events
If itโ€™s your case I gave a talk that explains non-temporal joins a bit more

https://www.youtube.com/watch?v=tiGxEGPyqCg&ab_channel=FlinkForwardโ–พ

j
awesome, thank you!
I've avoided ttl because we need data accuracy. My understanding is that if a record isn't used for X amount of time, it'll be disgarded, which would be bad for this usecase.
๐Ÿ‘ 1
But I will check out your talk, thank you!
So, i haven't had a chance to complete the video yet, but in one instance I found we didn't need the joins. I met with the client and he agreed. But now this is interesting. After the first 8million records were processed in the first 3 mins, it slowed down to a crawl again. This is what I'm noticing in the sinkmaterilaizer. Not all tasks are writing to kafka? The source is kafka and we are sinking back to kafka. Is it possible Kafka is the bottleneck? We have 3 brokers running ๐Ÿค”
s
Yeah, itโ€™s quite possible. One thing you can do: replace your Kafka sink with the Blackhole sink and see if you still have the issues.
j
ah, good point! will do, thank you
s
Also, the sinkmaterilaizer operator can be quite slow and it uses a lot of state. In certain situations itโ€™s safe to disable it with 
table.exec.sink.upsert-materialize: NONE
. Check more here.
j
By default, the materialize operator will be added when a distributed disorder occurs on unique keys.
In the case where everything is partitioned by the key and there is no joining, i'm assuming there wouldn't be a disorder of the keys?
s
I guess so, Iโ€™d still test it extensively
๐Ÿ™ 1
j
will do, still very grateful for this insight ๐Ÿ™‚
๐Ÿ‘ 1
sinkmaterializer...that was it. Even with blackhole it read in 10mil records in 1 min. At 4mins it was at 10.2mil. Turning off sinkmaterilizer it completed all 170mil (still to blackhole) in 4mins.
Just finished watching. Thank you ๐Ÿ™‚ Very helpful
๐Ÿ™Œ 1
Open in Slack
PreviousNext
Powered by