Apache Druid

The document of ARRAY_AGG () actually says
```Use of ORDER BY within the ARRAY_AGG expression is not currently supported, and the ordering of results within the output array may vary depending on processing order.```
So I guess the answer is no way?

yeah, the order in which results are merged can vary depending on the placement of segments among data servers

Hi <@U030C8H59T8> , thank you for getting back to me. Is it possible to maintain the order? Like using cluster by or something? Thanks!

What are you trying to do? Something like:
```select category, count(*), anyValue(jsonColumn) as "randomValue"
from table
group by 1 ```
So any full json would be ok?

I could not find a 'stringAny' function, but I did find a 'stringEvery' function, which checks to see if every element of the set satisfies a given condition.  So long those lines I am guessing stringAny would indicate if one or more elements of the set satisfied the given condition?

If that is the logic of this function, then in SQL we would generally implement that logic using a filtered aggregate, e.g.:

• select max(case when strField like 'x%' then 1 else 0 end)
The above aggregate expression would return either 1 or 0.

Nested JSON columns can then then accessed in place of the field reference, e.g.:

• select max(case when JSON_VALUE(jsonCol, '$.strField') like 'x%' then 1 else 0 end)
and of course you can use whatever conditional expression you want in a CASE statement.

Let us know if this is not what you were looking for either.

Thanks.  John

I believe `stringAny` how you put in ingestion spec a rollup equivalent for `ANY_VALUE(expr, maxBytesPerString)` in SQL

Looking for aggregation function in group by queries. especially select any value in the group. ANY_VALUE(expr, maxBytesPerString) throws exception that complex&lt;json&gt; not supported.

maybe `ANY_VALUE(TO_JSON_STRING(expr), maxBytesPerString)`

Screen Shot 2023-03-16 at 6.39.55 AM.png

Still not sure what you are looking to do.  If you can post your SQL statement here that should hopefully clarify.

Fyi this statement against the KTTM-Nested demo dataset works for me:

``` select adblock_list GroupName, 
        count(*) NumEntries,
        max(case when json_value(event, '$.type') like 'Save%' then 1 else 0 end) HasSave
   from "kttm-nested" 
  group by 1```
Here is a screenshot showing the statement and output:

Sorry, I didn't realize stringAny is a Druid Native SQL aggregate function ... okay then my suggestion for a SQL version changes to use the Earliest() function:

``` select adblock_list GroupName, 
        count(*) NumEntries,
        earliest(json_value(event, '$.type'), 1000) FirstType
   from "kttm-nested" 
  group by 1```

Thanks John for looking into this. I wanted to do this  from this example data <https://druid.apache.org/docs/latest/querying/nested-columns.html#native-batch-ingestion>
``` SELECT
  department,
  ANY_VALUE(product, 100),
  ANY_VALUE(details, 100)
FROM "nested_data_example"
GROUP BY department```

details column output is a string with StructuredData(---). I am trying to get as a json

Screen Shot 2023-03-16 at 9.51.14 PM.png

In general exporting the data from postgress into something like S3 and then loading it with Druid using batch ingestion would be more robust then trying to move data between the DBs directly

I was able to do that by adding postgresql-jar file to `lib` folder

I believe if you add `postgresql-metadata-storage` extension to your load list it’ll be enough

```druid.extensions.loadList=["postgresql-metadata-storage", "druid-s3-extensions", "druid-stats", "druid-histogram", "druid-datasketches", "druid-kafka-indexing-service"]```


I did that but it explicitly looks for postgresql-jar under lib which won't be added based on above

In kubernetes, the jar will be in /opt/druid/extensions/postgresql-metadata-storage/postgresql-xx.x.x.jar

And it is just dynamically loaded at startup.

I tried entering shell script of specific pod but it doesn't work like that. Any idea on how to approach this issue ?

do you know where exactly the problem is occurring, or do you have more details on the error?

"Deadline exceeded" sounds like some kind of connection problem

Hey <@U0344FW86DD>

The error message I was facing with JDBC connection is below :

An error occurred while communicating with Other Databases (JDBC)
Bad Connection: Tableau could not connect to the data source.
Error Code: FAB9A2C5
Deadline Exceeded
Generic JDBC connection error
Deadline Exceeded

As suggested by <@U04CGQ1P68Z> we tried using Druid Imply connector and are facing below error and we have a raised a ticket <https://imply.zendesk.com/agent/tickets/12411> with Imply support.

An error occurred while communicating with Druid Connector by Imply
Bad Connection: Tableau could not connect to the data source.
Error Code: FAB9A2C5
Deadline Exceeded
Unable to connect to the Druid Connector by Imply server. Check that the server is running and that you have access privileges to the requested database.

But we think the main cause of error is missing inbound rule for the external load-balancer for port or some network restriction with Druid cluster which we need to bypass with proxy. We are able to pass and access Druid data source with HTTP request using proxy only. We are not able to access Druid cluster via HTTP request without proxy. But we are not able to figure out where can we pass proxy settings (proxy_host, proxy_port) with JDBC connection/request from Tableau to Druid. We tried passing proxy settings in the properties file while connecting Tableau to Druid using JDBC connection, but it is not working as well and we are getting the same above error. cc <@U04T8067UUV> <@U031K06AYQ0> <@U043NEYH3NG>

hmm, I'm not sure what's going on here just from looking at the error— if you're in contact with Imply support they should be able to help you best!

Hey <@U040C9JCW8G> As discussed, I used this document as reference - <https://kb.tableau.com/articles/issue/unable-to-connect-to-athena-via-proxy-server> for providing proxy settings while connecting Tableau to Druid Cluster. cc <@U04T8067UUV>

Hey <@U040C9JCW8G> <@U0344FW86DD> Thanks everyone! we were able to connect Tableau to Druid after our data platform team whitelisted out VPN IPs to connect to druid.

<@U04UJ838K1N> Glad to here that you are connected Tableau to Druid.

at present you have to define the metadata on the extern. If you have your data on the filesystem you can use the console wizard to sample the data and write out the metadata for you

For the regular ingestion even if we don’t set `DimensionsSpec` it identifies dynamically from the file. Why is that not true for SQL based ingestion ?

It's because the planning for the SQL statement needs to know about the columns available in the input files

It's mentioned in the known issues for `EXTERN` here: <https://druid.apache.org/docs/latest/multi-stage-query/known-issues.html#extern>

We will be addressing this in a future release by making schemaless + SQL = :heart:, so stay tuned!!

Please check out this thread: <https://apachedruidworkspace.slack.com/archives/C0309C9L90D/p1675782972380179>

There is a `sql` input source in native batch ingestion, which supports ingestion from the `postgres` database. You can use that to ingest the data from the existing postgres database to druid. More info can be found here: <https://druid.apache.org/docs/latest/ingestion/native-batch-input-sources.html#sql-input-source>.

I don’t think there is an explicit maximum. But too many column merges at once during the segment building phase could get you in trouble with memory. You can throttle that step with maxColumnsToMerge parameter.

I would say _generally_ tens of columns is 100% totally fine, hundreds is potentially fine depending on how big they are, and thousands of columns is pushing it

If you find you need a lot of columns because you have a lot of complexity in your input data, you might prefer a design involving nested columns instead: <https://druid.apache.org/docs/latest/querying/nested-columns.html>

The Druid - Apache Dolphin meetup has started. Join here <https://imply-io.zoom.us/j/96326362484?pwd=QnRmaHZKWFM1SlZrZDhJMC9HWlpaQT09>

The recommended way to do it is by dumping your data with mysqldump and then loading it via batch ingestion in the console

os start by running something like `mysqldump -u [username] -p --tab=[output_directory] --fields-terminated-by='\t' --fields-enclosed-by='' --fields-escaped-by='\\' --lines-terminated-by='\n' [database_name] [table_name]`

<@U030W50CNAC> Thanks for your reply, I want to execute this process on automation through python scripts and dataflow is like MySQL to hive load into Druid .If you have any others options using python only let me know.

do you want `LATEST(expr, maxBytesPerString)` (the string version)? see <https://druid.apache.org/docs/latest/querying/sql-aggregations.html>

export (1).png

image.png

well you could store score events in Druid that change peoples scores and then aggregate across these to build the leader board

hi
the problem is just not (sorting)a leader board.
we should process new event consume it for past events match to rank 1 if remained more amount match it to second rank if not match add it to somewhere is best rank for it...:slightly_smiling_face:

these systems are <http://en.wikipedia.org/wiki/Online_analytical_processing|OLAP> but i think  we want OLTP

<https://redis.com/solutions/use-cases/leaderboards/>

<https://redis.io/docs/data-types/sorted-sets/>

you should consider milionse of events are comming per second and we want change state of each row/user ...they are related together ......

suppose the ram down what happen for stream

alive with million of events in the same leaderboard? that's unusual, and will require unusual software. I would recommend to enter in contact with redis enterprise
bc even clustering will not help you if you really want just one big board with a very very long tail

if you have a lot of small leaderboards, then you can use redis with replica for failvover, and kafka for replaying when things goes bad and you need to reprocess until the last savepoint

redis is nice for persisting it even we can replicate but it has not processing part ....
we should compute new board also when new orders comming always......it is a stream problem but also at same time a board

Sounds like a moving average, but instead a moving leader board. The problem is if you are always adding and always removing you have to always be recalculating. If you know the duration of time between something entering and leaving the board, you could have one real time stream and another delayed stream and the real time stream adds and the delayed stream removes. You are probably going to have to develop something bespoke.

i am thinking more to know whether it is streamble or not ..

I'm not sure I'm following, but if it's a moving avg problem, then it's possible to use kafka-stream to emit the avg when it changes, and capture the change to update the ranking somewhere else

kafka-stream internally uses a RocksDB to store the stateful info required by the windowing and other processing, and it's somewhat crash-resilient

I would start with ksql, instead of building direly over kafka-stream

Kafk-stream can not access to past events ....we discussed before kafka stream partioning data in network ...just spark can be model it in memory

the window is bigger than the kafka retention? because i'm pretty sure that you can read past data if it's stored in the topic

with this ksql code it writes a java program using kafka-stream to do the heavy part
it will keep emitting for each key at regular interval if there's changes
then you can use it to save into another topic to update the ranking in redis or other tool

ksql is sort of a spark but when the source and destination is a kafka

ksql is nothing just is sql lang of kafka

So I guess it would be that, but instead of avg it would be sum for this use case.

you write the instructions in sql, it compile into a program that can run in cluster mode across multiple workers, etc

That's really interesting. I will have to investigate using this in the future.

moving avg:rolling_on_the_floor_laughing:

It doesn't look like it is documented to me.

I'm not sure about this particular case, although, usually if something like this is undocumented it's because the API isn't stable enough yet. So it might be that. I think <@U030C8H59T8> wrote it so he might have more info

yeah, I think it has stabilized now probably, didn't document at first for that reason

Thanks very much, really appreciate the replies!

<https://github.com/apache/druid/blob/master/processing/src/main/java/org/apache/druid/query/aggregation/ExpressionLambdaAggregatorFactory.java#L103|https://github.com/apache/druid/blob/master/processing/src/main/java/org/apache/druid/query/aggregation/ExpressionLambdaAggregatorFactory.java#L103> for the current set of properties it accepts

if Kafka is in the middle, could you use Druid's builtin Kafka integration to read from there?

sorry. I mean ingest data directly from MQTT server (without using Kafka)

Any reasons why you can’t use Kafka ; Druid has native connection to Kafka

Hi. there is no reason why I cant use Kafka. Its just that my manager want me to explore the option (ingesting directly from MQTT server into Druid)

I suggest that you use Kafka since it’s native to Druid

1) yes, Druid works with Graviton; see here for example <https://aws.amazon.com/solutions/case-studies/zomato-case-study/>

2) for Apache Druid, security fixes are either applied to the latest major version as a patch release, or to the next major version, if the Druid PMC determines that the severity of the issue does not necessitate a patch release. Major versions are generally released once every 3 months

3) performance would be similar, as most of the delta between Apache and Imply versions of Druid are in areas other than perf-related

BTW, note that from a branding PoV there is no such thing as "Imply Druid" :slightly_smiling_face: — it is more accurate to say that Imply includes a custom distribution of Apache Druid

Perfect, thanks a lot <@U0344FW86DD>. Thanks for correcting too. Hope all query types like groupby, top N, aggregate etc works the same way as well.

I should add that the area the Imply really helps in is 2.

And 3 is better with Imply not because there is an out of the box difference but the experience and proprietary monitoring tools provided let you optimize  perf / TCO better.

There is a <#C030CMF6B70|> channel here, and also a google dev mailing list.

For the dev mailing list, see <https://druid.apache.org/community/#:~:text=Development%20mailing%20list%3A%20dev%40druid,us%20on%20Twitter%20at%20%40druidio>

Thank you <@U0411DE2SS0> so either raise it on dev or email it

<@U0411DE2SS0> is it ok to delete this message from general??

thank you very much :slightly_smiling_face:

oops, I deleted the wrong message, where I said "of course" , lol

The only thing I know to do currently is to group by all of the columns in your select list.  Once I do that it let's me sort on a non-__time field.

Technically this may be a similar task for the engine that a sort would entail anyway ...

Screenshot 2023-03-20 at 2.43.24 PM.png

If you want to order by on any column, then you can use the MSQ engine to run the query, that doesn’t have the limitation on ordering by __time only.

is this question meant for the Apache Spark channel?

Yes that is correct.  If you want to be absolutely sure, then create a small test datasource and try issuing the kill command against it to confirm that the used segments are left alone.  (I just did this on the wikipedia demo datasource to confirm it myself).

storage is too cheap, i didn't notice until yesterday

I think you meant to ask this in a Spark chat, this is for Druid, we don't have `Function2` :slightly_smiling_face:

ofcourse but maybe somebody works spark also:slightly_smiling_face: ....also there is no spark chat

Back with another edition of APAC Druid Drop-in. We will discuss ACID with Druid, sizing a Druid cluster and a pipeline with Flink, Kafka and Druid (why do it and how to do it).<https://www.meetup.com/bangalore-druid-meetup-group/events/291483018/?utm_medium=referral&amp;utm_campaign=share-btn_savedevents_share_modal&amp;utm_source=link|https://www.meetup.com/bangalore-druid-meetup-group/events/291483018/?utm_medium=ref[…]tm_campaign=share-btn_savedevents_share_modal&amp;utm_source=link>

Hi!
A few questions to help debug this further
• Is this occurring for a specific query only, (since you mentioned it runs in a second otherwise) or for all queries including simple ones which shouldn't timeout ever?
• Is the url provided correctly pointing to druid? (For example is there something in logs to say that the query has hit the API on the broker and started successfully) Is there any exception or error in the logs that is causing it to the query to fail and timeout?
The console also internally calls the same API, so this shouldn't happen.

1. Not to specific query -&gt; example am using 20 sql queries means in random fresh some of the query response and some are time-out.  Even in alternate response same query time-out which responds some time.

2. Yes it's pointing to Druid.  Reason if always time-out / denied then this must be problem.  But it works sometime and timeout sometime.

where the SQL query request logs will happen ?

Yes, there should be some logs in the broker

if the api is posting multiple queries then this may be an issue with not having enough resources for concurrent queries.

How many queries per second are you posting?

In the console you are likely running one query at a time

In a page 1 do 10 queries with 10 api call separetly

but, not in single API call - multiple queries

here 1 point I want to mention, actually for 1 of the data source to kafka data stream going on with high volume (about 20 Lakhs) in parallel. Do this processor usage / memory usage... lead to delay the response for other datasource SQL queries ? Just asking. ?

yes. Depends on how many nodes you have and how many threads.

given a historical server, only one segment is processed at a time

if there's multiple queries in the queue and just one server, if all queries have the same priority (default), the queries will have random execution times, because it's not guarantee that the historical server will read all the segments for the first (incoming) query, it will try to balance the load between all the active queries, that can lead to very poor performance if you have a lot of queries and just one processing thread, for exmple

anothing thing, how are you encoding/escaping the query in the console versus in the application?

if you are filtering by ID, try to make sure that it's using the index by using CAST(filtered_column AS varchar) = ?

that can drastically improve response time, if you ingested as string but are querying as int, this will lead to a lot of unnecessary conversions and not using the right indexes

&gt;  hopefully I think that in 26.0.0 (next version) with data catalog feat, Calcicte planning engine will get a refactor and may become smarter

One thing to note here - the Druid console uses limits and approximation by default, while an API call might not - do you get the same performance when you turn off the approximation and limit?

Thank You Renato, will check on your point.

Thank you Kyle. Will try it.

Renato, am currently using 25.0.0 version.

for the performance, it's fast as it's just a local lookup, but it can hurt if you are just filtering a result of this lookup, because it would mean that all rows would have to be matched and tested and probability discarded

You can control if the historical node becomes available for querying only after the lookup's is already in sync, but there's no automatic way to observe this, you can use the coordinator endpoint to get an overview of the cluster, but so far, there's no way to tell druid to not use a node if a datasource is not fully loaded yet

```druid.lookup.enableLookupSyncOnStartup=true
        druid.lookup.snapshotWorkingDir=/var/druid/lookups```