I have a flink python job successfully running. I'...
# troubleshootingn
Nathanael England
08/10/2023, 4:37 AMI have a flink python job successfully running. I'm not happy with the way things appear to be scaling though. I have a keyed co process function that is the bottleneck. One of my parallel handlers is seeing only 9 records per second on average. I know processing those records takes typically < 0.5 milliseconds and (rarely) at most 1-2 milliseconds. I can see that my handler is registering > 60 ms per second of "busy" time. It doesn't seem like this time is anything my application is doing. I would expect more on the order of 1-2 milliseconds busy time for this record rate.
Where can I find where all this extra busy time is being consumed by? I should only have < 1000 keys at this point.
Nathanael England
08/10/2023, 4:38 AMI also feel confident my task manager pods are not resource-constrained. Looking at prometheus metrics, my pods have plenty of headroom over the provided cores and CPU utilization is very low at ~100 to ~200 mcores.
d
David Anderson
08/10/2023, 4:40 AMThe flink UI has built-in flamegraphs you can use to see where the time is going. The answer is usually serialization. (The rebalance you’re doing is forcing serde overhead that you could perhaps avoid.)
n
Nathanael England
08/10/2023, 4:42 AMAh. I will enable that feature and take a look. As for rebalancing, that seems to have happened automatically with how I wrote my job. I never called
.rebalance()d
David Anderson
08/10/2023, 4:42 AMIf you are manually adjusting the parallelism on specific operators, that’s implemented as a rebalance.
n
Nathanael England
08/10/2023, 4:43 AMYup, that is what happened here. The other parts of the stream and (mostly) stateless and very fast. This co process function is a bit heavier and I wanted to bump that up to distribute the load. Then that processing output is just dumped to rabbitmq
d
David Anderson
08/10/2023, 4:45 AMThat sort of tuning can be counterproductive. It’s worth experimenting to see how the tradeoffs play out for your specific use case.
David Anderson
08/10/2023, 4:48 AMYou could also experiment with optimizing/changing the serializers (which can be worth up to a 10x speedup) and maybe try turning on object reuse (in combination with operator chaining), though that’s usually only a 5-15% speedup.
n
Nathanael England
08/10/2023, 5:10 AMNothing immediately obvious to my untrained eye
Nathanael England
08/10/2023, 5:10 AMk
Keyur Makwana
08/10/2023, 6:45 AM@David Anderson @Nathanael England Same with me I have my job with Java and using Table API,
By chance if you can help me here is the flame-graph
✅ 1
Keyur Makwana
08/10/2023, 7:01 AMd
David Anderson
08/10/2023, 9:41 PM@Nathanael England Looks like flamegraphs aren't useful for inspecting pyflink jobs.
David Anderson
08/10/2023, 9:43 PM@Keyur Makwana Looks like deserialization.
n
Nathanael England
08/11/2023, 12:35 AMAs an update on this, I believe my bottleneck is interactions with flink-managed state. If I don't interact with state at all in my process function, my busy time drops to zero while still handling a mild number of records per second.
Nathanael England
08/11/2023, 12:36 AMI first tried a no-op
process_elementNathanael England
08/11/2023, 12:36 AMHowever, when I do something that does require state interaction, my processing time blows back up. This is despite strongly declaring types throughout the job so that the use of slower pickle operations is avoided.
k
Keyur Makwana
08/11/2023, 3:32 AMOhh yes, Thanks @David Anderson
I have investigated and found I was fetching Kafka property each time on the Kafka function. By optimising sink I am now getting almost 8x faster speed
🚀 1
7 Views