Kafka + Spark scalability

Question

We have very simple Spark Streaming job (implemented in Java), which is:

• reading JSONs from Kafka via DirectStream (acks on Kafka messages are turned off)
• parsing the JSONs into POJO (using GSON - our messages are only ~300 bytes)
• map the POJO to tuple of key-value (value = object)
• reduceByKey (custom reduce function - always comparing 1 field - quality - from the objects and leaves the object instance with higher quality)
• store the result in the state (via mapWithState stores the object with highest quality per key)
• store the result to HDFS

The JSONs are generated with set of 1000 IDs (keys) and all the events are randomly distributed to Kafka topic partitions. This also means, that resulting set of objects is max 1000, as the job is storing only the object with highest quality for each ID.

We were running the performance tests on AWS EMR (m4.xlarge = 4 cores, 16 GB memory) with following parameters:

• number of executors = number of nodes (i.e. 1 executor per node)
• number of Kafka partitions = number of nodes (i.e. in our case also executors)
• batch size = 10 (s)
• sliding window = 20 (s)
• window size = 600 (s)
• block size = 2000 (ms)
• default Parallelism - tried different settings, however best results getting when the default parallelism is = number of nodes/executors

Kafka cluster contains just 1 broker, which is utilized to max ~30-40% during the peak load (we're pre-filling the data to topic and then independently executing the test). We have tried to increase the num.io.threads and num.network.threads, but without significant improvement.

The he results of performance tests (about 10 minutes of continuous load) were (YARN master and Driver nodes are on top of the node counts bellow):

• 2 nodes - able to process max. 150 000 events/s without any processing delay
• 5 nodes - 280 000 events/s => 25 % penalty if compared to expected "almost linear scalability"
• 10 nodes - 380 000 events/s => 50 % penalty if compared to expected "almost linear scalability"

The CPU utilization in case of 2 nodes was ~

We also played around other settings including: - testing low/high number of partitions - testing low/high/default value of defaultParallelism - testing with higher number of executors (i.e. divide the resources to e.g. 30 executors instead of 10) but the settings above were giving us the best results.

So - the question - is Kafka + Spark (almost) linearly scalable? If it should be scalable much better, than our tests shown - how it can be improved. Our goal is to support hundreds/thousands of Spark executors (i.e. scalability is crucial for us).

Answer 1

We have resolved this by:

• increasing the capacity of Kafka cluster
  • more CPU power - increased the number of nodes for Kafka (1 Kafka node per 2 spark exectur nodes seemed to be fine)
  • more brokers - basically 1 broker per executor gave us the best results
• setting proper default parallelism (number of cores in cluster * 2)
• ensuring all the nodes will have approx. the same amount of work
  • batch size/blockSize should be ~equal or a multiple of number of executors

At the end, we've been able to achieve 1 100 000 events/s processed by the spark cluster with 10 executor nodes. The tuning made also increased the performance on configurations with less nodes -> we have achieved practically linear scalability when scaling from 2 to 10 spark executor nodes (m4.xlarge on AWS).

At the beginning, CPU on Kafka node wasn't approaching to limits, however it was not able to respond to the demands of Spark executors.

Thnx for all suggestions, particulary for @ArturBiesiadowski, who suggested the Kafka cluster incorrect sizing.