Apache Druid

and lookup's are loaded per peon task as well, it's not shared between the middlemanager in the same container/master

You can speed up lookup loading by specifying more loading thread <https://druid.apache.org/docs/latest/querying/lookups.html|https://druid.apache.org/docs/latest/querying/lookups.html>

• historical node restart, non-persistent volumes causes slow start -&gt; probably needs more replicas to be able to restart/lose pods without having unavailable segments
• all historical has disk full? 
• historicals crashing due to OOM
• missconfigured _tiers and not enough servers for all the given tiers
• 

Also sometimes segment will be not loaded  and might be deleted  from deep storage. You can figure out the segment name from historical logs  and check in deep storage

i recently i had an issue with that i want to avoid this issue!

can you elaborate a bit more? How do you plan to use druid for this use case?

i dont know what have to use...how did you do it before

I think <@U031A9D19NU> is asking about what real-time matching engine entails. What event data is there and what are you looking to do with the data. In the Taxi case, is this about matching fleet availability to demand? 

situation is fuzzy in reality and so the docs are fuzzy too :slightly_smiling_face:

basically, if you want the best performance and your disks aren't super fast, you want to avoid hitting disk

Druid uses columnar storage and memory mapping to minimize disk reads, so you just need to carve out enough free memory for that to work

the amount you need depends on your working set— how much of your data do you expect to be involved in queries at any given slice of time?

the main things to think about there are how your data is partitioned (by time), clustered/secondary-partitioned (by some columns), and how many datasources (tables) you have

also how many columns you have &amp; what proportion of columns are likely to be active for queries in any given slice of time

as you can see: fuzzy :slightly_smiling_face:

generally, though, bottom line is 10% to 40% tends to work well for most people. although i have seen cases that require 100% and cases that are OK even on, like, 1%. so I'd advise keeping an open mind

it's always nice to deploy in a cloud environment so you can adjust to what hardware is best for your particular situaton

Thanks for your answer. `the higher the better` :wink: Yes I agree... but is there any abacus on that ? We have to deploy some Druid Clusters for many customers and need to know how much resources we need...
is a ratio of 1/20, 1/10, 1/5... is good or not... I don't ask for precise figures but based on exepriences maybe some of you have best practices...

:+1: agreed experiences with specific values of this would be interesting to hear. Anybody have stories to share?

just noticed this thread— I mentioned in the first thread that in my experience 10% to 40% tends to work well for most people (although there are certainly situations where it is outside than range!)

additional thoughts in <https://apachedruidworkspace.slack.com/archives/C0303FDCZEZ/p1679982190186529?thread_ts=1679490311.673789&amp;cid=C0303FDCZEZ>

My datasource is from Kafka latest version

Hi, I do not understand the issue. Is it possible that you have both 42,949,644 and 6.36 as values in different rows for the `br`  column?

Is the issue that you would expect the Druid UI to render `42,949,644` as `42,949,644.0` ?

No In the same column I got a different type of data. sometimes it’s 6.36 and sometimes a number like 42,949,644.

What was it showing to you before the update? (I still do not understand the issue)

alternatively, if the UI showed you `42,949,644.00` and `6.36` would that match what you expect?

yes, you are correct but you do not have to limit yourself to one aggregator. say you have a field called `age` and you want to be able to compute `min`, `max`, and `avg` later on...

well, for the `avg (arithmetic mean)` you kinda can if you also add a new metric for `count` and `sum` but yeah, it's not possible to get avg directly from the rollup. you can, however, make good assumptions, using thetasketch

you would define 3 metrics `min_age` , `max_age` and `sum_age`. Then you could do `MIN(min_age) as min_age` , `MAX(max_age) as max_age` , `SUM(sum_age) / SUM(count) AS avg_age`

you can use theta-sketch, then use DS_QUANTILES_SKETCH to get some quantiles back

note that I am assuming that you are also ingesting a `count` metric (you would be crazy not to do it with rollup)

yeah, I think once you enable rollup, the count will appear in the metrics list

if you are using the console wizard then it will put count in the the metrics list when you click to enable rollup, but if you are not using the console then you should make sure to add it yourself

I see. Yes, you are right, we are using the console wizard. Thank you for the explanation!

image.png

Nice. Please remember that the wizard is just a tool to help you! It has a handy shortcut to convert a dimension to a metric but that will only create 1 metric. if you want 3 like I outlined above you have to click `Add metric` and add it your self.

<https://druid.apache.org/docs/latest/ingestion/data-formats.html#protobuf>

Screenshot 2023-03-22 at 6.51.51 PM.png

Screenshot 2023-03-22 at 6.56.14 PM.png

Screen Shot 2023-03-22 at 7.13.12 PM.png

you have to use JSON_VALUE() to pull out individual values ... then they can be treated as individual fields ... is this what you were looking for?

Fyi if you are talking about JSON array elements, that is a different story ... currently they are stored individually like separate fields ... but true handling of such elements as arrays is being worked on ...

Not really, I want to get the grouping of keys like in the screenshot

Screen Shot 2023-03-22 at 7.47.59 PM.png

So is it JSON_PATHS() that you are looking for?

`JSON_KEYS`  return an array, and you want to make it a multi-value dimension (those group as you desire). What you need is some function that takes an array and gives you a MV. I am not sure that a direct function like that exists <@U030C8H59T8> can tell me if I missed it. As a workaround you could join the array into a string and then split the string into an MV. Try `STRING_TO_MV(MV_TO_STRING(JSON_KEYS("details", '$.'), '#'), '#')`

Well good news, shuffle joins are <https://github.com/apache/druid/pull/13506|coming in Druid 26>. Until then, for your specific use-case, could you compare earliest(Sal) with latest(Sal) when grouping on Id?

will you have only two rows for each id? You could compute a max and min for sal grouped by id and in the outer query compare the max and min and set the flag something like
```select id,case when maxsal!=minsal then 1 else 0 end from (select id,max(sal) maxsal,min(sal) minsal from table group by id)```

what do you mean? Druid will stream the response, you can request either by CSV or JSONL so you can parse each row without waiting for the EOL

<https://druid.apache.org/docs/latest/querying/sql-api.html#responses>

i am fetching data from druid using python but after every 5 min it throws and error and only able to fetch 1 million records but actual data present is more than 100 times than this .

try passing in the context query a new timeout (or no timeout) <https://druid.apache.org/docs/latest/querying/sql-query-context.html>

Query timeout in millis, beyond which unfinished queries will be cancelled. 0 timeout means `no timeout` (up to the server-side maximum query timeout, `druid.server.http.maxQueryTimeout`). To set the default timeout and maximum timeout, see <https://druid.apache.org/docs/latest/configuration/index.html#broker|Broker configuration>

Pass some positive value, the broker has a default of 5min, so you need to pass a value greater than this

If you are referring to editing the data within a segment -- segments in Druid are immutable (cannot be modified), but there are ways to edit data within segments by reingesting modified data on top of the existing segment, in whole or in part.

If you are referring to changing the datasource structure to have segments of different granularity -- that is done through reindexing or compaction.

Information on both of the above can be found in the docs:

• <https://druid.apache.org/docs/latest/data-management/update.html>
• <https://druid.apache.org/docs/latest/data-management/compaction.html>
What specifically would you like to do?

thanks
specifically speaking I want to edit the configuration of segments of my existing table so that from now on the segments are formed properly..
now the segments are being formed haphazardly and it is causing issue while running queries

Ingestion will generally result in segment fragmentation, i.e. multiple segments for the same time chunk, often smaller than ideal size.  This is due to the way the parallel ingestion processes work, and also the fact that the data coming in can have __time values spanning multiple time chunks.  Compaction generally cleans all that up.

The simplest way to get compaction going is to turn on auto-compaction on the datasource.  That will give you a good first pass at cleaning up the segment fragmentation due to ingestion.  There are other steps you can take beyond that to tune the system for better storage optimization and query performance, but just setting up default auto-compaction is usually a first good step.

Hello John, I have 1 question to learn, Doing compaction will reduce the performance of Querying data from an application &amp; Ingestion ?  Example case, If I have 10 data-sources and each have approx., 5 million records/rows already with 'n' number of segments, will it work without disturbing the other Queries / Ingestions ?   (am using single-server configuration, even-though if we increase the worker-capacity i.e. service slots)

Thanks to Viraj ! Thanks for his question !

John, 1 more question, Why apache-druid not provided that compaction feature as enabled by default ?  why suppose to manually enabled ?

It is set up to operate with a limited set of resources (something like 10% of the task slots by default.)

It also can have locking contention with ingestion, but there is an "offset" you can set so that compaction starts with older time chunks that are not currently being ingested into.  So for example if you get data in that has timestamps throughout the day, then a common setting for auto-compaction is to have a "P1D" offset so it will only compact yesterday's and older data.

There are still a few caveats here based on what data is coming in -- read up on lateMessageRejectionPeriod and earlyMessageRejectionPeriod parameters ... this is perhaps one of the reasons why this is not enabled by default.   That being said, there are some enhancements in the works to all but eliminate lock contention between compaction and ingestion.

Thanks <@U04C4593VDY>
I wasn't aware bout this segment fragmentation, can you shed some light on this ,  I didn't reall understood what exactly is segment fragmentation so later on I can understand about auto compaction that you mentioned

<https://druid.apache.org/docs/latest/data-management/compaction.html> : Each peon or worker task on Middle Managers will generate and publish segments async to deep storage. Ideally, each segment should reach 5M or 300-700MB but it’s not always the case. Hence the need for compaction

<@U04Q25Z05U0> - I do not see a native druid option to update the coordinator runtime.properties.

The runtime.properties can be viewed under /status/properties
<https://druid.apache.org/docs/latest/operations/api-reference.html#process-information>

The dynamic config settings for the coordinator are available in HTTP for viewing or updating:
<https://druid.apache.org/docs/latest/configuration/index.html#dynamic-configuration>

Other applications allow for updates of the static config. If you would like some example let me know.

Thanks for reply <@U03S0JXTM4N>

So with that GET request /status/properties, what exactly I retrieve? Will it also contain Coordinator static properties?

It is just because I would like to change it set also below configs to there:

druid.coordinator.kill.on = true #turn on periodic kill tasks
druid.coordinator.kill.period # how often to run kill tasks defaults to P1D

Can I make a POST request updating the configs like that?
And If I make the changes there and POST if possible, will Druid be able to run according to new configs automatically?

<https://druid.apache.org/docs/latest/operations/mixed-workloads.html#historical-tiering>

Hmm, how would this work, exactly? In the example provided in the documentation both tiers are hot: the nodes have to have all the data loaded locally, it can't just pull from deep storage when the query demands it.

If I wanted to move data to S3 for example, can I have the different tiers using different deep storages? How would the cold tier pull data out of the hot tier?

I can't just put deep storage on S3 for the hot tier because it wouldn't be fast enough for Druid's replication from broker to historical...

Is this sort of thing possible?
```{
  "type" : "loadByPeriod",
  "period" : "P3M",
  "includeFuture" : true,
  "tieredReplicants": {
      "hot": 1
  }
},
{
  "type" : "loadForever",
  "tieredReplicants": {
      "cold": 1
  }
}```
Will the hot tier above only load the latest 3 month and drop anything older than that (I noticed drop rules can't apply to specific tiers)....    When Druid adds the default loadForever rule to the bottom of the list, is it only for the default tier?

Or perhaps, since I'm adding a new cold tier, the hot tier is actually just my `_default_tier`.

To update; we tried this and it is working wonderfully.  We put two new data nodes into a "cold" tier that are much smaller and have access to cheaper disk array.

Thanks!

I didn't understand, actually what I want to do is to send the data from posgtreSQL to Apache Druid but I have some databases and I want to load them one by one

Hi Hircham, you should be able to load your postgesSQL tables into Druid in the same way as MySQL tables are loaded in <https://medium.com/@kyle.hoondert/loading-mysql-tables-into-apache-druid-28d6638f66a8|this> blog post
You might even be able to do it using the new MSQ ingestion like:
```INSERT INTO some_datasource
SELECT * FROM TABLE(
  EXTERN(
    '{
      "type":"sql",
      "database":{
        "type":"postgresql",
        "connectorConfig":{
          "connectURI":"jdbc:<postgresql://some-rds-host.us-west-1.rds.amazonaws.com:5432/druid>",
          "user": "admin",
          "password": "password"
        }
      },
      "sqls": [
        "SELECT * FROM some_table"
        ]
    }',
    '{"type":"json"}',
    '[
        {"name":"column1","type":"string"},
        {"name":"column2","type":"string"}
     ]'
  )
)```

Thank you, But I have a lot of databases, each database has a lot of tables which means it's gonna take much time to migrate them all, I want an approach that is able to load all databases at once

Hi <@U0505JZT318>, this will probably not be the case, there will be multiple uploads

are you referring to multiple tables currently is Postgress, or also different tables  stored in multiple DBs

Depending on the answer, you might want to relook at the data model to optimize for Druid, which will be better than just dumping all your existing data from postgress to druid

Hi <@U032M071H7X> I hope you're doing well,
Thanks for the answer, Now I see that I have to do multiple uploads so right now I'm reffering to upload one database from postgreSQL to Aapache Druid if I do this task the rest will be easy But I didn't find any way to do it
Do you have any suggestions ?

You are going to want to extract your tables from Postgress in files, and then ingest them into Druid.

there are best practices on the size of the files that you will ingest, for the upload to be performant

<https://druid.apache.org/docs/latest/ingestion/native-batch.html>

But in my case I have a lot of data each table has hundreds of columns and I have a lot of databases which means it's gonna take much effort to do it, is there any other approach, for example a script in python that can load the whole database to druid ?

not to my knowledge, what do you think <@U030MBK46BD>

I don’t know of any tools that would help with this. My thoughts are: better get scripting…

what would your `key` and `value` be in this case?

Key would be string and values would be some computes quantilesDoublesSketch - A result of `groupBy`  query is going to be used in a lookup table for quicker access.

Or is there a way to cast `string`  type to `quantilesDoublesSketch` ? I can store sketches as string in lookup table and when doing query - I can cast them to ``quantilesDoublesSketch``

group by query will not give you a sketch. It will get you a number

```{
  "queryType": "groupBy",
  "dataSource": "table_name",
  "granularity": "all",
  "dimensions": [
    "column_name"
  ],
  "aggregations": [
    {
      "type": "quantilesDoublesSketch",
      "fieldName": "double_sketch",
      "k": 256,
      "name": "baseline_quantile_sketch"
    }
  ],
  "postAggregations": [
    {
      "type": "fieldAccess",
      "fieldName": "baseline_quantile_sketch",
      "name": "baseline_quantile_sketch_raw"
    }
  ],
  "filter": {
    "type": "and",
    "fields": [
      {
        "type": "selector",
        "dimension": "__model_id__",
        "value": "079c6aa9-2027-4c0e-8fbf-c3c0817514d1"
      },
      {
        "type": "in",
        "dimension": "column_name",
        "values": [
          "area",
          "baths",
          "beds",
          "datePriceChanged"
        ]
      }
    ]
  },
  "intervals": "2015-08-28/2025-10-20"
}```
This is my query, intension is to store `column_name` as key and `baseline_quantile_sketch_raw`  as value.

I’m able to store `baseline_quantile_sketch_raw` as the value but in string datatype - wanted to see if this can be stored as sketch type?

&gt; group by query will not give you a sketch. It will get you a number
<@U03BJAMQ4H5> - Couldn’t really follow your comment, can you see the query I posted above..

if you just use an aggregation then you can get the sketch. The post aggregation will result in a number. I think what yo are trying to do is us a rilled table and join with another table. As far as I know this can't be done using a lookup. You need a join.

Is there a way that I can cast string type to sketch type and store/retrive sketch from the lookup table?

<@U04QV0CU7V3> Is there a quick way to verify?

```SELECT
  TIME_FLOOR(CASE WHEN CAST("t_hour" AS BIGINT) &gt; 0 THEN MILLIS_TO_TIMESTAMP(CAST("t_hour" AS BIGINT)) ELSE TIME_PARSE("t_hour") END, 'PT1H') AS __time,
  "channel",
  SUM("__count") AS "__count",
  DS_THETA(COMPLEX_DECODE_BASE64('thetaSketch', "theta_user")) AS "theta_user",
  DS_HLL(COMPLEX_DECODE_BASE64('HLLSketch', "hll_user")) AS "hll_user"
FROM "source"
GROUP BY 1, 2
PARTITIONED BY DAY```
May be using `COMPLEX_DECODE_BASE64`  with `DS_THETA`  or something similar should do the Job. Not sure though..

As far as I know you cannot go from string to sketch (sketch to string is possible as the sketches output string in the post agg for. Debugging 

network issues?  You might look in the overlord log, or in the console, go to the ingestion tab, and try to view the supervisor log from there.

I don't see any log from the Console , and we also don't have overlord POD running for druid

Can you please check overlord logs , logs will have details about supervisor.  If overlord is not  running ?  Also you can try connecting to kafka host from druid console and see if that connects.

Screenshot 2023-03-27 at 11.11.25 AM.png

Are you able to connect data from druid console after entering kafka details , as earlier screenshot i shared . Also check master POD and try login in and check overlord logs as well.

With reindexing you can perform schema changes, repartition data, filter out unwanted data, enrich existing data, and so on.

With compaction you can optimize segment size for better query performance.

it means whatever i can do with compaction i can do with re-indexing too.

You could setup auto-compaction which can be very useful despite it offering a subset of re-indexing's capabilities

Yes, We can do, There is an process called Auto compaction, which you need not do anything and druid will take care of compacting the segments for you

"whatever i can do with compaction i can do with re-indexing too." - I believe so; compaction is like one special case of reindexing, which is more general.

Compaction is simpler than re-indexing in two situations, one of which is to schedule it via auto-compaction.
The other is that it also doesn't require you to know or configure the schema  and granularities of the datasource as they are auto-determined

how many tasks are you running? Each task will create one segment for that time chunk

task for thaat table.? I have not manually run any task i just created the table 3 weeks ago and left it alone

ok...how many partitions in kafka and how many concurrent sub tasks in druid?

Also note that kafka task will create 1 segment every hour by default (the task will ingest for 1 hour and shutdown and new task will start)

This is to ensure that if task fails then data is committed and you need to go back only one hour

the kafka queue has 3 partitions but data is only on first partition

you can turn on auto compaction to compact all these into 1 file

okay, the auto compaction settings needs to be done or automatic settings will do that.?

you have to enable auto compaction. The default settings should be good for your sceanrio

but if i want to change the compaction settings i can do that right.? i saw the current compaction setting had offset at 1D but while making this table i had set it to 6M ,in fine tuning

task duration was set to be 6M in tunings

offset is to ensure that the data ingested most recently is not compacted (the real time task will typically hold a write lock on those segments)

change the offset to PT1H and enable compaction with everything else as default

you set task duration to 6 months? This is not a good thing. If the task fails then all the data of the 6 months will have to re-ingested. Real time tasks need to commit periodically.....you can use intermediate handoff period if you don't want the task to cycle

I think I have not understood this task duration setting

my final aim in this table is that i want my druid table to create only one segment every 6 months, because the data is less, it will be around 400mb hardly after 6 months

task duration essentially ensures that at the end of the duration the task stores the kafka offset and shutsdown after committing the data it has ingested to deep storage and another task will start and pick from that offset

so if a task fails before committing to deep storage then we just to start only from the offset that was stored by the previous task

okay, so does this task duration setting affect segment size.?

yes...each task will write one segment so in you case if the data cam in over 16 hours then you will have 16 segments

so set task duration 1 hour and enable compaction with PT1H offset. this will get you what you want

okay, thanks ,but this will create a segment every hour right.?

i want to create a segment every 6 months

no it will compact based on segment granularity and max rows per segment

so one segment for the granularity if the rows don't exceed max rows per segment

I Have set segment granularity as one year because data will be less

I done the settings as you said, thanks, but it is not reducing my already created segments. how to reduce them to one segment.?

You turned on compaction with offset PT1H?  And the data (__time) is over 1H old?  Do you see any compaction task logs in the Services view of the console?  Do they show success?

yes i did that, it is showing : Target: 'Default 5,000,000' under compaction column in my druid data sources

You need to look at the tasks and see if compaction tasks ran and succeeded 

i applied it for 2 tables, it is showing compaction task for only one and that table is also showing 46 segments for only kbs of data

I think you may still be running into a locking situation. If segment granularity of the real-time ingestion is set to 1 year, that means that it will hold a lock on the whole year while the streaming ingestion is running. Compaction will never be able to obtain the lock, because even after 1 hour of duration, as soon as a new event arrives in the next hour, the 1 year lock will be placed again and streaming task always has priority over compaction.
I haven't tried this, but I think you can use mixed segment granularity, streaming with 1 hour duration and segment granularity of 1 hour. say that at the edge you may get events from the previous hour in the current one, so that you'd have up to 2 hourly segments locked at any given time. You would then <https://druid.apache.org/docs/latest/data-management/automatic-compaction.html#enable-automatic-compaction|set compaction >with segment granularity of 1 year and a "skip Offset from latest" of 2 hours to avoid the streaming locks. I think this will achieve what you want, any streaming segments created more than 2 hours ago should accumulate into the 1-year segments. You'll need to use the JSON form of configuring auto-compaction so that you can specify the change in segment granularity in the granularitySpec. Here's an example I'm testing. It isn't streaming, but it is changing from hourly segment granularity into yearly:
```
  "dataSource": "kttm_transformed",
  "skipOffsetFromLatest": "PT2H",
  "tuningConfig": {
    "partitionsSpec": {
      "type": "dynamic",
      "maxRowsPerSegment": 5000000,
      "maxTotalRows": null
    },
    "type": "index_parallel"
  },
  "granularitySpec": {
    "segmentGranularity": "YEAR",
    "queryGranularity": null,
    "rollup": null
  }
}```
I'll come back with the result once the compaction job runs.

turns out that the auto-compaction did not pick up old hourly segments, I think that is because it looks at recent segments (not sure what the limit to recent is). I ran a manual compaction task covering the whole time interval and it worked:
```{
  "type": "compact",
  "dataSource": "kttm_transformed",
  "ioConfig": {
    "type": "compact",
    "inputSpec": {
      "type": "interval",
      "interval": "2000-01-01T00:00:00.000Z/2021-01-01T00:00:00.000Z",
    },
  },
  "segmentGranularity": "YEAR",
  "granularitySpec": {
    "segmentGranularity": "YEAR",
  }
}```

But I still think you can do this with auto-compaction because it will capture recent hourly segments and merge them into the year segment.

If you give it a try, let us know how it goes.

Also, how big is the 1-year segment (if you have one)?  Do you set inputSegmentSizeBytes at all?  If so, it needs to be bigger than the total size of all the segments being compacted.  If you don't, then what Sergio said.  :slightly_smiling_face:

thanks for the help, will try these and let you know

best "native" tool is the druid console.  But, DBeaver apparently <https://github.com/dbeaver/dbeaver/issues/15172|supports Apache Calcite Avatica>, which can connect to druid using JDBC.

yeah...dbeaver works. You just have to choose Apache calcite avatica from the data base list and provide the jdbc url
```jdbc:avatica:remote:url=<http://localhost:8082/druid/v2/sql/avatica/>```
dbeaver will automatically download the driver from avatica and connect

Hi Shraddha, When you mean connection url, I am assuming you're transforming data with Airflow and want to ingest in Druid? You can submit an ingestion spec (specifying input source and input format of choice) to Druid cluster running via <http://localhost:8081/druid/indexer/v1/task> (if `druid.host` config not specified)

Configuration information w.r.t. ingestion spec can be found here <https://druid.apache.org/docs/latest/ingestion/ingestion-spec.html>

Ingestion tutorial can be found here <https://druid.apache.org/docs/latest/tutorials/tutorial-batch.html>

yes. any memory not used by processes running on the node is used for memory mapping,

you are correct in that the base and replica tasks consuming from the same set of partitions will be used to respond to queries. So yes, there is a possibility that one of the two replicas is slightly behind and could therefore return variable results upon repeated queries. If both tasks have access to the same amount of CPU/Memory resources, then this difference should be negligible.

Thx for the answer Sergio. In the case when one task failed and restarted, the consumed kafka offset can be much different. In the this case, the query results can be significant different, right? Is there any mitigation of this inconsistency?

That's a good question. My understanding is that it is not mitigated but I could be wrong. Sounds like an interesting improvement request if that is the case.

This can be an issue for certain usage cases. Think about using Druid SQL to evaluate alerting conditions every N secs. <@U030K4UM3H7>, it would great if we can have some of the Druid developer/architect to confirm this design. If this is indeed the case, we would like to hear any ideas to solve this potential issue.

Hi Kai, what you are saying is technically true, there can be a condition where a task is rudely interrupted and needs to catch up, if a query is routed to that task while it is catching up it will appear to go back in time (vs the other task). In reality it is not really a problem that you need to concern yourself with, let me explain:

the most common reason that a task will be shutdown is for a software upgrade (upgrading to a new Druid version). in that case there are mechanisms to make sure that the new task can restore its state and does not need to catch up (other than a few seconds): <https://druid.apache.org/docs/latest/operations/rolling-updates.html#rolling-restart-restore-based>

The other thing that could happen is you might be swapping machines (let's say going to a different instance type) in that case, before shutting down the machine you can disable the middle manger letting it finish the tasks is it currently running but not take up any new tasks: <https://druid.apache.org/docs/latest/operations/rolling-updates.html#rolling-restart-graceful-termination-based>

by using these measures you effectively mitigate 99% of times when the task is restarted.

Note that in Imply manager (the in-your-cloud manager) and in Imply Polaris (the DB as a service) we use the techniques outlined above to ensure that you have a monotonically increasing `COUNT(*)` out of the box

so this leaves the case of what to do if you restart a task without pre-warning, on fresh hardware or VM (so restoreTasksOnRestart does not help), and your data volume is such that the task can not quickly catch up by reading the stream up to that point... what do you do then? One answer: is do not worry about that case - it will be rare. Another is make the handoff period shorter (so there is less to catch up on).

None of these mitigations will get you to absolute certainty but they can eliminate 99.9% of the times when a user would have otherwise seen data go back in time. In practice (from working with many customers) the measures outlined above are more than good enough

if you want to get to 100% certainty then you have to do it in the client. In you alert code, query for COUNT(*) as well as for the rest of the conditions and have your client side code ignore the results if COUNT(*) ever goes down for a given hour.

<@U030W50CNAC>, thx for the detailed explanation. Let me make sure I understand the  restored-based config in detail

Here, we deploy the druid cluster on the k8. The nodes of k8 be be patched one by one. From the Druid point of view, the MM and all the Kafak ingestion peons can be killed async (just like ctrl-c to kill a program in the terminal). Here, if we turned on restore-based config. When the MM pod restarted on another k8 node, it can retrieve the state information from the database (say Postgres or MySql depending on our config). And the MM would restart the peons from where the kafka offset they ingested at the time they got killed. The question is that how the existing ingested data which can be huge, say 200M depending on the config can be retrieve? They were on the old node which were gone. Wondering if my understanding of this restored-based config is correct here or maybe not right?

The reason I bring this up is that indeed this k8 node patch case is something we met in production. The graceful-termination-based approach would help much in this async mode. (Other pre-planned rolling upgrade case seems to be better candidate for the the graceful-termination-based approach.) Would appreciate your feedback here.

when you restart you k8 nodes do the new nodes have access to the disk of the old node?

Actually, I am not totally sure. We use EKS (network volumes for segment caching). This one can be re-attached with the same pod for sure. Let us say if we use EKS for temporary files for ingestion task, would that help?

If your MMs are a stateful set then this will be true. 

Right, for statefulset, the MM has identity and  EKS would reattach to the same pod. The question is that would the restarted peon pick up from where it left before? Or will the peon just get restarted to ingest from the beginning?

If you re-attach the same local storage and have `druid.indexer.task.restoreTasksOnRestart=true` set then it will pick up where it left before as I understand it

But in general, saying the MM uses instance storage, there is not such guarantee. Then the peon would restart from the beginning, right? So the conclusion is that it would be better to use EKS in this case?

I should note that I am not an expert on the topic of this thread, I am just letting you know how I understand it works. I wanted to link the docs above

I mostly interact with Imply managed Druid clusters and all of :point_up: is just set up behind the scenes so I never think about it too deeply

I actually did some research and expanded my understanding of things that I previously took for granted as part of answering your question.

Understood, sometime developers may not think too much into operations. Just to confirm the intention of the the two features (or how do we reason about the two approaches) restored-based and graceful-shutdown-based. The restore-based is designed for the async MM/peon failure while the graceful-shutdown is designed for graceful shutdown cases. In the restore-based config, it would be beneficial if the data state-info (partially generated segments) to be available to the restarted peons.

With all the above, there is still a chance of inconsistent query result in the real time query. One way I can think of is that we can add the kafka offset in the query context as a hint. That way, the lagging peon may serve the query only once its ingestion is passing the kafka offset. The query should also have an option to return the kafka offset in the query result.

Sounds interesting, but kafka offsets are per partition, and each ingestion task handles a subset of partitions, so this might work if we add all the partition offsets in the context and then have each task recognize its subset of partitions.

Right. In my previous post, I abstract kafka offset as one value. In fact, it is a set of value corresponding to all the partitions for specific peon, from implementation point of viewl

Sounds like an interesting proposal for a change.

I added an issue for improvement based on this discussion. <https://github.com/apache/druid/issues/13994>. Maybe I will also submit an PR next if this issue turns out big enough for our usage.

I do not understand the mechanism you are proposing who will set the hint?

Also are you saying that you feel like all the mitigation measures outlined above are not enough for your use case?

&gt;  I do not understand the mechanism you are proposing who will set the hint?
The idea is simple (though may not be the most elegant). When a query is served by a peon, the kafka offsets of the peon is returned. Next time the query is issued again, the previous offsets is also used as the input for the peons. If the serving peon see an offset later then the current offset it is consuming, it should wait (or fail the query) till it catch up.

This is the same idea of guaranteeing exactly-once ingestion from Kafka topics. To achieve exactly once, Druid associate the kafka offset with the segment in the meta store database. (Actually, I did not dig the code to confirm this. But this is just a common technic almost all the distributed system use with Kafka).

&gt; Also are you saying that you feel like all the mitigation measures outlined above are not enough for your use case?
Right, for some use cases, like alerting, if you can go back in time even once in a while, you own a lot of explaining to the user.

&gt;  you should add it to the GH issue
I am kind of new to the Druid open source community. Can you illustrate what does "GH" mean here? I just created an issue as <https://github.com/apache/druid/issues/13994>

you gave a more detailed explanation to me, and it made it more clear what you are proposing so I though you should comment or edit the issue you created to make it more clear

Just updated the description a little bit. Basically the same as what I illustrate <https://apachedruidworkspace.slack.com/archives/C0303FDCZEZ/p1680039318801959?thread_ts=1679951040.402899&amp;cid=C0303FDCZEZ|here>. Hope it makes it clearer. Feel free to add your comments in the issue too.

I don't know Airflow, but it looks like it, based on get_uri in their <https://airflow.apache.org/docs/apache-airflow-providers-apache-druid/stable/_api/airflow/providers/apache/druid/hooks/druid/index.html#module-airflow.providers.apache.druid.hooks.druid|doc>.

are you trying to query druid or kick off an ingestion task. The api end point will be different based on what you are trying to do

I wish I knew but I'm not really familiar with airflow

As others have mentioned there's a couple of possible endpoints you might be interested in

If it wants the Broker that'd be `http://&lt;hostname&gt;:8082/druid/v2/` (or maybe you just provide host and port and Airflow fills in the rest…?)

If it wants Overlord or Indexing Service that'd be `http://&lt;hostname&gt;:8090/druid/indexer/v1/` (or, again, maybe you just provide host and port?)

Hope this helps!

<@U03VCP3KG01> I am planning to put the together something with airflow and druid - blog plus samples. Should have it in a few days

Thank you everyone. The information that I got is, it is needed for ingestion. Based on that what should be the uri?

i would try `http://&lt;hostname&gt;:8090/druid/indexer/v1/` first or perhaps `http://&lt;hostname&gt;:8090/` (if airflow automatically adds the `/druid/indexer/v1/` part)

sure, thanks, let me share it with the team.

In one of the meet ups last week there was a question on ACID. I pulled together something to indicate a way of thinking about it in Druid <http://meditationsonbigdata.blog/2023/03/27/acid-and-apache-druid/|http://meditationsonbigdata.blog/2023/03/27/acid-and-apache-druid/>