Apache Druid

I have a cluster with historicals getting to 100% disk space and failing on `No space left on device` at `org.apache.druid.utils.CompressionUtils.unzip` . Is there a better way to handle it rather than clearing up the entire cache? I hope I can clear partially written segments (if theres a good way to detect these).

Do you have <https://druid.apache.org/docs/latest/operations/clean-metadata-store.html#segment-records-and-segments-in-deep-storage-kill-task|automated cleanup> enabled?  (I think it might be on by default lately, but don't remember for sure.)

You can shut down, clear the cache, restart.  But also make sure you're not just growing and need more historicals or disk space, so it doesn't just repeat.

`druid.coordinator.kill.on` needs to be true.
To clean the directory after that setting is turned on, you can actually just delete the unwanted folders under segment-cache folder and then roll restart historical one by one.

if druid is not clearing the old files from historicals, then it seems like a bug. Historicals also shouldn't be loading more than the size of configured segment cache. So if you see a historical not honoring that setting, it could be some bug.

you should file a GH issue if you suspect the same

Is that config from the druid operator? I'm not very familiar with it, but it seems to me that you are creating 12 middle manager pods and that each one will spawn up to 12 peons (druid_worker_capacity). The middle manager itself doesn't need a lot of memory, so you could reduce that. Each Peon within each MM pod will be using 5GB for heap and up to 10GB of direct memory (druid_indexer_running_javaOpts). So your pods should be configured for at least 13 cpus and memory 13*15GB =  195 GB.  If your pod request limits are smaller it is likely that it will cause 137s.

It is from druid helm chart not the operator. Where did the 13 came from? 15GB I understand that is heap plus direct memory

however resource limits for MM looks like:
```resources:                                                                                                                                                                                                                                    
       limits:                                                                                                                                                                                                                                     
         cpu: "32"                                                                                                                                                                                                                                 
         memory: 48Gi                                                                                                                                                                                                                              
       requests:                                                                                                                                                                                                                                   
         cpu: "32"                                                                                                                                                                                                                                 
         memory: 48Gi```
so that might be the problem

the instance types I am using for MM are `c5d.9xlarge` however I think I might need to use `i3.8xlarge` instead as they offer up tp 244GB ram in comparison to 72GB for the c5d

Yeah, bigger nodes would help or you can reduce the memory settings and test your ingestion workload to see how much you can do with the c5d.9xlarge nodes. It is 13 because I am counting the MM JVM as well as the 12 worker JVMs. You can probably also reduce the settings for the MM heap and directmemory.

the problem I guess too are realtime queries going on together with realtime ingestion, so that memory consumption will increase

I'm trying to understand all the memory footprint of real-time ingestion tasks as it relates to their ingestion and query activities. On the query side, in principle, you have a controlled set of threads : `druid.processing.numThreads` that each use a set of merge buffers `druid.processing.numMergeBuffers` of size `druid.processing.buffer.sizeBytes`  and all this is off-heap memory. The HTTP threads for the task are used to receive requests and functions as a queue, the requests are processed from this queue using  processing threads to resolve individual queries. I'm still not sure about what this activity does on heap, but I'm digging deeper :wink:

all right, thanks. Yeah the whole thing is why I started getting tasks killed with code 137 and all I thought of was probably memory allocation

however what you described with regards to params, seems to be more related to historical nodes, and I would relate that to historical type of queries. Tasks being exited with 137 code belong to middleManagers I think

and I think the problem is with MM because of resource usage across all MM, which seems to be over 90% in all 12 MM I have, sometimes 99% in some of them

Yes, The query side of the ingestion tasks is similar to what Historicals do. 

But it works on the intermediate persists and the in-memory row ingestion buffer instead of working on a set of segments.

How big are a your ingestion? Messages/second ? How many queries/second on real-time?

I would say over 90k messages per second ingested, in spikes perhaps over 120k. Whereas queries received, over last month roughly 200k queries

Screenshot 2023-03-02 at 16.35.13.png

Not sure if this count if what you look for

I'm just trying to get a sense of the volumes and concurrency. Have you seen <https://imply.io/blog/exactly-once-streaming-ingestion/|this doc>? It has a section "A few numbers" on results from running different message schemas and the ingestion rate they achieved per task. It can serve as a guide to what you might expect from your ingestion.

How did it go? What settings did you end up with? Were you able to test up to 120k messages per second?  Given that you have swings in the message rate, it may also make sense for you to look at the<https://druid.apache.org/docs/latest/development/extensions-core/kafka-supervisor-reference.html#lag-based-autoscaler-strategy-related-properties| lag based autoscaler> which will increase and decrease the task count dynamically based on the consumption lag.

With the instance change to `i3.8xlarge` the problem got resolved, not more failed tasks with 137 and mem usage seems normal now.

We are using Kinesis, is that autoscaler applicable to kinesis ingestion too? I was thinking in a similar approach but by monitoring lag metrics emitted to statsd and via datadog

<https://druid.apache.org/docs/latest/development/extensions-core/kinesis-ingestion.html#lag-based-autoscaler-strategy-related-properties|Kinesis ingestion >also supports it. One thing to consider is that you will need to have enough middle manager/indexer worker slots available. An interesting test for this would be to also use `kubernetes-overlord-extensions` to run without middle managers on k8s. Not sure how that will behave because it is still experimental, but it promises to provide more elasticity with workers since it spawns workers as k8s jobs, no MM, the Peons are spawned as pods. Not sure how adventurous you are feeling :smile:

I might test that out in the dev cluster tho and see how it goes. For a production environment tho, it might be risky if it’s still experimental

i dont know about this JetS3t, but you prob need to provide extra context…
• what’s druid version? 
• are you deploying with druid operator? which version? other k8s deployment?
• can you show how are you trying to mount the config map in the deployment?

you will still need to make it available on all the nodes. the extension has to be available inside the extensions directory.

not sure how to test if it is emitting something

I have this
```druid_extensions_loadList: '["druid-histogram","druid-datasketches","postgresql-metadata-storage","druid-kinesis-indexing-service","druid-parquet-extensions","druid-basic-security","druid-s3-extensions","druid-kafka-indexing-service","statsd-emitter","druid-avro-extensions","druid-lookups-cached-global"]',
                    druid_emitter_statsd_dogstatsdConstantTags: pulumi.interpolate`["cluster:${clusterName}"]`,
                    druid_emitter_statsd_hostname: 'localhost',
                    druid_emitter: 'statsd',
                    druid_emitter_logging_logLevel: 'debug',
                    druid_monitoring_emissionPeriod: 'PT15S',
                    druid_emitter_statsd_port: '8125',
                    druid_emitter_statsd_dogstatsd: 'true',
                    druid_emitter_statsd_includeHost: 'true',```

question here tho is `druid_emitter_statsd_hostname: 'localhost',` since the node has different pods running in it, one is druid and another is datadog with the agent, so  probably that cant be `localhost` but the IP of the DD pod?

This is depends on how DD agent is installed. Is this on Kubernetes?
DD agent is typically installed on the host of Kubernetes itself, so you have to pass the Host ip address into your pod via environment variable.

And then on boot, you read that environment variable and pass it to the config file.

Yes, DD is on kubernetes, each node has different pods running in it, Druid, DD agent, …  Druid prod has different IP than DD agent pod, so hostname cant be `localhost` , it would need to be DD’s agent pod IP I guess.

Do you mean it can be the node IP itself? some test I have been doing indicates that needs to be DD’s agent pod IP, if I do this from Druid pod/container it works:
```echo -n "custom_metric_druid_dd_agent_pod_ip_prod:60|g|#shell" | nc -u -w0 10.5.128.196 8125```
`10.5.128.196` is the pod where DD agent is running

oneslot.png

onetask.png

This is not because of SLOTS not available, If slots are not available the jobs will be in PENDING state, Here the state WAITING should be for other reason, could you check if it has any message printed in overlord/inside TASK log for what it is waiting for.

The jobs are running on same data source and if the running job is locking the time interval then the other jobs which need locking on the same interval will be in WAITING state till the time it gets the lock so that it can run further

Yes, it is the same case. All tasks are for same data source and same time interval or granularity in ingestion spec.

correct so jobs need to wait till it gets the lock on the timechunk

you should run a single task and specify multiple sub tasks. this will ensure that you don't get into lock issues

Is there any way to parallelize this? because i have huge number of different files for which I initiate different batch tasks..

<@U03BJAMQ4H5>  Do you mean increase maxNumConcurrentSubTasks": 1,  in tuning config? afaik, this would only help if I have a large file. but I have multiple (in thousands) small individual files....

<@U03BJAMQ4H5> {
  "type": "index_parallel",
  "spec": {
    "ioConfig": {
      "type": "index_parallel",
      "appendToExisting": true,
      "inputSource": {
        "type": "s3",
        "uris": ["xxxxxxx/some/object"],
        "properties": {
          "accessKeyId": "xxxxxxx",
          "secretAccessKey": "xxxxxxxx"
        }
      },
      "inputFormat": {
        "type": "csv",
        "columns": ["col1, col2"]
      }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "maxNumConcurrentSubTasks": 1,
      "partitionsSpec" :{
        "type":"dynamic",
        "maxRowsPerSegment": 7000000
      }
    },
    "dataSchema": {
      "dataSource": "d1",
      "timestampSpec": {
        "column": "ts",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": [
          {
	          "type":"long",
	          "name":"col1"
	        },
          {
            "type": "string",
            "name": "col2",
            "createBitmapIndex":true
          },

        ]
      },
      "granularitySpec": {
          "segmentGranularity": "DAY",
          "queryGranularity": "none",
          "intervals" : [ "2022-09-14T00:00:01.000000Z/2022-09-14T23:59:59.000000Z" ],
          "rollup":false
      }
    }
  }
}

Use a prefix instead of uri to recursively fetch from a folder and set maxnumconcurrentsubtasks to 4

yes <@U03BJAMQ4H5> we are trying for real time with high ingestion

also we are handling late data till 5 days

With real-time ingestion it will take sometime for the existing tasks to finish publishing and only then will the new spec take effect. Have the current tasks shutdown and new tasks started?

yes i suspended the datasource till all the realtime segments get pushed and then resumed the ingestion

Check the payload of the tasks.  Does it reflect the changes you made?

<@U04B91RG747> with realtime ingestion, segments will be published at the `taskDuration` specified in the config, or according to the conditions described in `intermediateHandoffPeriod`
<https://druid.apache.org/docs/latest/development/extensions-core/kafka-supervisor-reference.html#kafkasupervisorioconfig>

The default taskDuration is 1H, so I suspect what you are is seeing is DAY granularity segments being published every hour

to get DAY granularity segments, you can configure auto-compaction on the datasource to come in and pack the segments more densely

<@U02VCGJFER5> i have changed the task duration to 1 day,
the difference that is see in segment sections(router console) for that particular data source has start and end as 1 hour duration  for realtime segment of late data and current day realtime segment it has of 6 hour duration

can you paste a redacted version of your spec and a screenshots of the segments tab in the web console. It will be easier to tell what is going on if we can see the config

Hey <@U02VCGJFER5> <@U03BJAMQ4H5>,
Thanks for the input, after stopping the ingestion and doing the compaction on that segment for the day, i was remove that issue and now segments get created normally for day as expected

I haven't tried your case specifically .. have you tried different versions of syntax that accomplish the same thing, such as:

```having SUM("mt_attach_attempts")&lt;&gt;0```
?

Well, I tried `HAVING "kpi_case1" &lt;&gt; 0` and it works ; but also filters results with 0 (not only NULL)

yesterday I was re-reading druid expressions and noticed this one. Try it dropping the HAVING

```safe_divide(SUM("mt_attach_failures")*1.0, SUM("mt_attach_attempts"))```

```safe_divide(x,y) returns the division of x by y if y is not equal to 0. In case y is 0 it returns 0 or null if druid.generic.useDefaultValueForNull=false```
<https://druid.apache.org/docs/latest/misc/math-expr.html>

maybe it wont work as it’s an aggregation, but let’s see

Daniel, your safe_divide() idea works for me (against the wikipedia demo dataset), although I still have to use it within a Having clause, e.g.:

```having safe_divide(SUM("mt_attach_failures")*1.0, SUM("mt_attach_attempts")) is not null```
Oliver, a correction to my having clause (I can't update my post above), would check both numerator and denominator, e.g.:

```having SUM("mt_attach_attempts")&lt;&gt;0 and SUM("mt_attach_failures")&lt;&gt;0```
that should logically check the same thing as your CASE statement.

Thanks a lot for your answers, that's great !
It won't work for me: I also need the formula to be somehow "standard SQL" (because it is used to query on another DB and safe_divide is not supported).
But I found a workaround for that: I replace NULL with 'NaN'  when sending the query to Druid.
The formula sent to Druid is now  `(Case When SUM("mt_attach_attempts")&lt;&gt;0 THEN (CAST(SUM("mt_attach_failures") AS DOUBLE)/CAST(SUM("mt_attach_attempts") AS DOUBLE)) ELSE 'NaN' END) AS "kpi_case1",` and I adapted the filter `HAVING "kpi_case1" IS NOT NULL AND CAST("kpi_case1" AS varchar) &lt;&gt; 'NaN'` . And now it works for me. :slightly_smiling_face:

another note — if you like standard SQL then you may like setting `druid.generic.useDefaultValueForNull = false` in druid's server properties

this will put it in SQL compliant null handling mode. and will behave more like other RDBMS

if i understand you correctly— you can set `druid.service` to change the service name. however the default is typically something like `druid/broker` not `&lt;host&gt;:&lt;port&gt;`. so maybe i don't understand you correctly :thinking_face:

I have those set, they are things like druid/broker etc. But those aren't what displays in the UI. I assumed they were for zookeeper discovery.

But they also aren't unique. I could be using k8s pod names for example for all of my servers.

hm the UI might be simply concatenating the host and port then

if you did want to override them, you could use `druid.host` and `druid.port`

but just make sure the service is actually reachable on that host/port, else cluster connectivity will suffer

It would not be reachable. But hmm, I might be able to use k8s internal dns names.

Something like mypod.mydruidnamespace.pod.cluster.local. That's something at least.

image.png

i can repro this, i think it might be an issue with using the sampler and the kafka input format

if you want to use the web-console data loader (which uses the sampler api), i think you’ll need to just set the format as “regex” to get everything you want, and then change it to kafka input format at the last step (and add any additional dimensions/transforms/aggregations/whatever else if using stuff from the header or elsewhere)

created <https://github.com/apache/druid/issues/13879>

<@U030C8H59T8> thanks for troubleshooting, that was helpful to know its not only my end when I went click back in the console to try and parse the data. I will experiment around with the ingestionSpec and see how to extract the dimensions for kafka headers without using the "parse data" section.

It's a sampler issue, which is only used by the UI, so you could always submit the JSON supervisor spec directly (without data loader UI)

<https://github.com/apache/druid/pull/13900> is fix for sampler side of things

Do you have `GOOGLE_APPLICATION_CREDENTIALS=/path/to/service_account_keyfile` set on all your middleManagers ? <https://druid.apache.org/docs/latest/development/extensions-core/google.html#required-configuration>

I have a single server instance running, and credentials are set as an ENV variable. I set it on .bashrc. Is there anywhere else I should configure this?

I can verify they are the correct credentials because I can preview the data on the data loader wizard.

what's the replication factor?

What I've seen is that if you have 2 segments (just 2) and 3 nodes, on druid 0.21, they keep load/unloading between the nodes until I add more segments

but that was a rare case, bc usually the number of segments is &gt; than the servers, so I assume that was an edge case, also, I was running locally with docker

<@U03149ALHN3> do you have any tiering across 20 historical nodes?

The coordinator logs would give you more insights into why this is happening.

You could try switching from cachingCost to cost and see if it helps.  (Needs coordinator restart)

Cost is generally slower than cachingCost but the following changes to the dynamic config might help.
If you are on 25.0.0, please try setting `useRoundRobinSegmentAssignment` to true as this will assign segments in a round robin fashion across servers and balance them later with the chosen strategy. You could then increase `maxSegmentsToMove` , and set `useBatchedSegmentSampler`  to true for faster balancing.

the RF is 2, this cluster is on Druid 23.
We don’t have any tiering between 20 historical servers. They are all the same.

Thanks for informing me of `useRoundRobinSegmentAssignment`. I didn’t know that.

oh, I read it again, you have just 2 segments per date, now it makes more sense to me why you have 20 historicals
I was thinking "why so many spare servers"

Follow up question, the partitioning strategy does not use historical hostname as part of its key right?

Which means even though I only have 2 segments per partition date, they should still be spread out across 20 historical servers, right?

I think I also saw in the code that __time column is not used if you specified the hashed column explicitly, is my understanding correct?

If so, should I do this: `hashed(__time, myColumnId)` instead of `hashed(myColumnId)` ?

I also have a suspicion that when the data engineer created `myColumnId` , the spread is uneven. The ID looks strange, it’s not a normal UUID

i'd try `cost` instead of `cachingCost` and see if that helps… I've noticed some discrepancies between them and `cost` generally seems to do a better job

and yeah the hostname of the historical isn't part of the balancing

neither is the size of the segment, believe it or not

it tries to spread out segments from the same datasource and especially same interval

looking at number of segments with "similar" interval or same datasource on the target server when considering balancing

yes, you can. you can specify peon log4j config in `druid.indexer.runner.javaOptsArray` of `middleManager/runtime.properties`, and MM log4j config in `middleManager/jvm.config`

both with system properties like `-Dlog4j.configurationFile`

setting this -Dlog4j.configurationFile will overwrite DRUID_LOG4J value ?

+ I am running druid as docker containers ..

hmm i'm not sure. I am not familiar with how `DRUID_LOG4J` works in docker containers

what works for me on docker for middle-manager
1. set DRUID_SERVICE_LOG4J - the log4j xml content is configure to produce json entries log events  --&gt; will create file on this path - /tmp/conf/druid/cluster/data/middleManager/log4j2.xml
2. add JAVA_OPTS to docker env - JAVA_OPTS="-Dlog4j2.configurationFile=/tmp/conf/druid/cluster/data/middleManager/log4j2.xml"

i m trying to set storage config in environment file but when i look i side container file it is not over ridden any reasons?

can you please share the config .. I will try to help .. if not solved yet ..

<@U03PD5PMTHS> sure i have multiple node on docker compose file i have environment file 

# DRUID_SINGLE_NODE_CONF=micro-quickstart

 
druid_emitter_logging_logLevel=debug

 
druid_extensions_loadList=["druid-histogram", "druid-datasketches","druid-kafka-indexing-service","druid-lookups-cached-global","postgresql-metadata-storage","druid-multi-stage-query"]

 
druid_overlord_port=8090
druid_zk_service_host=zookeeper

 
druid_metadata_storage_host=postgres
druid_metadata_storage_type=postgresql
druid_metadata_storage_connector_connectURI=jdbc:<postgresql://postgres:5432/druid|postgresql://postgres:5432/druid>
druid_metadata_storage_connector_user=druid
druid_metadata_storage_connector_password=FoolishPassword


 
druid_coordinator_balancer_strategy=cachingCost

 
druid_indexer_runner_javaOptsArray=["-server", "-Xmx1g", "-Xms1g", "-XX:MaxDirectMemorySize=3g", "-Duser.timezone=UTC", "-Dfile.encoding=UTF-8", "-Djava.util.logging.manager=org.apache.logging.log4j.jul.LogManager"]
druid_indexer_fork_property_druid_processing_buffer_sizeBytes=256MiB

 
druid_storage_type=local
druid_storage_storageDirectory=/opt/shared/segments
druid_indexer_logs_type=file
druid_baseDirectory=/opt/druid/data

 
druid_indexer_logs_directory=/opt/shared/indexing-logs
druid_conf_dir=/opt/data

 
druid_processing_numThreads=2
druid_processing_numMergeBuffers=2

 
DRUID_LOG4J=&lt;?xml version="1.0" encoding="UTF-8" ?&gt;&lt;Configuration status="WARN"&gt;&lt;Appenders&gt;&lt;Console name="Console" target="SYSTEM_OUT"&gt;&lt;PatternLayout pattern="%d{ISO8601} %p [%t] %c - %m%n"/&gt;&lt;/Console&gt;&lt;/Appenders&gt;&lt;Loggers&gt;&lt;Root level="info"&gt;&lt;AppenderRef ref="Console"/&gt;&lt;/Root&gt;&lt;Logger name="org.apache.druid.jetty.RequestLog" additivity="false" level="DEBUG"&gt;&lt;AppenderRef ref="Console"/&gt;&lt;/Logger&gt;&lt;/Loggers&gt;&lt;/Configuration&gt;

Please refer to <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html#task-processing-threads-and-buffers>

this mainly happens when you have a single, huge column

it can happen if you have big, long strings or arrays or something like that

updating your processing buffers won't help; these buffers are different ones related to ingestion

you could try reducing your target segment size instead; that generally helps if you have these really big columns

Yes, thanks <@U0344FW86DD> that because we didn’t apply the subtask in the ingestion json then Druid get default is 1.

the error bounds can vary depending on the size of the sketch use
```HLL_SKETCH_ESTIMATE_WITH_ERROR_BOUNDS(ds_HLL(tmp_hash, 4, 'HLL_4'),3)```
to get the upper and lower bounds in the 3 sigma limit.

thanks. but whats the logic behind this. i would have imagined that with lgk=4, we would use lesser no of bits to represent the aggregate field and hence would get less no of unique values.
I cannot imagine why the could would go up!

I don't understand all the math, but estimates can be off both ways - over or under.  I think of it as, OK I have a few hash values here, from that, guess how many of the inputs were actually unique?  I get that it's kind of surprising that it doesn't remember that it started with 50 values, so it must be less than 50, but afaict it doesn't.  Seems like a good question for #datasketches in ASF slack, so I just asked there  - <https://the-asf.slack.com/archives/CHWKFEHUZ/p1678136969191449>

*Alexander Saydakov*  [1:48 PM]
keeping this count is an extra resource. what are the chances that it is going to help? usually the distinct count is much less than total count. I would suggest to start from a desired accuracy. Say, you want to be within 1% with 95% confidence. That translates to a particular K. Unbiassed estimate means it is as likely to overestimate as underestimate, and that is a good property. In some rare cases in some particular circumstances the estimate might be higher than total count. Why do you think it is a problem if you are guaranteed to be, say, within 1% with 95% confidence?

<https://the-asf.slack.com/archives/CHWKFEHUZ/p1678139286601589>

thanks Ben, i didnt know about that channel and dont have access, so thanks for carrying it forward.
"within 1% with 95% confidence" - where can i find a source on this info to understand this more?

got it here: <https://datasketches.apache.org/docs/Quantiles/OrigQuantilesSketch.html>

```LgK=4, LgT=20, factor=.83266, RSE=0.20```
The above means that with +/- 2 Standard Deviation, for 95.4% confidence, will have error “bounds” of -30% to +50% , if no.of unique values start crossing 8 in the data set?

with `APPROX_COUNT_DISTINCT_DS_HLL(tmp_hash, 4, 'HLL_4')` you have lgK = 4; a very small value, so you'll get high error

i'm wondering why you're wanting to use lgK = 4?

the default is 12 which is pretty balanced between size/speed/accuracy

an error table i made for a talk last year

the table doesn't go below 10 because the errors are too high imo

"i'm wondering why you're wanting to use lgK = 4?"
well, I am using groupBy queries and they being resource intensive, crashed at lgk=12 with wanting more aggregate buffer space.
Asking for more buffer space from the OPS team for my druid prod servers is also a big task.
So i started looking into the queries and found a variable to tweek. and brought the lgk down to 4 from 12. which worked.
Now that i know what it does, i would revert it back to 12 and also look into the topN queries. :slightly_smiling_face:

"an error table i made for a talk last year"
Also, can i have link to the talk, if its public.

it wasn't recorded but i hope to do a recorded one at some point!

ah, got it. i guess you got a resource limit exceeded error?

you could also ask your ops team to enable disk spilling using `druid.query.groupBy.maxOnDiskStorage`

I am afraid that might add disk-io latencey to an already heavy query -
25s currently with a ui timeout of 30s.
Also, it would result in regression + perf. testing all the other existing queries.
I want to keep the impact to a minimum. So looking into replacing groupBy with topN

you could page to disk and enable parallel combine in the historical. This could get you some performance benefit if the merge across segment results is the bottle neck

Thanks Gian and Vijay
Some observations:
case1: _*druid.query.groupBy.maxOnDiskStorage=1Gb*_
response time: 63s
case2: _*&gt; druid.query.groupBy.maxOnDiskStorage=1Gb + “numParallelCombineThreads”:10*_ 
response time: 97s
Latency seems to be increasing

yeah that's good to know! based on other cases i've seen, i'm not sure parallel combine actually helps in most cases. we're thinking about moving away from that and toward a different approach for big groupBys

hmmm.....the sketch is a lot faster on number fields. You could convert the guid to a number using a hash function in the pipeline before Druid. This will likely speed up the query a lot

this can happen if historicals run out of memory or cpu. I would run queries and monitor historical cpu and memory utilization

check historical logs for any messages, and system logs (for eg OOM-killer)

it was this error - <https://stackoverflow.com/questions/16789288/java-lang-outofmemoryerror-unable-to-create-new-native-thread>
I solved this by update ec2 node ami image which include update ecs agent / docker version

Hi , what logger I need to set on log4j config inorder to have the sql query executed on historical ..
I want that in case of query error , I will have it on logs and then I will be able to execute it on the router query console

try
```{
    "type": "equalDistribution",
    "affinityConfig": {
      "affinity": {
        "datasource1": [
          "mm-1:8093",
          "mm-2:8093",
          "mm-3:8093"
        ],
        "datasource2": [
          "mm-4:8093",
          "mm-5:8093",
          "mm-6:8093"          
        ]
      },
      "strong": true
    }
  }```

Is this the recent fix from druid ? The older config was working all these while

where are you posting and viewing these configs? which API or UI? (if UI could you post a screenshot?)

Screenshot 2023-03-07 at 8.22.48 PM.png

Screenshot 2023-03-07 at 8.21.47 PM.png

Under *Overlord dynamic config*. Refer these screenshots.

I think the UI split up the thing into two pieces

on the server it's a JSON object with `selectStrategy` and `autoScaler`

Yes UI shows two boxes`selectStrategy` and `autoScaler`
But json(older config) I posted used to work by pasting it under `selectStrategy` in UI.

the server didn't change so it was possibly a UI change. <@U030W50CNAC> did this change recently?

Not that I know off... the screenshots show a very old version of Druid (notice the greenish console tint) but I do not recall any changes to the UI there. ASAIK it always used to split the `selectStrategy` and `autoScaler`

I would not expect that posting the entire config under `selectStrategy` would ever work

Not sure if Im following the problem, but looks like your problem is actually to get `start` and `end` out of the group by (and you can also get rid of the outer select by adding a `having seg_num &gt; 1` clause in the inner select)

what is "table_name" - what table are you querying?  Also, there's a syntax error - `and` before `group by`.   Also, like Daniel said, you might be able to do it in one query instead of an outer and inner query.

this is not answering your question directly, but have you looked at druid's new json column feature? <https://druid.apache.org/docs/latest/querying/nested-columns.html>

it's a good way to deal with sparse stuff like the `qualifiers` map in your sample record

I’ve read about json column some time ago, but I wasnt considering it as I’m hoping to run queries like

```select qualifiers, count(*), count(distinct id) uniqueIds
from table
where qualifiers = 'type1=value1'

[type1=value1, type2=value2, type17=value17]  | 17 | 10
[type1=value1, type3=value3, type17=value34]  |  3 |  3
[type1=value1, type8=value2, type17=value34]  | 11 | 10```


with json columns that'd be

```SELECT qualifiers, count(*), count(distinct id) uniqueIds
FROM table
WHERE JSON_VALUE(qualifiers, '$.type1') = 'value1'````

the json columns are internally indexed so this is about as fast as the MV string approach

interesting, I was expecting MV approach to be like much faster due to the bitmap indexing.

the json approach is much more flexible, if it’s also as fast as the MV, it’s kinda a nobrainer. I’ll test it.

hope it works well for you!

yeah, we put a lot of effort into making the JSON column implementation in many cases just as speedy as not using JSON columns. since the API is so nice we want people to be able to use it without worrying about perf

Does it work with roll ups and LATEST functions?

yep: <https://druid.apache.org/docs/latest/multi-stage-query/concepts.html#rollup>

It’s working very well so far! Haven’t tried under heavy load or fancy queries yet, but so far so good

the escalator is important: it's how the servers authenticate to each other. so you should at least have a Druid metadata user for this

LDAP and Druid metadata users would be separate authenticators, but you can use the same authorizer for both if you like. the authorizer you define would need to be able to handle identities from both authenticators

When you talked about metadata users, what do you mean by this?

I assumed it meant users whose authentication info (username / hashed pass) are stored in Druid's metadata store

<@U0344FW86DD> thanks for your replies.
About
&gt; “the authorizer you define would need to be able to handle identities from both authenticators” 
Is this may be done by configuration only or some additional implementation is needed?

TIL: you can use the same authorizer for both. We always separate them out: basicAuth &amp; LDAP.
<@U04B91YT1BP> authentication and authorization are all configured using the Coordinator Rest API.

The way we do it is by:
1. configured authentication &amp; authorization for basicAuth.
2. configured authentication &amp; authorization for LDAP.
3. And then configure an escalator by first checking the basicAuth and if missing check LDAP.

auth-config

Hi Again,
Tried to define following but still when starting quickstart I’m getting errors about internal user `druid_system` is unable to authorize

Anything special I missed ?
```023-03-08T14:34:36,236 INFO [qtp456172878-137] org.apache.druid.security.basic.authentication.BasicHTTPAuthenticator - Exception authenticating user druid_system - Unauthorized
2023-03-08T14:34:36,474 INFO [qtp456172878-136] org.apache.druid.security.basic.authentication.BasicHTTPAuthenticator - Exception authenticating user druid_system - Unauthorized```

Ok I found that I can’t even Authz with basic settings without additional LDAP.
Might it be because of some users already stored in metadata in some way ?
```druid.auth.authenticatorChain=["MyBasicMetadataAuthenticator"]

druid.auth.authenticator.MyBasicMetadataAuthenticator.type=basic
druid.auth.authenticator.MyBasicMetadataAuthenticator.initialAdminPassword=password1  # admin user
druid.auth.authenticator.MyBasicMetadataAuthenticator.initialInternalClientPassword=password2 # druid_system user 
druid.auth.authenticator.MyBasicMetadataAuthenticator.credentialsValidator.type=metadata
druid.auth.authenticator.MyBasicMetadataAuthenticator.skipOnFailure=true
druid.auth.authenticator.MyBasicMetadataAuthenticator.authorizerName=MyBasicMetadataAuthorizer

druid.escalator.type=basic
druid.escalator.internalClientUsername=druid_system
druid.escalator.internalClientPassword=password2
druid.escalator.authorizerName=MyBasicMetadataAuthorizer

druid.auth.authorizers=["MyBasicMetadataAuthorizer"]

druid.auth.authorizer.MyBasicMetadataAuthorizer.type=basic
druid.auth.authorizer.MyBasicMetadataAuthorizer.initialAdminUser=admin
druid.auth.authorizer.MyBasicMetadataAuthorizer.initialAdminRole=admin
druid.auth.authorizer.MyBasicMetadataAuthorizer.roleProvider.type=metadata```

working-auth-config

After removing Derby DB again and first initializing basic auth and only then adding LDAP auth I succeed to work with both LDAP and metadata users.
These are final configurations in case it will help to someone.

oh yes, the metadata DB must be a networked DB (PG/MySQL), and not embedded like Derby.

Sorry guys if its too basic of a question. any hint would be helpful.

groupBy is always exact unless you do `forceLimitPushDown`

topN does the approximation you describe, which generally works well for "top heavy" distributions and generally does not work well for uniform distributed data

Druid does this automatically via checkpointing for seekable streaming sources

We have multiple consumers for kafka. Except Druid everything auto commits to Kafka. which helps us compare lag among consumers using metrics like "aws.kafka.sum_offset_lag".

Druid doesn't auto commit to Kafka which makes it difficult to compare lag with other Consumers.

Druid has "druid.ingest.kafka.lag" metric, which somehow not equivalent  to "aws.kafka.sum_offset_lag" to compare lag with other Consumers.

I'm using Datadog for monitoring purpose.

Druid doesn't use Kafka's builtin consumer group management; we manage our own offsets in Druid to ensure exactly-onceness

so the `enable.auto.commit` and `group.id` wouldn't be used

I would recommend using approx_count_distinct_ds_theta with the size you want. The theta sketch will you give you exact count upto the size of the sketch without creating performance bottle necks. Using exact count distinct usually causes issues and even if it works, it will likely be slow.

Thanks for the recommendation. We are certainly taking that into consideration, and will probably implement that as it seems like the only logical way to advance as data grows. However, I’m curious about why it happened suddenly when it was working fine. If you have any ideas/hints I would be happy to hear them.

from what i know errors with sequences are usually timeout errors. With exact counts this can happen if data volume, cardinality and query concurrency increases

It’s most likely the latter because I’m getting the NullPointerException back immediately without a delay.

Thanks for your help and input, much appreciated.

+1 to Vijay.  I'm curious, too, if you have multiple count distincts in the query, and if so, did you try setting `"useGroupingSetForExactDistinct"`: `true` in the <https://druid.apache.org/docs/latest/querying/sql-query-context.html|query context> (along with the other "exact" settings)?  Sounds like you're hitting a bug, I wonder if this would be a workaround.  Either way, sketches are a good idea in the long run.

Convinced the product people to just use sketches, but to answer your question <@U0411DE2SS0> No I don’t have multiple distinct in the query, one is enough to make it fail.

maybe the problems is way too many small segments, so your historicals have too many metadata structures to handle? Do you know how the average amount of rows per segment you have?

for basic cluster tuning, check <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html|here>

The memory mapping that Historicals do for their local cache of segments is OS driven. It uses that memory left over on the OS after the historical takes its Heap and max Direct Memory allocations. What error are you getting ? Perhaps the Historical settings are too high overcommitting the resources. What nodes are you running on? What are your historical JVM and config settings. Are the Historicals running on their own on the node? 

historical run on i3en.3xlarge ec2 node
as docker container
histrocial config (druid version 0.19.0)

```- Name: AWS_REGION
  Value: us-east-1
- Name: druid_historical_cache_useCache
  Value: true
- Name: druid_historical_cache_populateCache
  Value: true
- Name: druid_segmentCache_lazyLoadOnStart
  Value: true
  # set to 6.5 TB = 7146825580544 bytes (it should be smaller then /opt/data total disk space )
- Name: druid_cache_sizeInBytes
  Value: 7146825580544
- Name: druid_server_maxSize
  Value: 7146825580544
- Name: druid_segmentCache_locations
  Value: '[{"path": "/opt/data", "maxSize": "7146825580544"}]'
- Name: druid_segmentCache_locationSelector.strategy
  Value: mostAvailableSize
- Name: DRUID_XMX
  Value: '35g'
- Name: DRUID_XMS
  Value: '35g'
- Name: DRUID_MAXDIRECTMEMORYSIZE
  Value: 65536m
  #Value: 13312m
- Name: druid_processing_buffer_sizeBytes
  Value: 536870912
- Name: druid_processing_tmpDir
  Value: '/opt/data/processing/tmp'
  ```
segment size avg - from router - 54.27 MB
segment number of rows - dont know
Error is taking place after succed to load 98% of segments
ERROR -
```OpenJDK 64-Bit Server VM warning: INFO: os::commit_memory(0x00007f9b06001000, 65536, 1) failed; error='Not enough space' (errno=12)
#
# There is insufficient memory for the Java Runtime Environment to continue.
# Native memory allocation (mmap) failed to map 65536 bytes for committing reserved memory.```

what I tried is to delete oldest segments "unused" and relaod segments agian after cleaning localdisk

Did that work?  Here's an old <https://www.druidforum.org/t/native-memory-allocation-mmap-failed-to-map-65536-bytes-for-committing-reserved-memory/8187/7|thread by Vijay>, reducing segment count (eg compacting) is ideal.  Otherwise, might have to increase vm.max_map_count (eg, see <https://support.imply.io/hc/en-us/articles/360060443474-Historical-will-not-start-due-to-OOM-and-failed-to-map-errors-though-there-is-enough-heap-available-and-ulimit-is-set-very-high|here>).
VDid that work?  Here's an old thread from

it still on loading so wait to see if the delete old works ..

is historical will try to load all contnet of local disk to cache .. can this be limited by some configuration ?

how do I know to which value to increase vm.max_map_coun ?

historicals will load whatever segments are assigned to them by coordinator.  Best to compact to fewer segments, or maybe add historicals.  I'm not sure on what to set max_map_count to, just that it's currently too low for the number of segments you're loading.

• Druid UI provides a lot of statistics as to how many segments are generated per partition date and how many rows in segment file. This info will help you tune how many segment files are ideal.
• Historical data itself is a cache. It will load everything you tell it to load in the retention rules.