This message was deleted.

What data format are you using, and what is your observation in terms of ingested MB / second per task ? I’ve done a few large scale test experiments and I noticed ai can get about 2X troughput per task if I use protobuf instead of json

Mostly json and avro. Great point on trying out protobuf.

Been also playing with some Kafka consumer buffer settings but it did not seem to move the needle much

Also this was not obvious to me at the beginning but it seems like for best results you need to match number of tasks to number of Kafka partitions

Yeah, number of Kafka partitions is the upper bound. We had to resort to increasing our Kafka partitions because we were maxed out on number of ingestion tasks.

a performance analysis workflow i use: 1) is CPU usage "high enough" that you're making good use of your hw? Note that you don't want it at 100%, since then you have no room to recover from failures or run queries or handle spikes. But you don't want it at 5–10% either, that's too low. Probably 50–80% range is healthy depending on your data patterns. 2) if CPU usage is not high enough, figure out why. you might need to adjust configs to use more CPUs. or you might be blocked on something that isn't CPU, like disk I/O or network. 3) OTOH, if CPU usage is "high enough" next step is figure out what CPU being used on. here, profiling is the best tool. use your favorite java profiling tool; or, if you don't have one, you can take flame graphs manually: https://support.imply.io/hc/en-us/articles/360033747953-Profiling-Druid-queries-using-flame-graphs. a brief plug for imply 🙂 — if you're working with us, we have a continuous profiling tool we can enable for your cluster that lets you look at aggregate profiling info across the fleet or drill down into individual servers/tasks/datasources/etc. very helpful for finding bottlenecks and is part of what we do when offering https://imply.io/total-value-guarantee 4) once you find what CPU is being spent on, you can take actions to optimize. for example, if it's mainly in JSON parsing, then you can consider switching to a diff format. you can also ask around here for further input

We're getting around 20-30k events per second per thread. I can tell you what tuning parameters we used but I think the main bottleneck is disk I/O.

We are hitting 90% on CPU, so will try out non-json format. For persisting intermediate segments, we write tmp files to off heap memory. We also have set numPendingPersists to 3 so that ingestion won’t pause when tmp persists are happening.

4-8 million events per second

How many peons is that?

The main issue with throughput I’m finding is segments being handed off. If we hit 5 million rows for a single segment(maxrowspersegment), all on disk segments are merged and handed off to deep storage (appenderator-merge). Whilst this happens, ingest is paused for that task. I’m trying to find a way to force all merging/handing off to occur at the end of the task as I think that’s the key to ingest performance.

Frankly, I disagree; handoff is super important with extremely large datasets. I would rather pause in handoff than crash due to OOM. However, I agree that this process could be improved as it's exactly those high demand streams where handoffs will be occurring and the current levers we have for configuration (e.g. maxPendingPersists) don't seem to help any (at least in my testing).

maxPendingPersists

helps in controlling the amount of memory being used by ingestion tasks. It is mostly helpful BEFORE the handoff stage. Without

maxPendingPersists

configured to a reasonable value, we would see ingestion throttle. @JRob - we were able to get to 4-8 million events per second using around 600 ingestion tasks. So roughly around 7500-10000 events per ingestion task per second. Could you tell more about the tuning parameters you have used?

@Samarth Jain We are achieving 20-30K events per second per task. We are also still using middleManager as we are able to get the performance we need thus far. This tuning works for VMs sized at 16 cores and 64 GB of RAM. We have set -Xms128m and -Xmx128m in jvm.config. Other important tuning parameters from runtime.properties:

druid.worker.capacity = 12
druid.indexer.runner.javaOpts=-server -Xms1g -Xmx2g -XX:MaxDirectMemorySize=3g -Duser.timezone=UTC -Dfile.encoding=UTF-8 -XX:+ExitOnOutOfMemoryError -Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager

# Processing threads and buffers on Peons
druid.indexer.fork.property.druid.processing.numMergeBuffers=2
druid.indexer.fork.property.druid.processing.buffer.sizeBytes=268435456
druid.indexer.fork.property.druid.processing.numThreads=1

# Group By Query Tuning
druid.query.groupBy.maxMergingDictionarySize=1073741824
druid.query.groupBy.maxOnDiskStorage=107374182400

Notes: • We set

druid.worker.capacity

slightly under # of cores to account for historicals running on same nodes • We tested the peon memory allocation significantly and finally settled on 1 GB of Heap (expandable to 2 GB) and 3 GB of DirectMemory. Druid needs a lot of Direct Memory. • I forgot how we settled on a merge buffer of 256 MB but it was a factor # of workers and available memory • We allocated 1 GB to the merging dictionary for queries (we have some might large queries) • Disk storage of 100 GB is sort of a catch-all for overflow; I would rather slow down a query than halt it altogether. Works of us but might need tuning if you have low disk space. Our ingestion tasks typically use all of the default settings. We typically use HOURLY rollup. I only have one topic that needs more than one task; there I have it set to 10 tasks in order to ingest the nearly 300K events/second that topic generates.

If you have spikes in Kafka throughput, you can use Kafka auto-scaler to optimize resource utilization. The Kafka autoscaler is still experementaal feature. More details here: https://imply.io/blog/auto-scaling-real-time-kafka-ingestion-ftw/