Apache Druid

MicrosoftTeams-image (9).png

MicrosoftTeams-image (10).png

how are you running it on kubernetes ? it seems like you are using k8s extension and not zk.
is the extension configured correctly ? do you have appropriate rbac ?

I have heard of scenarios where the task is instructed to shut down prematurely, from the console's perspective the task was aborted, but from the task's perspective it shut down successfully.  I don't know if this is the case here though ...

image.png

operator_role_and_rolebinding.yml

Yes Adheip, that is correct. We are using k8s extension.
below is how we have configured the k8s extension and rbac

I am not sure what you are talking about, I routinely see super high cardinality string dimensions, with millions and millions of values.

Is your only concern the size of the index in the segment?

if that is truly your concern you can turn off bitmap indexes for specific dimensions (that will have a perf hit on filtering on those dims)

numeric dimensions do not have bitmap indexes

Thanks for the reassurance here!
I'm glad to hear that both strings and numeric dimensions can scale to that degree. I know the bitmap indexes are compressed but I haven't played with how compressed they can be for longer bitmaps. In my case numeric is a more natural representation of this dimension, and I don't think we'll usually filter to just one value, so the filter perf hit won't be too bad.

Is there any interesting interaction with rollup being true / false? not knowing too much about the implementation, with rollup enabled you'd have to partition / shuffle the data by dimension to eventually get to one row for that dimension combination.

nested columns and soon, new in druid 26, numeric columns created with ‘auto’ schema do have indexes; they use a cardinality based threshold to decide when to use or not use the indexes in query processing

<https://github.com/apache/druid/pull/13977> has some low level details

disclaimer that this stuff is new and somewhat experimental

but based on the measurements will likely apply that strategy to classic string columns at some point in the future

also since those auto numeric columns have bitmap indexes, they are larger than classic numeric columns, in exchange for faster filtering

at some point in the future i’d like to add pure range based numeric indexes as a middle ground for very high cardinality numeric columns instead of a bitmap per value, but haven’t gotten there yet

you should look at the coordinator log and historical logs to determine what is going wrong with the load process.

<@U056P4MRMRN> Are you just deploying Druid locally on your laptop? If so, requesting Max tasks:3 is too many for the quickstart deployment. It is configured for a max of 2 tasks. When I try to run with 3 tasks it shows me this message:

If I run it anyway, it still finishes and uses only 2 tasks. It was done in 18 seconds. What is different about what you see? I tested with version 25.0.0

<@U030K4UM3H7> I have installed it on VM, with version 25.0.0 only, it allowed me to run with setting of max 3 tasks, (earlier by default it was set to max task 2) I changed it when it took similar kind of time with max task 2 setting. I tried running it , but there is no result it is still running, its been 70 hours now

<@U03R5H3B750> ,I am checking logs under log folder, log/historical.log, Sharing the snaps of log files (the task is running since 5th may), can you guide me what could be the cause and how can we resolve it.

The error “Address already in use” says there is already a service running and you want to start again a service in the same port.

Should we cancel the job and wait for the data source to be fully available

yeah, how can I check id data source is fully available?

In coordinator, I also get to see , I have no server message

As it is not finding the coordinator, it wont proceed ahead, should we stop and start the server?

If you navigate to druid console and ser ices page, do you see coordinator listed their?

Check here if coordinator service(Leader)  is available

I do not see Historical service being run here, Can you check the logs for historical and try to start the historical service.

I am seeing below error in the historical file

The default port for the historical service is 8083, Since the router is already bound to 8888, the historical is unable to start. Is there a reason for using a different port on the historical?

which version are you running for the operator ?

<https://github.com/datainfrahq/druid-operator#kubernetes-version-compatibility>

26.0 introduces an actual Array data type, so I assume you mean a multi value string column. You can find all the <https://druid.apache.org/docs/latest/querying/query-context.html|query context options here>. The one that you are looking for is `groupByEnableMultiValueUnnesting` which is part of the <https://druid.apache.org/docs/latest/querying/groupbyquery.html#advanced-configurations|group by query context options found her>e.  I tested it on 25.0 and it is working, it even respects your choice of MV behavior at ingestion, I tested with ARRAY to see that it respected the order of the elements, and distinguished between different orders when grouping:
inline data:
```{"mv_column":["a","b","c"],"metric":1}
{"mv_column":["d","a","b"],"metric":3}
{"mv_column":["b","a","d"],"metric":5}```
query context:
```{
   "groupByEnableMultiValueUnnesting":"false"
}```
SQL:
``` SELECT "mv_column", count(*) FROM "inline_data" GROUP BY 1```
Results:

Wait, what!? I tried another SQL:
```SELECT MV_TO_ARRAY("mv_column"), count(*) FROM "inline_data" GROUP BY 1```
Now it works:

so I wondered if the query context was doing anything, I removed it and the result was still correct, same as above. In 26.0 with Array data type and UNNEST functions, this gets more natural IMHO.

also tried to filter with:
```SELECT MV_TO_ARRAY("mv_column"), SUM("metric"), count(*) 
FROM "inline_data"  
WHERE MV_TO_STRING("mv_column",100) like '%c%'
GROUP BY 1```
and this:
```SELECT MV_TO_ARRAY("mv_column"), SUM("metric"), count(*) 
FROM "inline_data"  
WHERE "mv_column" like '%c%'
GROUP BY 1```
and both worked.
After rereading your question I see that what you want is that only the item that matches be in the results, not the other items in the row that matched. This will at least give you the right number of rows, even though it still shows the other values in the MV column:
```SELECT MV_TO_ARRAY("mv_column"), SUM("metric"), count(*) 
FROM "inline_data"  
WHERE "mv_column" like '%c%'
GROUP BY 1```


I tested unnest on 26.0-SNAPSHOT build where it is still experimental but the results are what I think you are expecting:
```SELECT n."mv_item", SUM("metric"), count(*) 
FROM "inline_data", 
     UNNEST("mv_column") AS n("mv_item")  
WHERE n."mv_item" like '%d%'
GROUP BY 1```

`groupByEnableMultiValueUnnesting` is only for (native) groupBy, the first query you, <@U030K4UM3H7>, wrote planned to a `topN`

this flag is used in MSQ ingestion (to make sure that your ingested data does not expload)

at query time, as you have discovered, the correct thing to do is to wrap you MV value in an `MV_TO_ARRAY(...)`

If you're using druid SQL, there's also a new (I think) function, `MV_FILTER_ONLY`.  (See <https://druid.apache.org/docs/latest/querying/sql-functions.html#mv_filter_only|here>._)

that's a good note but that would only let you express an `IN` filter, not a `LIKE` filter

<@U030K4UM3H7> when should we expect release of 26.0 version 

I hear it will be soon. You can follow it’s progress at <https://github.com/apache/druid/milestone/50|https://github.com/apache/druid/milestone/50>

sorry, I was busy for the rest of the day and didn't came back for the answers
I was indeed only expecting the matching values in the results, when I tried using a other query, the results where the same (I think there apache calcite just rewrote it removing the subquery), I just ended filtering in the spreadsheet after downloading

Could work, worth a try...  This <https://support.imply.io/hc/en-us/articles/115004960053-Migrate-existing-Druid-Cluster-to-a-new-Imply-cluster|KB> might give ideas for other options.  (It says it's for Imply or Imply Cloud, but I don't think it's really specific to that.)

I think the option <@U0411DE2SS0> linked is going to be better— using HTTP is going to be unwieldy and tougher to scale!

We evaluated both options. The http ingestion works if we put an nginx as proxy in between, which converts the GET requests into proper query POST requests. We then have the freedom to configure an endpoint with parameters, e.g. to query chunks of data. It requires some setup but it might be more error prone to migrate data between different druid versions or setups (local storage vs s3 storage).

The deep storage migration works like a charm for our purposes as well, so we have our toolkit ready. Thanks!

Not that I know of.  (We'll see if anyone else has a way.)

<@U0411DE2SS0> thank you for answering :)

This is the PR that swapped the defaults: <https://github.com/apache/druid/pull/13092>. So you can use that as a guide if you'd like to undo stuff

IMO, it's easier to set `druid.serverview.type` in `common.runtime.properties` since multiple services do use the serverview. Doesn't hurt anything to do it there vs. the server-specific files. But, yeah, it's Broker and Coordinator

thanks! so sounds like my colleague and i are both right — one of us is saying "broker and coordinator" and the other is saying "eh might as well do common" :slightly_smiling_face:

You'll probably want to set it up for all services, so I think that would be the common.runtime.properties file. <https://druid.apache.org/docs/latest/configuration/index.html#tls|The docs on this are here. >

Screen Shot 2023-05-08 at 4.19.27 PM.png

did you mount the directory to your docker as volume ?

you should specify a path, probably needs a `/` or something.

yes. that seems right to me. very likely you already have overridden druid.indexer.runner.type to something else since the default is local and no one uses that.

For Hadoop, Druid indexing runs as a regular map/reduce job on YARN, so you can use all your standard YARN and M/R approaches to tune things (splits, # mappers/reducers, container size, etc)

That being said, most people use one of Druid's builtin options, like <https://druid.apache.org/docs/latest/multi-stage-query/index.html> which don't require Hadoop and is IMO easier to configure. So you could also try that

If nothing set default is 1 or we can set as per requirement. Not sure if we can add minimum  concurrency
<https://druid.apache.org/docs/latest/ingestion/native-batch.html>

thank you for answering
yes, I know that it’s default is 1, and to enable concurrency we need to set the max
but as I said, even when setting it to a high number, it sometimes ends being executed in a single subtask, not benefiting at all from the concurrency
so is there a way to force a minimum concurrency?

For native batch, `splitHintSpec` controls the number of tasks that will get launched: <https://druid.apache.org/docs/latest/ingestion/native-batch.html#split-hint-spec>

For SQL-based ingest (<https://druid.apache.org/docs/latest/multi-stage-query/index.html>) when `taskAssignment: max` (the default) we always launch as many tasks as possible below your `taskCount`. when `taskAssignment: auto` it's based on the size of inputs, but will also always remain below `taskCount`.

Screenshot 2023-05-10 at 11.27.31 AM.png

You can specify in console compaction config, Please check the below screenshot for reference.

MaxSplitSize:- this is related to the “split hint spec” used for parallel ingestion: <https://druid.apache.org/docs/latest/ingestion/native-batch.html#size-based-split-hint-spec>
It imposes a limit on the maximum number of bytes that can be processed by a single parallel sub task. I think the case where you may want to adjust it from the default would be if you get OOMs with the 1GB default (lower the limit).

I am facing issue in applying compaction on a table, i have posted the issue in general channel

What are the changes you are making in the new spec?

defining max rows per segment, earlier i only mentioned the month granularity

but i need even more compact segments with 5million records in each as the size is very less

I will try on my cluster and let you know

I thought the Coordinator would pick up on config changes in the compaction spec, I just can't find the source file right now that will show explicitly what is checked.  I would think granularity and/or partitioning config changes should trigger a new compaction.

I wonder what change exactly you made? Could be something that isn't stored as part of the compaction state, in which case a new compaction wouldn't be triggered

Gian can you please explain , I didnt get

Hi Hicham, at Imply every Druid cluster our software spins up has tls/ssl on by default. The way we set it up it is literally impossible to not have/use TLS/SSL in any of our managed environments. So I can assure you it is totally battle tested. Here are the relevant docs: <https://druid.apache.org/docs/latest/operations/tls-support.html>

It's addressed by <https://github.com/apache/druid/pull/12357>

try setting `inSubQueryThreshold` to 0 in your query context and see if that helps

Actually, I saw that fix. But I also need to parse the query before sending it to Druid using the same calcite parser Druid uses.

so its your parser taking time? may I ask why do you need to parse the query? once parsed, does druid get the sql query or the native json query?

Yes, the parsing is taking quite some time ~40+ seconds. I need to parse the query because I need to introduce extra where clauses based on permissions. So, I parse the query to create SqlNodes then convert it back to sql and send it to druid.

A workaround is to use *MV_OVERLAP*(myCol, ARRAY['myVal1', 'myVal2',...]).  Not sure this will work forever, although it should imo.  Planner doesn't try a million variations, just sends it on.

I tried `MV_CONTAINS` before and that didn’t work. Will Try *`MV_OVERLAP`* .

Also
```select l from mytable
INNER JOIN (values ('a'), ('b')  ... &lt;thousands_of_values&gt;)  t(v) on mytable.l = t.v```
seems to work.

Nice, that's more intuitive, gtk, thank you!

you could also use native queries (JSON) which is what SQL plans into, if you end up needing to bypass the SQL parser and planner

I was thinking about sending it as json queries but I think I read somewhere that using SQL is the preferred way, as well some optimizations are only available via sql. Plus, the redundant parsing by druid didn’t seem to matter much, at least for small queries.

For some reason when I parse the query it’s a lot slower (40s) than when druid does it (&lt;1s). My code
```SqlNode parseSql(final String query) throws SqlParseException {
        final SqlParser parser = SqlParser.create(query, SqlParser
        .configBuilder()
        .setCaseSensitive(true)
        .setUnquotedCasing(Casing.UNCHANGED)
        .setQuotedCasing(Casing.UNCHANGED)
        .setQuoting(Quoting.DOUBLE_QUOTE)
        .setConformance(DruidConformance.instance())
        .setParserFactory(new DruidSqlParserImplFactory()) // Custom sql parser factory
        .build(););
        return parser.parseStmt();
    }```
 seems identical to how druid does it in the CalcitePlanner <https://github.com/apache/druid/blob/master/sql/src/main/java/org/apache/druid/sql/calcite/planner/CalcitePlanner.java#L227>

yeah, it's true in general SQL is the preferred way

hmm you're saying Druid parses it a lot faster than Calcite does when you embed it in your own app?

Seems odd… do you have the same version of Calcite?

Profiling might be useful to hint at what's going on (yourkit or sjk flame graph)

I was missing a RelOptPlanner to optimize queries before doing the parsing. Adding that solved that issue. Thanks!

Druid doesn't have indexes as explicit objects like traditional RDBMSs do.  Instead it stores data organized by time first, then secondarily by a method of your choosing.  Following that, data is stored in columnar format with dictionary coding and compression used based on datatype.

There is some good reading material on this.  I suggest you check out the following:

Basic segment partitioning info: <https://druid.apache.org/docs/latest/ingestion/partitioning.html#time-chunk-partitioning>

Imply Youtube channel 3-part series on seconary partitioning: <https://youtu.be/7caGijktulo>

Apache Druid Basics and ingestion (free) online courses: <https://learn.imply.io/page/learn-apache-druid>

$HOME isn't going to resolve to an environment variable. Adding some logs if anything is showing up there would be helpful as well.

<https://druid.apache.org/docs/latest/configuration/index.html#configuration-interpolation|https://druid.apache.org/docs/latest/configuration/index.html#configuration-interpolation> has information on the types of interpolation available. You might want `${env:HOME}` if the user running Druid has that file in its home directory.

druid.query.logger.type doesn’t seem to exist (<https://druid.apache.org/docs/latest/configuration/index.html>). You could use `druid.request.logging.type` instead.

Ok Frank, thank you. Let me check the suggested one.

You also have an older monitor name from before the code was transferred to Apache

It's called `org.apache.druid.server.metrics.QueryCountStatsMonitor` now

ok Gian, thanks let me check with that too.

Gian &amp; Frank Cao,  the QueryCountStatsMonitor  &amp; druid.request.logging.type worked. thank you.  But i made this changes in broker runtime properties than only worked. when I applied to common runtime properties, it's not worked.

note:  common runtime properties already have 1 JVM monitoring.

why is it looking at localhost, I would focus on that.  It may be that druid.host needs to be set in the config files for each process

I think because I ran it locally on my machine

have you also set `druid.enableTlsPort = true`?

if so, the rest of your configs look ok, so there might be some issue with your keystore or truststore

I fixed this problem but I got another one

Hi Pramod - can you share some info / logs / use case, I'm generally supportive, but looking at it makes me wonder if there's a configuration issue

I know people only normally put together a PR if they've exhausted most options, so, I think it will help to share notes on the things you tried, the config setting, any recordings/traces/flamegraphs/logs that highlight the issue

I've added couple comments for areas that stand out, but I'm not super familiar with much of the code.

esp the semantics of the single thread guaranteeing ordering of operations looks risky to me

When rollups are enabled (I believe this setting is per segment) then the requred metadata is stored with the field to enable further rollups/merging.

the metadata for monthly segment would know its seen X and Y before?  I suppose this is something that is testable too.

Oh, no.  When you ingest data you ingest in either append or replace mode.  If append mode then it is assumed the metrics will be added to.  If replace, then you are replacing all data within a segment of time.

i suppose the scenario I'm suggesting is that the second time Druid sees X and Y, it will reingest it into "NONE" granularity at first, and then later it will try to rollup to "MONTHLY" where it has previously rolled the first time it saw X and Y

in this case, the MONTHLY segment may treat the second instance of X and Y like as if they were some newer data, and not recognize its seen it before?

I am not sure of the rules around ingesting data with different levels of query granularity ... my guess here is that Druid will not know there is replacement data ... once the data is rolled up you cannot "un-roll" it to find individual records to replace.

As far as the merging process, here is the doc page section on query granularity handling during compaction: <https://druid.apache.org/docs/latest/data-management/compaction.html#query-granularity-handling>

... and two sections down: <https://druid.apache.org/docs/latest/data-management/compaction.html#rollup>

thanks, your guess is similar to my suspicion.

For Druid the preferred approach would be:

• keep the data at the more granular level
• use the "latest() aggregate function to always grab the most recent value for X and Y during query time
• optionally create a summary table (datasource) that you regenerate periodically using that "latest" data query.
The only other alternative I know of would be to orchestrate updates to the original data ... that is a batch operation and takes a bit of work, but can be done if you can't use one of the other approaches.

PSA - I have re-enabled the stale bot on github PRs and issues. So if you see an issue being marked stale, that you think shouldn't be, please leave a comment on the issue.

This gets generated for Native batch ingestion tasks not specific to compaction:

From guide:
```In the partial segment generation phase, just like the Map phase in MapReduce, the Parallel task splits the input data based on the split hint spec and assigns each split to a worker task. Each worker task (type partial_index_generate) reads the assigned split,

The partial segment merge phase is similar to the Reduce phase in MapReduce. The Parallel task spawns a new set of worker tasks (type partial_index_generic_merge) to merge the partitioned data created in the previous phase.```
<https://druid.apache.org/docs/latest/ingestion/native-batch.html>

yeah in particular they could be subtasks of `compact` or `index_parallel`

Thanks <@U03R5H3B750> and <@U0344FW86DD> !!!

Is there a way to differentiate the `single_phase_sub_task`  - like which one trigger due to `compact` or `index_parallel`  ?
Looking to configure overlord dynamic config so that `compact` task are send to seperate tier ?

Hi all. Wondering if anyone knows how to use dask as a replacement for pandas while processing druid data. Any docs/pointers would be really helpful. The background to this request is: some of the druid queries we run return a lot of data that's larger than the machine memory. Dask is be able to process datasets that are larger than memory, and I'd like to explore it. Thanks in advance.

if you are running the quickstart, stop the services then remove (`rm`) the `var` directory

then start the services again druid should be good as new

this works because in the quickstart config the bar directory is your everything (metadata store &amp; deeps storage)

if you have a cluster you will want to clean/reset each of these individually as per how your cluster is configured

whats your end purpose? If you just want to get rid of the data and sql tables, you can simply delete all data. The tables will stop showing too.

I haven't tested this, but from the docs I see you need to load `"simple-client-sslcontext"` in the loadList. Have you done this? the other difference I noticed is the `druid.client.https.trustStoreType=PKCS12` which is set to `jks` in the example <https://druid.apache.org/docs/latest/operations/security-overview.html#update-druid-tls-configurations|here>.

I loaded "simple-client-sslcontext", Yes I changed trustStoreType=PKCS12but It won't work