Hello! I’m running PoC with Pinot for quite heavy data. I’ve got table with ~ 20 billions rows, 5 predicates (pred1 cardinality ~ 5 millions uniqs, pred2 and pred3 the same - tight correlation, last pred5 has very low cardinality, tens of values). I need to achieve the best possible speed for lookups by these predicates for whole range (20-50 billions/rows). Currently my table (realtime) is creating for these predicates bloom & inverted indices. Second problem is ingestion rate - apparently there is no problem to get ~ 160k/s documents which is insane in contrast to resources needed, but at the same time the query performance is very bad - 6 servers are pretty busy with ingesting and GC thus query is pretty bad, 20-50 seconds. My current setup is 6 servers, 1 controller. Split commit enabled to s3. Because there will be low QPS, I need to achieve low memory allocation for indices/segments. Do I need to consider some kind of bucketing/hidden partitioning for predicate values or is Pinot able to handle these data in SLA ~ 1000-3000 ms only with proper indexing? I can imagine some sort of work delegation for servers, e.g. consuming/segment creating ~ 3-4 servers and for querying allocate 6 servers. PS: I’ve got replication 1 for space saving as final total will be ~ 20 TB, segment size is currently 460MB (but in table is set to 1GB). Ingesting from 36 kafka partitions. Servers have spin drives. PS2: not sure about effect of bloom because is still in rebuild process… Any improvements, thoughts or tricks are welcomed! 🙂

first question: which version of pinot are you using?

0.9.3

to understand the slow queries, it would be good to get a profile during query execution - you can collect this by running

jcmd <realtime server pid> JFR.start duration=60s settings=profile filename=slow_realtime_query.jfr

I’ll capture flight profile later, but regarding to other questions: 1. ingestion - only flattening of json file (e.g. 8 filelds, via JSONPATH…, one field kept as “rich” JSON object in STRING see tests_table config attached 2. Cardinality is described in json - 4x predicate with ~ 6 millions unique values, 1 predicate with cca 60 unique values 3. Other responses/traces I’ll post when cluster will be ready. I’m suspecting that kubernetes cluster has additional issues.. Running on pretty strong bare metal, but old kernel, sometimes I see broken pipe and network connectivity issues between Pinot components

if you're doing jsonpath I recommend trying the latest docker images on 0.10.0-SNAPSHOT - I fixed a lot of performance problems in the jsonpath lib

Unfortunately all strings.. but I think that the mostly used predicated could be converted to long

using an onheap dictionary for pred5_cardinality_60 (and only pred5_cardinality_60) might be helpful to save on allocations

Ok, and what about inverted index? Is it even helpful here in combination with bloom? Or bloom would be enough for lookup?

it would be good to know what difference you get from the nightly docker images before reconfiguring anything

Just upgrading 🙂 should be running already

hard to say, need to see the response metadata and a profile really

at least per segment is uniform, with high selectivity

ok good, let's just wait for the data to see what's going wrong - your expectations are realistic - but some features are more mature than others

Is it OK to capture flight profile from 1 server (from 6)?

yes, it should be evenly split

Flight profile - search by mostly used predicate, response in 50 sec. Currently table is “only” 7 billions.. but without already mentioned improvements such as sort etc..

thanks

yes, it’s for different value but now with time 126 sec “exceptions”: [], “numServersQueried”: 6, “numServersResponded”: 6, “numSegmentsQueried”: 3094, “numSegmentsProcessed”: 1872, “numSegmentsMatched”: 109, “numConsumingSegmentsQueried”: 36, “numDocsScanned”: 162, “numEntriesScannedInFilter”: 0, “numEntriesScannedPostFilter”: 7614, “numGroupsLimitReached”: false, “totalDocs”: 7028875200, “timeUsedMs”: 126651, “offlineThreadCpuTimeNs”: 0, “realtimeThreadCpuTimeNs”: 0, “offlineSystemActivitiesCpuTimeNs”: 0, “realtimeSystemActivitiesCpuTimeNs”: 0, “offlineResponseSerializationCpuTimeNs”: 0, “realtimeResponseSerializationCpuTimeNs”: 0, “offlineTotalCpuTimeNs”: 0, “realtimeTotalCpuTimeNs”: 0, “segmentStatistics”: [], “traceInfo”: {}, “minConsumingFreshnessTimeMs”: 1645106201566, “numRowsResultSet”: 10

and the query (redact the params)?

select * from epc_pgw where a_party_imsi = ‘xxxx’ limit 10

virtually all the time is spent on processing JSON

Ok it means that ingestion is fully saturating all threads and there is no space to query, right?

GC looks fine, only two STW events but thy were under 10ms

@Pavel Stejskal Query is not doing much, it’s mostly bcos of resource contention between ingestion threads and query threads.. Pinot has the ability to isolate consuming servers and query servers..

For “select count(*)..” 73 seconds - with stopped ingestion. jfr attached “exceptions”: [], “numServersQueried”: 6, “numServersResponded”: 6, “numSegmentsQueried”: 3099, “numSegmentsProcessed”: 1858, “numSegmentsMatched”: 1765, “numConsumingSegmentsQueried”: 36, “numDocsScanned”: 2742, “numEntriesScannedInFilter”: 0, “numEntriesScannedPostFilter”: 0, “numGroupsLimitReached”: false, “totalDocs”: 7031042881, “timeUsedMs”: 73528, “offlineThreadCpuTimeNs”: 0, “realtimeThreadCpuTimeNs”: 0, “offlineSystemActivitiesCpuTimeNs”: 0, “realtimeSystemActivitiesCpuTimeNs”: 0, “offlineResponseSerializationCpuTimeNs”: 0, “realtimeResponseSerializationCpuTimeNs”: 0, “offlineTotalCpuTimeNs”: 0, “realtimeTotalCpuTimeNs”: 0, “segmentStatistics”: [], “traceInfo”: {}, “minConsumingFreshnessTimeMs”: 1645106324027, “numRowsResultSet”: 1

it looks like the servers are just saturated by ingestion

~ 3000 ms. I’ll rebuild segments according your suggestions

oh, I don't mean query latency, but end to end latency from event creation to showing up in queries, how fresh does the data need to be?

near real-time if possible. I’d say 30 seconds

You don’t have to rebuild segments

It will reorder (sort by co1) automatically?

I don’t think you need it right now.. let’s validate the separation of consuming from non consuming nodes..

I’ve got additional 6 servers, how to tag them for “query” only?

thanks Peter 🙂 I was searching for that and could not find it..

One question.. segment itself does contain any index? I see during rebalancing a lot of messages that new server is creating bloom index for just downloaded segment PS: could be only for old segments because I changed indexes later ..

Segments in deep store does not have any indexes .. they get created when they are downloaded for the first time

And one question to controller.. I think there could be potential bottleneck as well, because responses are pretty slow. I don’t have a dedicated disk for controller and I’m not sure if there is some high I/O traffic. E.g. loading of tables (meta) takes around 15 seconds…

can you take a profile of the controller? It can indeed be a bottleneck for segment completion

Sure. I'll post details tomorrow when rebalancing will be done

hi - is it still performing well?

Yes, 220 millions rows and by 1 predicate results ~ 14 ms, always under 20ms. Almost the same as for empty table. There was something weird before. Anyway, helm chart with tag latest could possibly messed my deployment, because when I did a server exapansion, latest was possibly different version and some hosts were downloading images.. I’m not sure if all my images was the same 😞 should be explicitly stated in helm chart

hmm, it would be good to get to the bottom of whatever happened, keep us posted if you see the same thing again

Hello, I’m experiencing the same issue with latency again. Around 1 billion rows everything seems to be fine, p95 ~ 60ms. Close to 2 billions I’m getting more and more timeouts (10sec). jfr file attached witch multiple queries… Running with: 6 servers, each has own spinning 1 drive 8TB, each 48GB memory. The latency is stable till Memory Mapped Usage is aroud 45 GB per server, then is getting worse. Disk has decent performance but is it ordinary spinning drive (120 mb/s, read 400 iops, 120 write iops). Table is currently ~ 1.1 TB (2232 segments).

could be page faults

No, I don’t have an access to hosts directly, just to kubernetes

ok so there's no way you can do kernel level configuration?

Yes I can, but I have to describe what to do to my colleague which has access

ok, that sounds painful, let's figure out if it would be beneficial first

We’re running on latest RHEL but it has kernel 4.18 or something…

we run this to get perf

apt-get update && \
     apt-get install -y --no-install-recommends linux-base linux-perf openjdk-11-dbg

Anyway, is there some limit for MMAP regarding to size of data in total? Per server

we install async-profiler like this

ARG TARGETPLATFORM
RUN case $TARGETPLATFORM in \
       linux/amd64) arch=x64; ;; \
       linux/arm64) arch=arm64; ;; \
       *) echo "platform=$TARGETPLATFORM un-supported, exit ..."; exit 1; ;; \
     esac \
     && mkdir -p /usr/local/lib/async-profiler \
     && curl -L <https://github.com/jvm-profiling-tools/async-profiler/releases/download/v2.5.1/async-profiler-2.5.1-linux-${arch}.tar.gz> | tar -xz --strip-components 1 -C /usr/local/lib/async-profiler \
     && ln -s /usr/local/lib/async-profiler/profiler.sh /usr/local/bin/async-profiler

Ok let me understand. If I got server with 48 gigs of Ram, how many segments can this sever hosts in gigs? Rough estimation. To be performing well

I'd break it down differently

Current setup for server: -Xms1G -Xmx16G -XX:+UseG1GC -XX:MaxGCPauseMillis=200, container has limit 48GB

what's the segment size in MB?

250MB

my ex-linkedin colleagues recommended slightly more server memory for a heap of that size based on their production experience - 64GB

so xmx=64? Is there some penalty getting over 32 GB (compression…)

no I mean xmx=16, server has 64GB

sure 🙂 becauase I’m struggling with basics like this..

@Mayank may be able to help later

I’ll contact my colleague to install profiler and check hugepage in kernel

you need to set

-XX:+UseHugeTLBFS

to make use of it

Yes I’d like to understand the limits and bottlenecks as well. But regarding to hosted segments and total size, I still don’t understand what i Pinot capable to handle. If there is limit just disk space, or some special mapping between system memory to segments via MMAP… For example my case, server with xmx=16, 64 gigs ram, how many data it should be capable to handle?

There isn't really a simple formula and realtime and offline nodes behave differently. The time taken here is spent in conversion of the realtime segment to an offline segment and having a faster disk or more memory would certainly help with this operation, because it's doing things like translating data into more efficient formats for faster queries as the offline segment will exist for a long time. It's writing to disk, doing compute-heavy operations, buffering. So it's more a question of having the right resources for this particular operation than what the right number of segments is.

Yep got it. The sorting is for the mostly used predicate, should be converted to long later. Otherwise whole input is json therefore for projection is there few big structures. All columns expect 5 predicates are noDictorinary.. without indexes

depending on your latency requirements, you could do 1TB per offline node with 16 cores 64GB, very roughly speaking

so basically same ram footprint as data size for offline low latency?

my response on size was how much data can fit in RAM on the realtime servers with 48GB

Latency p95 ~ 2000-3000 would be fine. Thanks for estimation. I'm ready to design infra exactly for Pinot. We have 396gb ram available per server but I'm not sure about disk drives, how big issue would be spinning drives. If spinning is no go, we can adapt drives as well after PoC.

I don't think same ram as data for low latency has been suggested, just how much data is served from consuming real time servers.

Yes I have. My current state: table with 4,9 billions rows, size 1.85 TB (reported size), 3806 segments. Spread evenly on 6 servers: Server_pinot-server-3.pinot-server-headless.pinot.svc.cluster.local_8098 624 Server_pinot-server-4.pinot-server-headless.pinot.svc.cluster.local_8098 609 Server_pinot-server-5.pinot-server-headless.pinot.svc.cluster.local_8098 665 Server_pinot-server-0.pinot-server-headless.pinot.svc.cluster.local_8098 644 Server_pinot-server-1.pinot-server-headless.pinot.svc.cluster.local_8098 619 Server_pinot-server-2.pinot-server-headless.pinot.svc.cluster.local_8098 645 My query: select * from epc_pgw where a_party_msisdn=‘xxx’ <-- predicate cardinality ~5millions, sorted by this and has bloom + inverted index limit 10 Response: “exceptions”: [], “numServersQueried”: 6, “numServersResponded”: 6, “numSegmentsQueried”: 3806, “numSegmentsProcessed”: 1492, “numSegmentsMatched”: 112, “numConsumingSegmentsQueried”: 6, “numDocsScanned”: 137, “numEntriesScannedInFilter”: 0, “numEntriesScannedPostFilter”: 6439, “numGroupsLimitReached”: false, “totalDocs”: 4902830003, “timeUsedMs”: 19090, “offlineThreadCpuTimeNs”: 0, “realtimeThreadCpuTimeNs”: 0, “offlineSystemActivitiesCpuTimeNs”: 0, “realtimeSystemActivitiesCpuTimeNs”: 0, “offlineResponseSerializationCpuTimeNs”: 0, “realtimeResponseSerializationCpuTimeNs”: 0, “offlineTotalCpuTimeNs”: 0, “realtimeTotalCpuTimeNs”: 0, “segmentStatistics”: [], “traceInfo”: {}, “minConsumingFreshnessTimeMs”: 1645359464059, “numRowsResultSet”: 10

50% of the time is spent decompressing raw string columns, 30% pruning segments via bloom filters, 15% looking up dictionary codes of string columns.

Thanks! Final projection will be much more narrow. But whats is weird is inconsistency it response time. Sometime I get timeout, sometime response in 3 seconds. Seems like there is some cache… but this could be related to disks as well. Till now all directs to SSD and conversion to numeric as much as possible 👍

Can you remove bloom filter?