Apache Druid

I haven't tested this but I think you can do this with the `switching` request logger. Something like:
```druid.request.logging.type=switching
druid.request.logging.nativeQueryLogger	# don't specify, defaults to None
druid.request.logging.sqlQueryLogger=emitter```

<@U030K4UM3H7> Hi so i was able to configure it such that I have sql queries publishing to a kafka topic. However, it seems that it also logs internal sql queries that Druid uses when nagivating between pages in the UI. Is there a way to filter or differentiate between internal sql queries vs external sql queries (i.e. those run in the query tab)?

I didn't expect internal SQL queries, I thought all internal queries are native. Can you show me an example of such a query?

Yea so when logging sql queries, I get one that looks like this:

```SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, DATA_TYPE FROM INFORMATION_SCHEMA.COLUMNS```
I get this logged when switching to the "Query" tab in the UI.

I am also running this from the router. Is that perhaps why? If I put the configs for the broker, would that eliminate these queries?

Nevermind, it seems it still logs all of it from the broker side as well.

Yeah... Those are not really internal queries, at least not from the database's perspective. The Druid console behaves like another Druid client. But I agree that it would be good to filter those out.  Maybe something like `druid.request.logging.excludeSchemas=["information_schema","sys"]`
Do you want to create a github issue with the feature request? You can do that here: <https://github.com/apache/druid/issues/new/choose>

Ah I see. Yea I can create an issue.

I had one last question. Is it possible to put the native queries into one topic and the sql queries into another topic? I tried configuring with `switching` and setting the types of both the `nativeQueryLogger` and `sqlQueryLogger` to `{"type": "emitter", "feed": &lt;feed-name&gt;}` but it seems that it only logs the sql queries to the topic I created using the `kafka-emitter` extension. It logs the proper sql query and feed, but the native query is nowhere to be found. I feel like I'm missing some configs. Thank you!

That sounds interesting, I have not tried that yet, but would love to see what you come up with that works. Seems doable in principle.

Yea will definitely let you know. Also, I was curious if you knew. But do you know if `kafka-emitter` supports emitting with formats other than JSON (e.g. Thrift)?

I am also considering filtering on request logging. Have you had any luck?

I tried this but it didn't work for me :disappointed:
```druid.request.logging.type=switching
druid.request.logging.sqlQueryLogger=file
druid.request.logging.dir=log/
druid.request.logging.filePattern='request-'yyyy-MM-dd'.log'
druid.request.logging.durationToRetain=P7D```
I also tried this:
```druid.request.logging.type=filtered
druid.request.logging.sqlQueryTimeThresholdMs=5000  // Log requests that took longer than 5s
druid.request.logging.delegate.type=file
druid.request.logging.dir=log/
druid.request.logging.filePattern='request-'yyyy-MM-dd'.log'
druid.request.logging.durationToRetain=P7D```
But again no luck.

The only one I got working is this (but it logs WAAAAY too much):
```druid.request.logging.type=file
druid.request.logging.dir=log/
druid.request.logging.filePattern='request-'yyyy-MM-dd'.log'
druid.request.logging.durationToRetain=P7D```


can you check if you are page faulting a lot?

<@U04NHCE780G>, do you mean that if the memory is not enough large for all the mapped segment files (or put another way, all the files are mapped to the historical process) we start to see "thrashing" of paging in and paging  out?

i think the historicals download the active segments and mmap all of them, whether they are queried or not.  So the OS decides which pages get cached….so i would imagine if you see a lot of page faulting, you need to adjust your memory to disk ratio.

thx, that is what I suspect. It would be great if you can share some code pointers.

<@U04NHCE780G>, It took me some time to find the pgmajfault statistics. It actually turned out the why the `query/segment/time` is high, the page fault seems to be low. Though now that well correlated. Does this point to other forces that cause the query/segment/time to be high?

Screenshot 2023-06-12 at 7.05.39 PM.png

Screenshot 2023-06-12 at 7.06.42 PM.png

and this is the pgmajfault avg over all historicals:

You can try setting `druid.segmentCache.lazyLoadOnStart` to `true`. That will reduce the startup time a bit.

You could also set `druid.segmentCache.numBootstrapThreads` a bit higher instead. I have found setting it to the number of CPUs (including hyper threading) to be a pretty good value.

<@U031A9D19NU> <@U03TBGF0LTG> thank you. It's definitely faster now :grinning:

You can use this tutorial: <https://druid.apache.org/docs/latest/tutorials/tutorial-kafka.html>

yeah, Druid metrics are generic enough, as you could set it up with a graphite endpoint, http endpoint, etc. They don’t follow a specific monitoring tool’s naming guideline, such as Prometheus’s naming best practice you linked.

Afaik, currently, the `prometheus-emitter` itself doesn’t do any custom translation besides adding a few extra label dimensions to outbound metrics if configured. It _might_ be possible to add the desired formatting, let me know if you’ve some ideas.

Have you tried start-druid instead of run-druid?  I'm not sure of the differences (or why there are two commands), or whether it would work better for you.

start druid just seems to start all the services on a single node setup..
```$ /apache-druid/druid-26/bin/start-druid router
usage: start-druid [-h] [--memory MEMORY] [--services SERVICES]
                   [--config CONFIG] [--compute] [--verbose]
start-druid: error: unrecognized arguments: router```

I'm still not sure of the different uses intended between the two, but I believe it can do more.  See the examples in <https://github.com/implydata/druid/blob/master/examples/bin/start-druid-main.py> , eg.

Eg: `start-druid -m=100g -s=broker,router`
            Starts a broker and a router, using a total memory of 100GB.

i dont have these python scripts on the default 0.26 druid install. These are the only files available in bin:
```broker.sh
coordinator.sh
dsql
dsql-main
generate-example-metrics
greet
historical.sh
java-util
jconsole.sh
middleManager.sh
node.sh
overlord.sh
post-index-task
post-index-task-main
post-index-task-main3
run-druid
run-java
run-zk
service
start-cluster-data-server
start-cluster-master-no-zk-server
start-cluster-master-with-zk-server
start-cluster-query-server
start-druid
start-druid-main.py
start-micro-quickstart
start-nano-quickstart
start-single-server-large
start-single-server-medium
start-single-server-small
start-single-server-xlarge
supervise
verify-default-ports
verify-java```

Which python scripts?  I see start-druid in there, and start-druid-main.py.  Sorry if I'm missing something.

But you might be right, it might only be intended for quickstarts, idk.

It's documented under "<https://github.com/implydata/druid/blob/f5a3c5fa1d1b046ac295080a62da31090a04bbb6/docs/operations/single-server.md?plain=1#L25|single-server>" so it might not be intended for other uses.

I see what you mean though, there's broker.sh, coordinator.sh, but no router.sh.  For background, I guess you can use `&amp;` etc.  So I guess I don't have an better answer to your questions, sorry.

Yeah… thats what i been doing just nohup with &amp; but figured i would ask on here if there was a better solution :disappointed:

<@U053ZB6A7S9> I do see the following scripts (shell and python) in what you posted above:
```start-druid
start-druid-main.py```
I think what you’re missing is the `-s` argument to the shell script `start-druid` . Try:
```./apache-druid/druid-26/bin/start-druid -s router```

`start-druid` will look at available memory on the node where it starts (or the max you specify it can use) and split up the memory intelligently among the services you want to bring up. It tries to make the best use of the resources available to start the services you request.

only command that worked for me was run-druid router
also, it seems to ignore $DRUID_CONF_DIR path, since even if i specify another dir, it still looks for config in /conf directory

Hi! I have question while setting Historical.
I'm trying to set druid.cache.sizeInBytes and want to know if there are good ration between druid.cache.sizeInBytes and memory.
Default value written in document is min(1Gib, maxmemory / 10), is this just enough?

I don't think it is multiplied.  You need one merge buffer to run each group by query ... when the broker runs a query it sends a subquery to each historical and real-time ingestion task, and then merges the results it receives back.  So each node in question would just need one merge buffer to serve that query.

I'm not sure why a bit higher ... possibly there are Group By queries that the Broker can execute without Historical involvement, i.e. real-time segments only?

Seems like a hadoop issue.... I found this old conversation which seems relevant: <https://stackoverflow.com/questions/22138664/hbase-map-reduce-and-sequencefiles-mapred-output-format-class-is-incompatible>

How are you submitting this job? Can you share the load spec?
Have you tried<https://druid.apache.org/docs/latest/multi-stage-query/reference.html#sql-reference| SQL Based Ingestion> as an alternative?

Thanks for your inputs <@U030K4UM3H7>, I am submitting thru web console,
This is MR job error :
```Error: class com.fasterxml.jackson.datatype.guava.deser.HostAndPortDeserializer overrides final method deserialize.(Lcom/fasterxml/jackson/core/JsonParser;Lcom/fasterxml/jackson/databind/DeserializationContext;)Ljava/lang/Object;
Error: class com.fasterxml.jackson.datatype.guava.deser.HostAndPortDeserializer overrides final method deserialize.(Lcom/fasterxml/jackson/core/JsonParser;Lcom/fasterxml/jackson/databind/DeserializationContext;)Ljava/lang/Object;
Error: class com.fasterxml.jackson.datatype.guava.deser.HostAndPortDeserializer overrides final method deserialize.(Lcom/fasterxml/jackson/core/JsonParser;Lcom/fasterxml/jackson/databind/DeserializationContext;)Ljava/lang/Object;```

```org.apache.hadoop.io.serializer.SerializationFactory: Serialization class not found: 
java.lang.ClassNotFoundException: Class  org.apache.hadoop.io.serializer.avro.AvroReflectSerialization not found```

Was this working before? If so, what changed? The error indicates a missing jar or something that is not in the class path. I am not very familiar with the hadoop functionality, but could it be that you are missing a Druid extension?

Thanks for your inputs. Druid extension added

Hi! Imply offers a managed Druid solution from the original authors of Druid through a variety of deployment types! Would you like to be run through these options?

I'm afraid the customer also requires the data to remain on premise, so I'd need to use a self-hosted offering

Yes, we can offer that, quite a few of our customers have similar requirements. Would you like to arrange a call to discuss requirements and our offering?

Hi, here is link to the offering - <https://imply.io/imply-enterprise>

Please let me know if you have any questions

*June 23rd: Apache Druid Lunch and Learn in Atlanta, GA*
Hi All! If you are in the Atlanta area join us for an Apache Druid Lunch and Learn on Friday, June 23rd starting at 12pm. This is a great opportunity to network with other Druid community members.
The event is free to attend and local pizza and beer will be provided. Learn more and rsvp here: <https://imply.io/events/apache-druid-lunch-learn/>

Screenshot 2023-06-14 at 13.51.41.png

Screenshot 2023-06-14 at 13.22.29.png

You will not see data from an `index_parallel` job until it completes

Thanks Kyle! How much time would you guess that amount of data take to complete being inserted into datasource? How can I estimate?

Very difficult to estimate this. It can be affected by the organisation of the source files, any secondary partitioning chosen, number of dimensions, cardinality of dimensions… too much to give a simple answer

Okay, is there any chance where I can see the Progress or so? like how much data left to be consumed or so?

not with a native index_parallel - if you run with MSQ you get some progress bars / rate etc.

To echo Kyle, MSQ is superior to index_parallel in pretty much every way. One specific improvement is that it gives you much better feedback of how stuff is going and how long it will take. You can convert an index_parallel to a SQL (MSQ) with just a handful of clicks: <https://druid.apache.org/docs/latest/tutorials/tutorial-msq-convert-spec.html>

Thank you so much! Will have a look on that one! <@U030MBK46BD> <@U030W50CNAC>

Screenshot 2023-06-20 at 15.40.48.png

Screenshot 2023-06-20 at 15.41.03.png

Hey <@U030MBK46BD>, <@U030W50CNAC>

I do not see the same options as documentation shows unfortunately :confused: What can I do to enable "Convert ingestion spec to SQL" ?

Hmm - What version are you on? Might be time to talk to your team about upgrading. If I have it right, I think you’re on something like the Imply equivalent of Druid `0.21`

<@U030MBK46BD> We have Imply version 2022.01.6 LTS

Yes - this is too old to have MSQ - you need to upgrade

Hi Everyone! There’s a new <https://imply.io/podcasts/|Tales at Scale podcast> episode available! <@U0322ET4ZGE> walked me through Druid Operator. It’s worth a listen, especially if you’re looking to simply the management of Druid clusters in Kubernetes: <https://open.spotify.com/episode/5hYro5L98UlGomnpGbfx0g?si=a99fddbf58da4e59>

<https://druid.apache.org/docs/latest/operations/clean-metadata-store.html> -- Coordinator / runtime.properties

You have to configure this on the Coordinator.

image.png

Hi All ,

can someone suggest what could cause this issue
as it was working fine and suddenly stoped working

and all other with the same ingestion conf are working fine

The supervisors run on the Overlord. Could you please check from there (assuming it runs on a different host than the coordinator) and check its logs as well?

<@U0319HD8HEC> could it be because of resources

I'm not sure. Could you please check the supervisor status from the web-console?

The magnifying glass icon beside the supervisor in the actions column

This is what is see

```{
  "dataSource": "eber_gateways_sensors_data",
  "stream": "eber.gw.sensors.data",
  "partitions": 3,
  "replicas": 1,
  "durationSeconds": 604800,
  "activeTasks": [],
  "publishingTasks": [],
  "latestOffsets": {
    "0": 1982944,
    "1": 1982593,
    "2": 1982209
  },
  "minimumLag": {
    "0": 347306,
    "1": 347624,
    "2": 347179
  },
  "aggregateLag": 1042109,
  "offsetsLastUpdated": "2023-06-16T08:56:02.877Z",
  "suspended": false,
  "healthy": false,
  "state": "UNHEALTHY_SUPERVISOR",
  "detailedState": "UNABLE_TO_CONNECT_TO_STREAM",
  "recentErrors": [
    {
      "timestamp": "2023-06-16T08:54:47.582Z",
      "exceptionClass": "org.apache.druid.java.util.common.ISE",
      "message": "org.apache.druid.java.util.common.ISE: Previous sequenceNumber [1635638] is no longer available for partition [0]. You can clear the previous sequenceNumber and start reading from a valid message by using the supervisor's reset API.",
      "streamException": true
    },
    {
      "timestamp": "2023-06-16T08:55:17.585Z",
      "exceptionClass": "org.apache.druid.java.util.common.ISE",
      "message": "org.apache.druid.java.util.common.ISE: Previous sequenceNumber [1635638] is no longer available for partition [0]. You can clear the previous sequenceNumber and start reading from a valid message by using the supervisor's reset API.",
      "streamException": true
    },
    {
      "timestamp": "2023-06-16T08:55:47.587Z",
      "exceptionClass": "org.apache.druid.java.util.common.ISE",
      "message": "org.apache.druid.java.util.common.ISE: Previous sequenceNumber [1635638] is no longer available for partition [0]. You can clear the previous sequenceNumber and start reading from a valid message by using the supervisor's reset API.",
      "streamException": true
    }
  ]
}```

It appears that the topic offsets stored in Druid's metadata are no longer available in the Kafka stream.

Was the lag continuously increasing for the supervisor prior to this? Or perhaps the supervisor was left in a suspended state for a long time?

the topic is there in kafka and receiving data

yup it was for someother datasource to but i re-submited the ingestion conf which resolve the issue

Yes, it just means that the stored offset (`1635638`) is not in the retention period of the topic.

You could reset the offsets, but that could lead to data loss

You can resubmit the supervisor to read from the latest checkpoint if that is ok

resubmit the supervisor how can i do this. ??

if i increase the retention period of this topic , i think that will fix the issue

&gt; if i increase the retention period of this topic , i think that will fix the issue
Yes, that might help prevent this issue

You could try a hard reset if you want to continue reading from latest offset

If you want to read from the earliest available offset, you may have to set `useEarliestOffset`  to true in the spec as well.
<https://druid.apache.org/docs/latest/development/extensions-core/kafka-supervisor-reference.html#kafkasupervisorioconfig>

```{
  "dataSource": "eber_gateways_sensors_data",
  "stream": "eber.gw.sensors.data",
  "partitions": 3,
  "replicas": 1,
  "durationSeconds": 604800,
  "activeTasks": [],
  "publishingTasks": [],
  "latestOffsets": {
    "0": 1985770,
    "1": 1985330,
    "2": 1984909
  },
  "minimumLag": {
    "0": 350132,
    "1": 350361,
    "2": 349879
  },
  "aggregateLag": 1050372,
  "offsetsLastUpdated": "2023-06-16T11:12:15.269Z",
  "suspended": false,
  "healthy": false,
  "state": "UNHEALTHY_SUPERVISOR",
  "detailedState": "UNABLE_TO_CONNECT_TO_STREAM",
  "recentErrors": [
    {
      "timestamp": "2023-06-16T11:11:00.260Z",
      "exceptionClass": "org.apache.druid.java.util.common.ISE",
      "message": "org.apache.druid.java.util.common.ISE: Previous sequenceNumber [1635638] is no longer available for partition [0]. You can clear the previous sequenceNumber and start reading from a valid message by using the supervisor's reset API.",
      "streamException": true
    },
    {
      "timestamp": "2023-06-16T11:11:30.260Z",
      "exceptionClass": "org.apache.druid.java.util.common.ISE",
      "message": "org.apache.druid.java.util.common.ISE: Previous sequenceNumber [1635638] is no longer available for partition [0]. You can clear the previous sequenceNumber and start reading from a valid message by using the supervisor's reset API.",
      "streamException": true
    },
    {
      "timestamp": "2023-06-16T11:12:00.263Z",
      "exceptionClass": "org.apache.druid.java.util.common.ISE",
      "message": "org.apache.druid.java.util.common.ISE: Previous sequenceNumber [1635638] is no longer available for partition [0]. You can clear the previous sequenceNumber and start reading from a valid message by using the supervisor's reset API.",
      "streamException": true
    }
  ]
}```
ive increased the number of retnetion to 30 days from 7

That might help prevent this from happening in the future. Right now, the supervisor offsets need to be reset after suspending the supervisor

After the hard reset, resubmit the supervisor with `useEarliestOffset`  set to `true` if you want to ingest all the available data.

but it says ill lost all my data , is there any other way to do soo

You won't lose any exisiting or available data. Only the data which was not ingested and is no longer available due to retention period having passed

You can reingest that data if you have it

ive suspended it should i hardreset it now. ?

thats what you are suggesting if im correct or just suspend it and resume it again

Just want to reiterate that the data that was not ingested and was past retention period won't be available. You may have to reingest it

Hi All. I have a druid cluster running in Kubernetes and I am trying to setup another cluster in the same kubernetes space. I have changed all the names and added suffix -uat for example coordinator is not coordinator-uat, broker is now broker-uat etc. Web-UI gave error attached here. I understand the router was looking service /druid/broker so i changed the configuration as
```druid.service: broker-uat```
still the same error. My questions are: 1) should I give druid/broker-uat or just broker-uat? 2) Router will not only look for broker but for also coordinator and other services, how can I override the default service names for all? 3) I think i would need to make this change for all the services. I mean for coordinator I need to give all services with -uat and similarly for MM and Historical and broker? Thanks

I believe this is as per design and you should follow the suggestion documented in guide:
<https://druid.apache.org/docs/latest/querying/scan-query.html#time-ordering>

Thanks, but I didn't get it, i have 13 segments but the error is saying 51 segments per time chunk and today i got another error saying 563 segments ,on the same table

Generally the segment count is picked from the metadata table, please check in the metadata table druid-segments and druid-pending segments table. May be delete the unused segments and retry.

okay thanks, how to check in the metadata.? can you just give a brief .?

Login to your metadata database, describe the table and see the columns present and accordingly you can form your query to check.
something like: select dataSource,  used, count(*) from druid_segments group by dataSource, used ;  used=0 means segments are unused and can be deleted.

The data loader wizard is having an issue the `__time` column in your input data because it is formatted as `iso` (`2023-06-16T10:33:01` ) rather than millis (`1686936781000`) as would be natural if this data was exported from Druid you have 3 options:
1. Rename the `__time` column in your data to literally anything else
2. Change the type of the `__time` column in your data to millis
3. Write the spec by hand, it will still work if your `timestampSpec` as `{"column": "__time","format": "iso"}`

No `IF` but you might be able to use `CASE`<https://druid.apache.org/docs/latest/querying/sql-functions.html#case>

Hi Shantha,

Shuffle join is just join ... you have to set the query context though in order for it to be used.

You can run select statements directly in the Query tab of the web console, or you can submit them through the MSQ task API ... although for the latter you have to retrieve the results in a separate API call.

Here's an example join on the kttm demo database ... it's self-joining but if you try to run this with the native engine it will give you the familiar `subquery max[100000]` error:

```select a.session, 
       count(b.session) numMatches, 
       count(distinct b.client_ip) numIPs
  from "kttm-nested-v2-2019-08-25" a 
  join "kttm-nested-v2-2019-08-25" b on a.session = b.session
 group by a.session```

To elaborate on that, shuffle join only exists within MSQ. Using MSQ for ingestion (including shuffle joins) is a fully supported good to go feature. Using MSQ for querying (i.e. running queries in MSQ that return results to you rather than write them to a datasource) is experimental and the "interactive" (result getting) API is definitely going to change in the future (see: <https://github.com/apache/druid/pull/14416>) when that change goes in (hopefully in Druid 27) it will also be a fully supported non-experimental feature. For today (Druid 26) you are welcome to use shuffle joins for ingestion use-cases, you can have a play around with the querying but do not build workflows around it just yet - I would advise to stick to running interactive queries from the console which will update in tandem with the APIs.

You can select shuffle joins (aka sortMerge joins) via the UI like so: (it just sets a context flag)

Thank you for the detailed explanation! For the ingestion MSQ usecase, do the query_worker nodes talk to each other for shuffling? While using MSQ for querying, does it still use the brokers and historicals for the query or are query_controller and query_worker tasks used for that as well?

Yes the query_worker tasks talk to each other for shuffling. When using MSQ (with the current APIs) you do not engage with the broker or historical at all (this will likely change in the future)

this is really cool work! Excited for the future :slightly_smiling_face:

I have no idea what "enhanced fan-out" is but could you elaborate on the usecase that you have where you are reading a single stream into 6 datasources? Are they all differently filtered somehow? Differently rolled-up?

They have different roll-ups and column order to optimise for specific types of query

Hi Saurabh,

I don't know the exact names of the metrics  to look at, but at a high level you can start by checking the following:
• Broker level 98% query response time, vs Broker level subquery time and subquery TTFB, will tell you the latency split between Broker and data nodes
• Compute level query time, segment scan time and # segments scanned
• If you have real-time streaming ingestion, then for the Compute metrics break out Peon from Historical so you can see if there is a noticeable difference between real-time vs historical segment scanning performance. 

Hi Team, Any updates on this thread. Pls suggest I am doing the poc

I've dealt with the APPLY functions briefly before ... I think the equivalent would be a correlated subquery, either in the select list, or as a CTE.

Can you provide the (or one example of a) SQL statement with the Apply clause that you would like to translate?

Kusto function mv-apply
====================

Table1
| where True | mv-apply r=col1 on (where r.location in ("INDIA") | summarize new-col1=make_list(r))
| project col1,new-col1,xcol1

Screenshot 2023-06-20 at 16.24.00.png

You need to use range, single_dim or hash partitioning to achieve perfect rollup. If you use MSQ <https://druid.apache.org/docs/latest/multi-stage-query/reference.html#sql-reference|SQL-Based Ingestion> you can do range partitioning with a REPLACE statement that does aggregation with the SELECT/GROUP BY and the PARTITION BY &amp; CLUSTERED BY clauses to specify time partitioning and secondary partitioning respectively.

In this case, if you are using an index_parallel spec, look for the config `forceGuaranteedRollup` in the `tuningConfig` and set it to false. That will allow ingestion to run, and then you can use compaction to compact the data to get optimal rollup

As Sergio suggested, +1 for using MSQ if you are on a Druid version that supports it. It's much nicer, and there's a lot less tuning like this

Thank you for replies! <@U030K4UM3H7> <@U02VCGJFER5>

so in my spec , I have "rollup": true, in "granularitySpec": { section , and "forceGuaranteedRollup": true in tuningConfig as you said

what is the difference between those? I for sure want to enable rollups during ingestion time when creating segments, because we have some metric columns also specified in metricSpec, counts, doubleSum etc.

would it be okay if I set "forceGuaranteedRollup": to false and keep the rollup=true in granularitySpec? would it still do the job for me and rollup during ingestion?

we are feeding our datasource via kafka supervisor and it is set to fully compacted. after I run index_parallel, it would then do the job of ""forceGuaranteedRollup":true" yeah?

this is how my granularitySpec looks like in index_parallel

"granularitySpec": {
        "type": "uniform",
        "segmentGranularity": "DAY",
        "queryGranularity": "HOUR",
        "rollup": true,
        "intervals": ["2023-06-14T00:00:00.000Z/2023-06-15T23:59:59.999Z"]
      },

&gt; would it be okay if I set "forceGuaranteedRollup": to false and keep the rollup=true in granularitySpec? would it still do the job for me and rollup during ingestion?
Yes, it is ok to set forceGuaranteedRollup to false, and rollup to true. What this means is that data is rolled up within an individual segment, but not across segments

If you want the data to be perfectly rolled up, you can change the partitioning spec to hashed or range (I think it's called single_dim on older versions of Druid), this will tell the ingestion spec to shuffle the data so that they are clustered more efficiently than in dynamic partitioning

&gt; we are feeding our datasource via kafka supervisor and it is set to fully compacted. after I run index_parallel, it would then do the job of ""forceGuaranteedRollup":true" yeah?
I don't understand this. Are you using the index_parallel job to compact the data, or do you have auto-compaction scheduled for the datasource and this index_parallel job is just some backfill?

Hi Siddharth,  you can definitely load up the content on all of your CSVs into the same datasource (table) to be accessed as a single set of data ... is that what you were referring to?

Hi <@U05AFP2RSPQ>, you can ingest multiple csv files for one ingestion the `EXTERN` function:
<https://druid.apache.org/docs/latest/multi-stage-query/concepts.html#read-external-data-with-extern>

List the files as part of the `uris`  For example, for S3:
``` "uris": ["<s3://foo/bar/file.csv>", "<s3://bar/foo/file2.csv>"]```

I think this is already supported if you use the `kafka` Input Format. You can definitely get headers and stuff. I don't think it will automatically use the kafka timestamp as the timestamp field, but you could probably make that work with a transformspec. <https://druid.apache.org/docs/latest/development/extensions-core/kafka-ingestion.html#kafka-input-format-supervisor-spec-example>

:point_up_2: and AFAICT you can use it as the primary timestamp by referencing it with `kafka.timestamp`

Oh man, I scoured the documentation for this and couldn't find it. Thanks for the direct link!

I've been using `index_parallel` like a sucker!

:grin: It is fairly new, 0.23 or 24, if I remember correctly.

I'm using 0.23.

I replaced:
```      "inputFormat": {
        "type": "json"
      }```
With the example from the documentation:
```      "inputFormat": {
        "type": "kafka",
        "headerLabelPrefix": "kafka.header.",
        "timestampColumnName": "kafka.timestamp",
        "keyColumnName": "kafka.key",
        "headerFormat": {
          "type": "string"
        },
        "keyFormat": {
          "type": "json"
        },
        "valueFormat": {
          "type": "json"
        },
        "findColumnsFromHeader": false
      }```
Unfortunately when I do this it breaks the UI in 0.23 (it doesn't know how to parse it as kafka format). I also notice kafka is not a valid Input format on the "Parse data" tab in 0.23.

I pushed it through but I just get errors:
```org.apache.druid.java.util.common.parsers.ParseException: Unable to parse row [fb5d5708-6a51-4334-afd7-217e1f61ecd2] (Record: 1, Line: 1)```
So I might have to upgrade my version of Druid before I try this again :disappointed:

based on the release notes it looks like support was added for the kafka format in the web console in 26.0.0. it probably still works if you can submit to the task endpoint

I think this is the relevant PR (which was merged into 0.23.0) <https://github.com/apache/druid/pull/11630>, so you should be able to submit the spec programmatically, but not through the UI which caught up later.

How many are you able to ingest now?

Whats the size of the records?

Are you splitting the ingestion into multiple tasks?

Do your middle managers use SSD disks?

We able to achieve the ingestion rate 130k/sec.
Number records 200 billion.
We have splitted the ingestion into multiple tasks.
I don't think using ssd disks for middle manager but the druid cluster deployed in k8s.

use mm-less and increase partitions.  spread the load