ingesting with high cardinality dimension have low performance

[quenlang]

Hello, @himanshug

I met a huge performance when I ingested with high cardinality dimensions. The uri_id and host_ip have a cardinality range 2000000000 to 5000000000.

With the above two dimensions, 1 task consumed 2w/s. When I incremented task count to 18, the datasource only consumed 20w/s. Even though I incremented task count to 36, the datasource ingestion throughput always keeps 20w/s.
one of the tasks log:
index_kafka_APP_NETWORK_DATA_MIN_JSON1_f25e9db0956e624_bebmfiod.txt

Then I removed uri_id and host_ip from the dimension list, 18 tasks consumed 37w/s, seem as 2w * 18=36w. I incremented task count to 36, throughput was 60w/s.
one of the tasks log:
index_kafka_APP_NETWORK_DATA_MIN_JSON1_361e588173127a7_peommncl.txt

I'm not understander that why increment task count but the throughput does not improve with high cardinality dimensions.

The num of Kafka topic partition is 36. If I partition data by hash(all dimensions) into topic partitions to reduce the dimensions cardinality which needs to be rolled in the task, does it improve the performance?

Is there any way to resolve this problem? Can you give me some advice?

[himanshug]

what is w ins w/s ? and yes, partitioning by hash(all dimensions) might help improve roll up and performance in general .

slightly related: you might have done this already but it would be a good idea to consider why those dimensions really need to be dimensions and could you use sketches instead if you just wanted cardinality and histograms on those.

[quenlang]

Thanks for the quick reply!

what is w ins w/s ?
2w/s means the ingestion throughput is 200000 per second.

I'm confused that the throughput does not improve with high cardinality dimensions when I increased the task count from 18 to 36. A single task can consume 200000/sec, I understand the peon task just work independent. So I increase the task count the throughput can be improved. But It does not work.
Can you show me why? Thanks a lot!

[himanshug]

I would expect the throughput increase when going from 18 to 36 tasks given that you have 36 partitions in kafka topic. I can't think of a good reason why throughput did not increase accordingly. Did you verify that supervisor was able to spin up all 36 tasks ?

[RestfulBlue]

did you hit your hardware limit? like cpu/disk/network?
Which hardware configuration do you have? How much cores you have configured in kafka? How much disks utilization do you have on your kafka broker? what types of disks is it?

[quenlang]

@himanshug @RestfulBlue Thanks a lot!
There is no hardware limit. I partitioned row data with all dimensions into the Kafka partitions instead of the random partition, the throughput was improved.

I have another problem is that when a big query performs on the historical node, does it need high level disk iops or disk bandwidth? or both ?

Thanks!

[himanshug]

does it need high level disk iops or disk bandwidth? or both ?

haven't done any real studies there but SSDs are better in general due to the random reads that Druid does and higher IOPS disks typically have high throughput as well that is usually true for SSDs. I would say "both".
(disk perf doesn't matter much if there is available memory to hold all the loaded segments in page cache all the time, because in that case most queries wouldn't touch the disk)

[quenlang]

@himanshug Thanks a lot!
I had upgraded my disk to SSDs.
Now I have 6 historical nodes with 64 core 256GB memory and 8*4TB NVMe SSDs.

I found the query performance was terrible. My cluster ingested 200GB data and per segment size was 700~800MB. When I performed 100 queries with 20 parallel thread，the max response time always 12~14s and the CPU utilization of every node nearly reached 100%.

When I performed 100 queries with 1 parallel thread，every query can be responded within 1000 ms and the CPU utilization of every node under 50%.

Why druid used so much CPU cores when performed concurrent queries? Is there any way to reduce CPU usage?

Druid uses one CPU core to handle per segment, suppose there are so many segments a query need to search, CPU easily becomes the bottleneck.

Historical node configuration:

druid.service=druid/historical
druid.port=8083

# HTTP server threads
druid.server.http.numThreads=50

# Processing threads and buffers
druid.processing.buffer.sizeBytes=535544320
druid.processing.numThreads=63
druid.processing.numMergeBuffers=4

druid.historical.cache.useCache=false
druid.historical.cache.populateCache=false
druid.historical.cache.unCacheable=["select"]
druid.cache.sizeInBytes=4073741824


# Segment storage
druid.segmentCache.locations=[{"path":"var/druid/segment-cache","maxSize":25000000000000}]
druid.server.maxSize=25000000000000

[himanshug]

@quenlang I hope you figured things out by now, but just in case...

it sounds like 1sec is the query time when things don't have to wait anywhere for resource as in the single threaded case.

for the parallel case, I guess 12~14s came from queries waiting in historical for processing thread pool (I would expect that be reported in query/wait/time metric at historical) .

at historical , query is processed by processing each segment (needed for this query) in parallel using the processing thread pool which is of limited size (63 in your case) , when you sent parallel queries all cores were utilized.

utilizing all the cores is actually a good thing :)

things can be tuned based on the use cases , try to check the query metrics ( https://druid.apache.org/docs/latest/operations/metrics.html ) and take a look at https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html if you haven't .

[stale]

This issue has been marked as stale due to 280 days of inactivity. It will be closed in 4 weeks if no further activity occurs. If this issue is still relevant, please simply write any comment. Even if closed, you can still revive the issue at any time or discuss it on the dev@druid.apache.org list. Thank you for your contributions.

[stale]

This issue has been closed due to lack of activity. If you think that is incorrect, or the issue requires additional review, you can revive the issue at any time.