This message was deleted.

We use auto compaction

Auto compaction uses lower priority locks such that your ingestion will overrule the lock and auto-compaction job will be canceled if they collide. There is no auto-scheduling of SQL-based ingestion in Druid at the moment, but cron-based should be fine. I'm curious, how often are you running those batches? Average data size per batch?

We have segment granularity of 1 hour, intermediate handoff set as 15 mins. Idea is that we run SQL ingestion every 1hour to replace 1 hour old data issue with this approach is potentially SQL ingestion might run on the same time segments compaction job trying to compact. From the doc

REPLACE statements acquire an exclusive write lock to the target time range of the target datasource. No other ingestion or compaction operations may proceed for that time range while the task is running.

would that mean compaction waits while SQL ingestion is running on a segment?. As the time range should align with the PARTITIONED BY clause, It means that we can only REPLACE one hour time range. Our one hour data would be 15mil rows and each row would be 100bytes, that is ~1.5 to 2 GB data per batch.

The compaction job that collides with the REPLACE will be canceled, at the next round of compaction it will try again. 2GB per batch sounds not be a problem with appropriate resources. Just to understand your approach. You are ingesting from a stream and then following it up with batches? why are the batches needed?

We get the partial records for a dimension "tags" which we need it to be aggregated into one value. My understanding is that compaction wouldn't rollup dimensions, Is that correct. Is there any other approach to aggregate dimensions after ingesting

dimenstions(_time, id, tags) metrics( duration)

{"timestamp": "2011-01-12T00:00:00.000Z", "abc", ["t1","t2","t3"],  1}  #row1
{"timestamp": "2011-01-12T00:00:00.000Z", "abc", ["t3","t4","t5"], 2}  #row2
{"timestamp": "2011-01-12T00:00:00.000Z", "abc", ["t5","t6","t7"], 3}  #row3
{"timestamp": "2011-01-12T00:00:00.000Z", "abc", [], 4}                #row4

we're expecting 
_time, id, tags (union), duration(max)
{"timestamp": "2011-01-12T00:00:00.000Z",  "abc",  ["t1","t2","t3", "t4","t5","t6","t7"], 4}

Yes, you can aggregate the tags at ingestion, here's an example using SQL based ingestion with some inline data similar to yours. The data:

{"timestamp": "2011-01-12T00:00:00.000Z", "trip_types": ["walk","drive","fly"], "duration":6}
{"timestamp": "2011-01-12T00:00:00.000Z", "trip_types": ["walk","bus","walk"], "duration":53}
{"timestamp": "2011-01-14T00:00:00.000Z", "trip_types": ["drive","fly","drive"], "duration":60}
{"timestamp": "2011-01-14T00:00:00.000Z", "trip_types": ["jump"], "duration":0}

The SQL based ingestion query, you'll need to set "Enable Group By multi-value unnesting" to True in the query context:

REPLACE INTO "inline_data" OVERWRITE ALL
WITH "ext" AS (SELECT *
FROM TABLE(
  EXTERN(
    '{"type":"inline","data":"{\"timestamp\": \"2011-01-12T00:00:00.000Z\", \"trip_types\": [\"walk\",\"drive\",\"fly\"], \"duration\":6} \n{\"timestamp\": \"2011-01-12T00:00:00.000Z\", \"trip_types\": [\"walk\",\"bus\",\"walk\"], \"duration\":53} \n{\"timestamp\": \"2011-01-14T00:00:00.000Z\", \"trip_types\": [\"drive\",\"fly\",\"drive\"], \"duration\":60}  \n{\"timestamp\": \"2011-01-14T00:00:00.000Z\", \"trip_types\": [\"jump\"], \"duration\":0}    "}',
    '{"type":"json"}',
    '[{"name":"timestamp","type":"string"},{"name":"trip_types","type":"string"},{"name":"duration","type":"long"}]'
  )
))
SELECT
  TIME_PARSE("timestamp") AS "__time",
  ARRAY_CONCAT_AGG( MV_TO_ARRAY("trip_types")) AS "trip_types",
  SUM("duration") "duration"
FROM "ext"
GROUP BY 1
PARTITIONED BY DAY

The table ends up with:

__time                   trip_types                                 duration 
2011-01-12T00:00:00.000Z ["walk","drive","fly","walk","bus","walk"] 59
2011-01-14T00:00:00.000Z ["drive","fly","drive","jump"]             60

We ingest real time from Kafka. Is there a way to ingest realtime using SQL based ingestion?

No, not yet, I'm hoping that gets added. I think the equivalent operation might exist in real-time ingestion using rollup. I'll take a look.

Thanks that would solve my use case

@tilak chowdary Hey, I have not forgotten about this. I tried to get this to work with real-time ingestion, using an expression in the metrics spec to do the aggregation of the arrays at rollup. It did not work, but this sounds like an interesting use case for streaming. It reported an error once the ingestion task received any message that it does not support ARRAY<STRING> datatype for this operation. Do you mind creating an issue with array aggregation use case for streaming ingestion and I'll add comments on my tests?

Thanks, Sure I will and update after the bug is created