Apache Druid

Hi Pavithra,  I'm not sure I understand exactly what you are trying to do ... if by "raw value" you mean you have an integer value representing (Unix) time, then you can do a min() on that value within the group, and apply a conversion function like MILLIS_TO_TIMESTAMP() to display it in timestamp format.  Is that what you were looking for?

E.g. here I am testing out some operations on the wikipedia datasource ... I don't have a secondary (integer) time field, but I converted __time to an integer so I could apply the min() function

```select channel, MILLIS_TO_TIMESTAMP(min(TIMESTAMP_TO_MILLIS(__time))) from wikipedia group by 1```

It would really help if you provided an example. Like a table with 3 or 4 rows and what your minimum actual time calculation you want to see come out of that

It sounds like you are referring to the datacube/dashboard visualization layer in Imply Pivot?   That is not part of OSS Druid.

Otherwise, "views" as in relational DB views, are not a Druid feature either.

You can, however create and refresh "summary" datasources fairly easily in Druid using the new MSQ SQL Ingestion engine:  <https://druid.apache.org/docs/latest/multi-stage-query/index.html>

if "views" here refer to <https://docs.imply.io/latest/druid/operations/views/view-manager/>. then yes, they are not available in OS druid.

One possibility is that the password for the metastore is not correct. Please check that

This is probably due to the form of latest, if that column is a string. I think using `LATEST(expr, maxBytes)` form (<https://druid.apache.org/docs/latest/querying/sql-functions.html#latest>) should resolve this.
`LATEST(expr)` expects that expr is a numeric value.

<@U03NCEXDNJK> Hi
I've tried that too with maxBytes = 16 and 32 but it gives the same error-
```select
            key_score_id ,
            key_entity_id ,
            key_entity_type ,
            key_started ,
            latest(evaluation_started, 16) from "int.cp.druid.intelligent_scoring_test_druid_ingestion"
            group by key_score_id ,
            key_entity_id ,
            key_entity_type ,
            key_started```


<@U031A9D19NU> Hi,
I'm using
```2022.09.0-iap```
druid version

<@U052F1P10TD> A few questions,
• What is the engine you are using for this? You can see it next to the run button. I'm guessing that it is sql-msq-task, and that you are using an EXTERN source in the FROM clause?
• Is the type of the column `int.cp.druid.intelligent_scoring_test_druid_ingestion` string? Currently, MSQ only supports LATEST (irrespective of form of LATEST) on string columns.(<https://druid.apache.org/docs/latest/multi-stage-query/known-issues.html#select>)
If the column is not a string, I believe that sql-native engine does support it. The reason it is using sql-msq-task is that reading from an external source (EXTERN) is only used in msq. If you ingest it into druid first, an sql-native task would work.

Alternatively, you could change the column type to string, depending on your other use cases.

<@U03NCEXDNJK>
• I'm using "sql-native" engine
• the type of the column "evaluation_started" is BIGINT

Also i need this column as BIGINT only so can't change the type to string

I believe `sql-native` should support integer values for `LATEST(expr)`. Could you post the full query and stack trace for the exception?

<@U03NCEXDNJK> sure,
Query -
```select
            key_score_id ,
            key_entity_id ,
            key_entity_type ,
            key_started 
            ,
            latest(evaluation_started, 32) 
            from "int.cp.druid.intelligent_scoring_test_druid_ingestion"
            group by key_score_id ,
            key_entity_id ,
            key_entity_type ,
            key_started```
Full Error-
```Error: Unknown exception

java.util.concurrent.ExecutionException: org.apache.druid.java.util.common.UOE: Cannot make VectorValueSelector for column with class[org.apache.druid.segment.column.StringDictionaryEncodedColumn]

java.lang.RuntimeException```


<@U052F1P10TD> I didn't notice the `group by` clause due to the indentation, my mistake.
The above exception is what is thrown from MSQ engine, so that is interesting to see it for a sql-native query.
Could you try setting "vectorize": false in the query context and see if that resolves this?

this is likely a bug. you should file a GH issue with the stack traces.

To clarify since it could be misunderstood, I meant that currently doing a latest(numeric) in MSQ throws the above exception, not that it can't be thrown in sql-native.

Hey <@U03NCEXDNJK>
sorry i'm new to druid ui, where can i set the query context? I'm using druid's query tab to run my query.

Screenshot 2023-04-10 at 4.07.02 PM.png

There should be an option when clicking on the engine `Edit context` .

<@U03NCEXDNJK>
Thanks a lot, this worked for now, but my actual use case is to query this via looker api and there i don't see an option to pass this property.

<@U052F1P10TD> I think this is a bug, we should not need to manually disable vectorize to make the query run. I believe that Latest() should support vectorization as well. Could you raise a github issue for the same as Abhishek mentioned?
I'm not familiar with looker api, you should probably be able to pass the context with the query if it is using the http endpoint

are you able to connect using the kafka console consumer?

I am able to connect to a topic without ssl settings from console but not one with ssl

Check with ssl from kafka console consumer not the druid console.

I am able to connect to the topic from kafka console consumer

Do you mean the deep storage? I think backing the folder for each data source and also backing the metastore should get you what you need

Please try `payload = ['index_some-task-id_2023-04-07T09:43:37.370Z', 'index_another-task_2023-04-07T10:00:00.000Z']`

Thanks! This is working now! Grateful - been stuck for days :smile:

Do you get any error, Could you share screen shot for both the output?

You are right:

```select count(*), count(channel) from (select channel, count(*) numItems from wikipedia group by channel) where channel = 'x'```
returns no records ... whereas this:

```select count(*), count(channel) from (select channel from wikipedia) where channel = 'x'```
Correctly returns zero counts.

I will log an engineering ticket for this.

Thanks.  John

I think this happens because Druid returns no buckets if don’t match anything for `GROUP BY`.
This is consistent with the behaviour of SQL, if we explicitly mention a group by, i.e.
```select count(*) from wikipedia where channel = 'x' group by 'dummy_key'```
returns no records.
However if the `GROUP BY` clause is not present then we return the single record with count as 0 (which is where Druid’s behaviour is diverging from the expected SQL behaviour)
Can you raise a bug in Druid’s issues so that we can track it. Thanks! (<https://github.com/apache/druid/issues/new?assignees=&amp;labels=&amp;template=other.md&amp;title=>)

if you remove the `group by` clause then because you are using the count aggregator you still get an implicit `group by ()` clause and it returns a single row. This is consistent with how other dbs behave. Here is MySQL for example <http://sqlfiddle.com/#!9/a6c585/292295>

Kanthraj regarding your original question you said "i just added the count()" you do not "just" add `count`  in you query adding an aggregate function (count), by the rules of SQL, adds an implicit `GROUP BY ()` that totally changes your query. In Q2 you do `count(num)` what are you trying to do? Because `count(num)` means give me a count of all rows where `num` is not `null` . Is that really what you want? Ususally when people use the `count` function they either do `COUNT(*)`- the count of rows or `COUNT(DISTINCT x)` the number of distinct x values.

I have experimented with this a bit.  The problem as I see it happens under some conditions when your subquery is a group by,  the native SQL also types the outer query as a "groupBy", which results in "select count(*) with group by" behavior, i.e. no rows returned.  The outer query here is not a group by, regardless of what the type of subquery is.

I ran some more variations to see what would happen:

```select count(*) from (select channel from wikipedia where channel = 'x') where channel = 'y'```
Returns one record with zero (0) count.

```select count(*) from (select channel, count(*) from wikipedia group by channel) where channel = 'y'

select count(*) from (select channel, count(*) from wikipedia where channel = 'x' group by channel) ```
Each of the above statements returns no records.  In this case the native outer query type is "groupBy".

Interestingly:
```select count(*) from (select channel, count(*) from wikipedia where channel = 'x' group by channel) where channel = 'y'```
Also correctly returns one record with zero (0) count.   The native outer query type is "scan" .  So even though the subquery is a group by, the addition of the filter condition in the subquery avoids the translation issue for the outer query.

i found a work around for this count(*) not getting displayed i just added a limit at the end i'm getting the result. This is not a correct solution but a workaround i found all of a sudden.

Hmmm ... I just tried these:

```select count(*) from (select channel, count(*) from wikipedia group by channel) where channel = 'y' LIMIT 10

select count(*) from (select channel, count(*) from wikipedia where channel = 'x' group by channel) LIMIT 10```
with no luck.

Please do an Explain Plan on your query that is working and let us know what the top level "queryType" is.

Thanks.  John

hi, just curious to know whether any LLVM used in druid for code generation on the fly from SQL query?

Could you structure the map as a json object? `{"key1":"value1","key2":value2"...}`
If so, then yes, Nested column ingestion will columnarize and index each key and make them accessible through JSON_VALUE functions.

Actually, just looked it up. Protobuf is not yet supported for nested column ingestion. As of 25.0 it supports JSON, Parquet, Avro, &amp; ORC.

I was looking at the code, specifically at <https://github.com/apache/druid/pull/13546>. the code seems to be suggesting that nested column is supported for protobuf.
I tried ingesting a protobuf map and tried storing it in a column with type json, using druid 25.0. It seems to be working fine so far.

That’s great! We’ll have to adjust the docs.

• it looks like totalNumMergeTasks default is 10 , For the task that merge shuffled segments, you can set the `totalNumMergeTasks` in the `tuningConfig`.
<https://druid.apache.org/docs/latest/ingestion/native-batch.html>

Hi <@U03QYUNNFCN> looks like it is not working and from the doc that parameter is only for Hashed or single dim , we are doing Range and multi dim partitioning

Seems like you might need to configure the  <https://druid.apache.org/docs/latest/development/extensions-contrib/prometheus.html|Prometheus Emitter> and use <https://learn.netdata.cloud/docs/data-collection/monitor-anything/Any%20Application%20-%20Generic%20Collectors/Prometheus-endpoints>

There is no max limit on the number of dimensions you can have. We have see people successfully working with druid in productions from a few to around a hundred dimensions .
Check below thread for more details.

<https://groups.google.com/g/druid-user/c/OgvnG3nVTNM?pli=1>

Thousands of columns is both fine and routine there is no hard limit but stuff will eventually start to degrade in performance as you add dimensions. What exactly fails first depends on way too many factors to list out. In SQL based ingestion there is a limit of 2000 columns you can generate per query (see <https://druid.apache.org/docs/latest/multi-stage-query/reference.html#error_TooManyColumns>) that limit was picked as a known safe limit. So consider &gt;2000 as being "at your own risk" (but some people do that)

thanks my columns are 7000 but works fine and query seems not that bad. i am just curious really thank you

You should find complete details in below two links:
Also i have attached one sample SPEC:

<https://druid.apache.org/docs/latest/ingestion/native-batch.html#parallel-indexing-example>
<https://druid.apache.org/docs/latest/ingestion/native-batch-input-sources.html#s3-input-source>

```
{
  "type": "index_parallel",
  "spec": {
    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "s3",
        "prefixes": [
          "s3://****/splittest/"
        ]
      },
      "inputFormat": {
        "type": "json"
      }
    },
    "tuningConfig": {
      "type": "index_parallel",
      "partitionsSpec": {
        "type": "dynamic",
        "maxRowsPerSegment": 1000,
        "maxTotalRows": 1000,
        "numShards": 5,
            "partitionsSpec": {
                "type": "hashed",
                "numShards": 5
            },
      }
    },
    "dataSchema": {
      "dataSource": "splittest",
      "timestampSpec": {
        "column": "timestamp",
        "format": "iso"
      },
      "dimensionsSpec": {
        "dimensions": [
          "isRobot",
          "channel",
          "flags",
          "isUnpatrolled",
          "page",
          "diffUrl",
          {
            "type": "long",
            "name": "added"
          },
          "comment",
          {
            "type": "long",
            "name": "commentLength"
          },
          "isNew",
          "isMinor",
          {
            "type": "long",
            "name": "delta"
          },
          "isAnonymous",
          "user",
          {
            "type": "long",
            "name": "deltaBucket"
          },
          {
            "type": "long",
            "name": "deleted"
          },
          "namespace",
          "cityName",
          "countryName",
          "regionIsoCode",
          "metroCode",
          "countryIsoCode",
          "regionName"
        ]
      },
      "granularitySpec": {
        "queryGranularity": "none",
        "rollup": false,
        "segmentGranularity": "day"
      }
    }
  }
}```


Hi team, will the `decommissioningNodes` array in coordinator config be auto cleared at a certain point. Like after the nodes in the array is killed?

Yes, this should be easily done using manual compaction/reindexing ... in the future a more comprehensive lifecycle management feature may help automate this in a more elegant manner.

Yes, a datasource can have mixed granularity segments in it. There are two things to note here: (1) you generally want segments for a certain granularity to represent data that starts and ends on that granularity. So the `month` segmented data should only start on a month boundary, as in you do not want to, say, reindex 53 days of data into month granularity as those would not align with month starts and end. (2) I am sure you have a good reason to do this but this comment is mostly for other people reading this: this approach of having less granular segments is not some "cheat code" to better performance. You might get better performance or you might do all the work to set it up and not get any perf gains. This is because Druid always wants to divide your data into ~5M rows per segment (or some number). So if you have day segments and you have 900M rows of data in 30 days in `day` segments you will have 900M / 5M ~= 180 segments, if you reindex to month you will have 900M / 5M ~= 180 segments - as you can see if your data is large enough you will still get the same amount of segments even if you go to a less granular segment granularity. This will only give you benefits if your data is small (but then do your really need to optimize it?) or if you couple the reindexing with other operations that reduce the data size.

i am also rolling up values as the data gets older (as i go from hour =&gt; day =&gt; month granularities), so there should be some benefit here.

Hi guys, I just found a pleasant article and wanted to share it. Adding support for Druid in Ibis was very easy and took only a couple of hours, thanks to pydruid and druid's docker support! (<https://voltrondata.com/resources/new-ibis-backend-shipped-in-four-hours-druid>)

You would have to orchestrate the timing, but yes you can do the aggregate rollup data creation with MSQ SQL:  <https://druid.apache.org/docs/latest/multi-stage-query/api.html>

You can do something like this: <https://druid.apache.org/docs/latest/data-management/automatic-compaction.html#set-frequency-of-compaction-runs>

You can also run a manual compaction for a longer time period, and it will process all time chunks within the time period one after another, only using the same amount of resources that a single auto-compaction uses (in your case it sounds like three at a time.)

<@U035FNDU6HH> If i change the `druid.coordinator.compaction.period` in the `coordinator/runtime.properties` file, do I need to restart the cluster?

you’ll also need
```druid.coordinator.dutyGroups=["compaction"]
druid.coordinator.compaction.duties=["compactSegments"]```
in order for `druid.coordinator.compaction.period` to take effect.

Do you have any IaC surrounding your deployment? There isn't any direct integration from Druid to read secrets manager secrets directly. It can read it from env variables if you have a way to get that set on the machine before launching druid.

Hi,
The lateMessageRejectionPeriod controls rejection of data before a certain period during streaming ingestion.
You likely need to configure your datasource's retention rules if the data got dropped earlier than expected -&gt; <https://druid.apache.org/docs/latest/operations/rule-configuration.html>

How long is the task duration? Have you seen any segments published to deep storage? 

Hey <@U030K4UM3H7> Task duration is 1 hour,
Sorry i'm new to druid so by events published to deep storage you mean, events showing in datasource tab correct?

Hey <@U0319HD8HEC>
I've updated the retention rules in datasource and earlier it was default 36 hours. So i've added a new rule for 90 days.
So hopefully this will work now!
Thanks for you suggestion!

Seems like the retention was the issue. I was just thinking that you may have been only looking at real-time "in-middle-manager" data and you may have had some issue at publishing time which made the data "disappear". Glad you found the answer!

I found <https://support.imply.io/hc/en-us/articles/4406982671251-Assigning-ingestion-jobs-to-run-in-a-specific-tier-of-middleManager-nodes-based-on-worker-category-in-Cloud-|this article> that explains how to do this using Overload Dynamic Config -&gt; Select strategy and an "equalDistributionWithCategorySpec"

Hi <@U030K4UM3H7>
Thanks for the article. We have already deployed similar configs

```{
  "type": "equalDistributionWithCategorySpec",
  "workerCategorySpec": {
    "categoryMap": {
      "single_phase_sub_task": {
        "defaultCategory": "subtask_category",
        "categoryAffinity": {}
      },
      "index": {
        --
      },
      "index_parallel": {
        --
      },
      "compact": {
        --
      },
      "kill": {
        --
      }
    },
    "strong": true
  }
}```
however now we need to distinguish between `single_phase_sub_task`  from compaction task and index_parallel_task. is there any way to do that?

Why not dedicate compaction into its own mm tier? 

ok if i dedicate compact to own tier will `single_phase_sub_task`  from compaction task will go to that tier

to my knowledge compaction is a simple long running process. It doesn’t spawn single_phase_sub_task

yes for us it spins subtasks as it has huge data

Putting compaction inside its own tier is how we do it on all of our clusters because its disk and memory requirements are very different than ingestion.

i am trying to use below config but some how jobs are in pending state for the second datasource  . we are using Druid 25 and ds2 datasource we are using different set of MM but when we submit a job it is in pending state and not getting assigned to any MM .
```{
  "type": "equalDistributionWithCategorySpec",
  "workerCategorySpec": {
    "categoryMap": {
      "index_parallel": {
        "defaultCategory": "_default_worker_category",
        "categoryAffinity": {
          "ds1": "ds1_batch",
          "ds2": "ds2_batch"
        }
      },
      "partial_dimension_cardinality": {
        "defaultCategory": "_default_worker_category",
        "categoryAffinity": {
          "ds1": "ds1_batch",
          "ds2": "ds2_batch"
        }
      },
      "partial_index_generate": {
        "defaultCategory": "_default_worker_category",
        "categoryAffinity": {
          "ds1": "ds1_batch",
          "ds2": "ds2_batch"
        }
      },
      "partial_index_generic_merge": {
        "defaultCategory": "_default_worker_category",
        "categoryAffinity": {
          "ds1": "ds1_batch",
          "ds2_perf": "ds2_batch"
        }
      },
      "partial_range_index_generate": {
        "defaultCategory": "_default_worker_category",
        "categoryAffinity": {
          "ds1": "ds1_batch"
        }
      },
      "partial_dimension_distribution": {
        "defaultCategory": "_default_worker_category",
        "categoryAffinity": {
          "ds1": "ds1_batch"
        }
      }
    },
    "strong": true
  }
}```


This issue is fixed after setting up `druid.version=0` tasks are assigned to right MM

If you want to ingest data from minIO, this thread might be helpful: <https://www.druidforum.org/t/fail-to-configure-data-ingestion-from-minio-object-storage/6783/11>

it is possible. We use minIO in our CI systems as well.

log.txt

Could you please share the overlord logs during this time as well? Also, what is the configured task duration?

If data is persisted to deep storage, it means that segments were successfully committed and that the offsets are stored in the metadata irrespective of the status of the supervisor / tasks

overlord-coordinator_log.txt

I have attached the coordinator log. Because in docker we are having only the container for coordinator but not for the overlord

The streaming ingestion supervisor runs on the overlord. Do you mean to say that you're running the overlord within the coordinator using `druid.coordinator.asOverlord.enabled=true` ?

Also, the logs seem to start at 11:37. It'd help to have the task logs and overlord logs from the same time period

it is using the server timings.. so it is showing the 11:37

Yes, I meant that the task log you attached is from 10:39 whereas the coordinator log is at 11:37. Could you please share logs for both the task and overlord during the same time window?

Strange, it appears that the overlord shut down the task but it's not clear why.
You may want to also see the error message within the task status for taskId; `index_kafka_kttmss_3578bdd37efd9dc_bllnefel`

log.txt is the complete log for the same task..

image.png

I'm referring to the status. You can view it using the web-console

Code 137 usually means an OOM. Could you please try reducing maxRowsInMemory for this supervisor? (Maybe to 100000)

Please refer to <https://druid.apache.org/docs/latest/operations/basic-cluster-tuning.html> and also go through your MM logs for more info if the above step doesn't help

ok.. let me try that..
and one more this.. as I mentioned earlier this deployment is on AWS ECS.. but I did a docker deployment on EC2 (t2.large), there everything is working fine. For ECS I have given 8vCPUs and 16 GB RAM, t2.large is 2vCPU and 8 GB RAM...

You have to set these server properties on different historicals, then it should show up correctly.

If both settings are in the same config file, the historical will only pick the latest config in the property file (based on your observations)

I used <https://druid.apache.org/docs/latest/tutorials/index.html> to set up local env and using start-micro-quickstart to run server

This doc mentions that the properties should be in historical/runtime.properties .. so I need to create folders under historical to create multiple runtime.properties ?
<https://druid.apache.org/docs/latest/operations/mixed-workloads.html#historical-tiering>

ah I see, the quickstart runs only 1 historical server. To test multiple tiers, you will need to run at least 2 historical servers, either locally, or elsewhere.

You can see the `bin/start-cluster-data-server` as an example script to start just the data server. The conf file for the historical that this script starts is in `conf/druid/cluster/data/historical`

I do not think there is a direct extension available but i found below some references, that may be helpful:

<https://stackoverflow.com/questions/64459885/how-to-read-druid-data-using-jdbc-driver-with-spark>

<https://groups.google.com/g/druid-user/c/92S0gYqUhvg?pli=1>

Hi Plaban, actually there is a an extension, we built it, you can find it on here: <https://github.com/deep-bi/spark-druid-segment-reader>

Thanks <@U032BN7D4Q6>,

I did look into the segment reader process, and I am also looking for below so that I can run some comparisons.
<https://github.com/apache/druid/blob/spark_druid_connector/docs/operations/spark.md>

Thanks <@U03R5H3B750>,
I am looking for a more internal integration rather than JDBC or HTTP.

Hi All, a question related to the *`query/failed/count`* metric. I have some queries that are failing validation due to `SqlPlanningException` . I am trying to understand if such (SQL validation) failures account for *`query/failed/count`*  metric. Would appreciate any help. Thanks!

Yes. Specific configs for deep storage on HDFS are in the<https://druid.apache.org/docs/latest/development/extensions-core/hdfs.html#configuration-for-hdfs| docs here>.

There is <https://github.com/grafadruid/druid-grafana>. They have a slack channel too I think.

MSQ SQL ingestion makes this fairly easy:

<https://docs.imply.io/latest/druid/multi-stage-query/>

You would basically issue a SQL statement like:

```INSERT into datasource2 
select * from datasource1 
 where __time &gt;= CURRENT_TIMESTAMP - INTERVAL '14' DAY 
   and __time &lt; CURRENT_TIMESTAMP - INTERVAL '7' DAY 
PARTITIONED BY ALL```
And if you need to replace data in datasource2 then you would use the REPLACE/OVERWRITE syntax

I'm not sure I have the syntax exact, but hopefully this gives you enough to work with.

Screenshot 2023-04-14 at 1.53.55 PM.png

The web-console also provides a guided experience to help with generating the re-index spec to load data from one datasource into another (see the `Reindex from Druid` tile)

<@U02VCGJFER5> but this does not give an option to reindex periodically correct

Job scheduler is a feature being discussed, but not in the product right now.  So for now you would have to submit jobs through the API and coordinate the scheduling your own (e.g. cron)

Fyi auto-compaction runs on a schedule, but that is specifically for single datasource compaction jobs, I don't know that you could jury-rig that to do something else (am guessing not).

This information helps. Thanks <@U04C4593VDY>