I need help in Kafka streaming with avro schem

Question

< Neha Pawar> I need help in Kafka streaming with avro schema Here is my avro schema Main object is Metric It contains one MetricSource nested object I could stream Metric asvc fields like product productversion metricPath I dont know how to map MetricSource I would like to map as follows MetricSource time MetricSource metric MetricSource metricValue MetricSource category MetricSource subCategory Metric product Metric productVersion Metric metricpath ```Metric avsc =========== namespace com blah name Metric type record fields name product type string null name productVersion type string null name MetricSource type com blah MetricSource null name metricPath type type array items string null MetricSource avsc =========== namespace com blah name MetricSource type record fields name metric type string null name metricValue type string null name time type long name timeOffset type double name category order ignore type null string default null name subCategory order ignore type null string default null ```

Neha Pawar · Answer

Use `transformFunction jsonPathString MetricSource $ time lt default value gt `

Neha Pawar · Answer

This is an example to extract time

Neha Pawar · Answer

Machhindra · Answer

Do I need to extract at query time I was hoping to map the fields at the ingestion time

RK · Answer

< Machhindra> you need to do this while ingestion in your table config file json

RK · Answer

Like this

Machhindra · Answer

< RK> thanks for suggestion I added the transformFunction to table schema as follows It worked ``` name metric dataType STRING transformFunction jsonPathString MetricSource $ metricClass null ```

Neha Pawar · Answer

Put this in table confg not schema

Machhindra · Answer

okay It is strange that it worked even I put it in table schema I will move to table config

Neha Pawar · Answer

like this <https docs pinot apache org developers advanced ingestion level transformations column transformation>

Neha Pawar · Answer

yes it used to be in schema We have deprecated it So better to move to table

Machhindra · Answer

Ah

Machhindra · Answer

< Neha Pawar> I managed to get data from our kafka topic to Pinot So far good progress Here is how my table look like as of today ```metric metricValue entityMap metricTime ====================================================================================== CPUTIME 2210 Plan myplan Pkg mypkg Server myDb2 1622076960000 WAITTIME 12 Plan myplan Pkg mypkg Server myDb2 1622076960000 MQBPUSE 50 MqBPool 2 MqQmgr MQA3 1622076960000 MQBPUSE 110 MqBPool 3 MqQmgr MQA3 1622076960000 WLMPI 0 5 ZosWmSysSum jobs 1622076960000 MQQFULL 10 MqQueue myqueue MqQmgr CSQ4 1622076960000 CPUTIME 2220 Plan myplan Pkg mypkg Server myDb2 1622083650000 WAITTIME 12 Plan myplan Pkg mypkg Server myDb2 1622083650000 MQBPUSE 50 MqBPool 2 MqQmgr MQA3 1622083650000 MQBPUSE 110 MqBPool 3 MqQmgr MQA3 1622083650000 WLMPI 0 5 ZosWmSysSum jobs 1622083650000 MQQFULL 10 MqQueue myqueue MqQmgr CSQ4 1622083650000 ``` This is operational metric data Just like Prometheus 300K Minute There are different metrics being collected at regular intervals The metrics are measured for given entity in the entityMap e g CPUTIME metric is collected for a particular plan package and Db2 server in the first row Here are some of the queries I wish to fire on this table ``` Get me timeseries for the given metric plan package and server select from metricTable where metric= CPUTIME and Plan= myplan and Pkg= mypkg and Server= myDb2 and metrictime lt 1622076960000 Get me two timeseries for the given two metric plan package and server select from metricTable where metric IN CPUTIME WAITTIME and Plan= myplan and Pkg= mypkg and Server= myDb2 and metricTime lt 1622076960000 Get me all timeseries for the given plan irrespective of package or server select from metricTable where metric= CPUTIME and Plan= myplan and metricTime lt 1622076960000 Get me top 10 highest CPUTIME package for a given Server Get me average CPUTIME for packages for a given Server with time window of 10 minutes Get me list of Plans for a given Server Get total CPUTIME used for a given Server``` How should I flatten the entityMap column so that I can satisfy my queries Not sure if multi valued column would be helpful

RK · Answer

< Machhindra> while selecting try giving column name using json format column name

Neha Pawar · Answer

< Kishore G> does this look like a good fit for using json index

Kishore G · Answer

yes this is perfect case for json index

Neha Pawar · Answer

< Machhindra> use

Machhindra · Answer

Thanks Neha I reading through the documentation Once you apply JSON index I will have to use queries like ```SELECT FROM metricTable WHERE metric= CPUTIME and JSON MATCH entityMap $ Plan = myplan ``` Instead of ```select from metricTable where metric= CPUTIME and Plan= myplan ```

Neha Pawar · Answer

correct Tagging the author < Jackie> for more help if you need

Machhindra · Answer

I was hoping to run queries as select from metricTable where metric= CPUTIME and Plan= myplan

Machhindra · Answer

this is more intuitive and work seamlessly with Superset

Machhindra · Answer

I am trying to learn how can I use query SELECT FROM metricTable WHERE metric= CPUTIME and JSON MATCH entityMap $ Plan = myplan from superset

Jackie · Answer

We are working on an effort to add the `JSON` data type and simplify the query from `SELECT FROM metricTable WHERE metric= CPUTIME and JSON MATCH entityMap $ Plan = myplan ` to `SELECT FROM metricTable WHERE metric= CPUTIME and entityMap Plan= myplan `

Jackie · Answer

Currently it is not supported yet

Jackie · Answer

< Xiang Fu> Can we do the `JSON MATCH` within the superset

Xiang Fu · Answer

I think you can

Xiang Fu · Answer

just check the generated query from superset

Machhindra · Answer

< Neha Pawar> < Xiang Fu> < Jackie> < RK> Added JSON index as suggested ``` jsonIndexColumns entityMap autoGeneratedInvertedIndex false createInvertedIndexDuringSegmentGeneration false sortedColumn metric ``` Here is how my table looks ```select from metric v6 where entityMap = null limit 10 category entityMap groupId metric Type74Subtype1 Volser HFSMND 1aeadda1 86e1 4ee4 aa4e d067a381db0a DCTAVG VOLSER Type74Subtype1 Volser NMD008 632afe50 a42c 4c15 b0cd 9c6a4e0e08b1 DRTAVG VOLSER Type74Subtype1 DeviceNo 0x6E06 eadee40b 3c02 49f8 a295 0b059606a3f5 DQTAVG DEVNUM Db2Subsys Db2Bufp BP42 Db2Ssid DH3G f2a9ccc7 6aa6 4713 8382 f7b96f62c124 IDBBREAD``` Notice that the entityMap JSON object differs for each row Following query fails ```SELECT FROM metric v6 WHERE JSON MATCH entityMap $ Volser = HFSMND errorCode 200 message QueryExecutionError njava lang RuntimeException Caught exception while running CombinePlanNode n tat org apache pinot core plan CombinePlanNode run CombinePlanNode java 157 n tat org apache pinot core plan InstanceResponsePlanNode run InstanceResponsePlanNode java 33 n tat org apache pinot core plan GlobalPlanImplV0 execute GlobalPlanImplV0 java 45 n tat org apache pinot core query executor ServerQueryExecutorV1Impl processQuery ServerQueryExecutorV1Impl java 294 n tat org apache pinot core query executor ServerQueryExecutorV1Impl processQuery ServerQueryExecutorV1Impl java 215 n tat org apache pinot core query executor QueryExecutor processQuery QueryExecutor java 60 n tat org apache pinot core query scheduler QueryScheduler processQueryAndSerialize QueryScheduler java 157 n tat org apache pinot core query scheduler QueryScheduler lambda$createQueryFutureTask$0 QueryScheduler java 141 n tat java util concurrent FutureTask run FutureTask java 266 n tat java util concurrent Executors$RunnableAdapter call Executors java 511 n tat shaded com google common util concurrent TrustedListenableFutureTask$TrustedFutureInterruptibleTask runInterruptibly TrustedListenableFutureTask java 111 n tat shaded com google common util concurrent InterruptibleTask run InterruptibleTask java 58 n tat shaded com google common util concurrent TrustedListenableFutureTask run TrustedListenableFutureTask java 75 n tat java util concurrent ThreadPoolExecutor runWorker ThreadPoolExecutor java 1149 ``` After facing these challenges I decided to create new REALTIME table with the indexes on from the very beginning The number of rows are 2 8M vs 93M on the previous table Now I dont receive the ERRORS but no result is shown either

Machhindra · Answer

For the community Following query ``` SELECT FROM metric v7 WHERE JSON MATCH entityMap Db2Bufp= BP20 ``` works now Team Pinot recently changed the syntax of the JSON MATCH which is not included in `0 7 1`

Machhindra · Answer

It looks like I hit another bump I cant fire simple query in console today This may be due to sortedcolumn or json index ```select from metric v7 limit 10 errorCode 200 message QueryExecutionError njava lang IndexOutOfBoundsException n tat java nio Buffer checkIndex Buffer java 551 n tat java nio DirectByteBuffer getInt DirectByteBuffer java 684 n tat org apache pinot core segment memory PinotByteBuffer getInt PinotByteBuffer java 144 n tat org apache pinot core io util FixedByteValueReaderWriter getInt FixedByteValueReaderWriter java 35 n tat org apache pinot core segment index readers sorted SortedIndexReaderImpl getDictId SortedIndexReaderImpl java 83 n tat org apache pinot core segment index readers sorted SortedIndexReaderImpl readDictIds SortedIndexReaderImpl java 108 n tat org apache pinot core segment index readers sorted SortedIndexReaderImpl readDictIds SortedIndexReaderImpl java 33 n tat org apache pinot core common DataFetcher$ColumnValueReader readStringValues DataFetcher java 439 n tat org apache pinot core common DataFetcher fetchStringValues DataFetcher java 146 n tat org apache pinot core common DataBlockCache getStringValuesForSVColumn DataBlockCache java 194 n tat org apache pinot core operator docvalsets ProjectionBlockValSet getStringValuesSV ProjectionBlockValSet java 94 n tat org apache pinot core common RowBasedBlockValueFetcher createFetcher RowBasedBlockValueFetcher java 64 n tat org apache pinot core common RowBasedBlockValueFetcher lt init gt RowBasedBlockValueFetcher java 32 n tat org apache pinot core operator query SelectionOnlyOperator getNextBlock SelectionOnlyOperator java 82 ```

Jackie · Answer

This is unexpected I think it is caused by adding a sorted column during the consumption

Jackie · Answer

Can you please file an issue for this with the details

Machhindra · Answer

I will try zookeeper exited I restarted it now Restarted server Lets see As I am running in the docker container now I might have lost the data as well

Machhindra · Answer

< Jackie> I restarted the pinot cluster with the clean start This time I did not add the sorted index on metric column There are 52M records I dont face the problem noted above I suspect sorted index is the culprit

Jackie · Answer

< Machhindra> Have you tried starting a new table with sorted index on the metric We don t usually put sorted index on metrics but it should work

Machhindra · Answer

I will try once I finish my benchmarking gt We don t usually put sorted index on metrics Can you please explain why here is my table There are about 1500 unique metrics These metrics are collected at 1 min+ interval Most of the queries will involve metric in predicate ```metric metricValue entityMap metricTime ====================================================================================== CPUTIME 2210 Plan myplan Pkg mypkg Server myDb2 1622076960000 WAITTIME 12 Plan myplan Pkg mypkg Server myDb2 1622076960000 MQBPUSE 50 MqBPool 2 MqQmgr MQA3 1622076960000 MQBPUSE 110 MqBPool 3 MqQmgr MQA3 1622076960000 WLMPI 0 5 ZosWmSysSum jobs 1622076960000 MQQFULL 10 MqQueue myqueue MqQmgr CSQ4 1622076960000 CPUTIME 2220 Plan myplan Pkg mypkg Server myDb2 1622083650000 WAITTIME 12 Plan myplan Pkg mypkg Server myDb2 1622083650000 MQBPUSE 50 MqBPool 2 MqQmgr MQA3 1622083650000 MQBPUSE 110 MqBPool 3 MqQmgr MQA3 1622083650000 WLMPI 0 5 ZosWmSysSum jobs 1622083650000 MQQFULL 10 MqQueue myqueue MqQmgr CSQ4 1622083650000 ```

Jackie · Answer

Oh so `metric` is actually a dimension column

Jackie · Answer

Sorted index should work Please let us know if you run into the same problem after adding the sorted index

Machhindra · Answer

yes I am reading from my Kafka topic I need to flatten the schema I wish if I could create a table as follows ```CPUTIME WAITTIME MQBPUSE 1400 metrics Plan Pkg Server MQBPool MqQmgr ZosWmSysSum metricTime ======================================================================================================================== 2210 myplan mypkg myDb2 1622076960000 12 myplan mypkg myDb2 1622076960000 50 2 MQA3 1622076960000``` This table has METRICS DIMENSIONS and TIMESTAMP Data is similar to prometheus metrics

Jackie · Answer

Any reason why you want to flatten the metrics Flattening all 1500 metrics would cause a too wide table

Neha Pawar · Answer

have you tried using `jsonPathString` from the initial suggestions for getting the fields out of entityMap this is a finite set right `Plan Pkg Server MQBPool MqQmgr ZosWmSysSum`

Machhindra · Answer

Ultimately I want to enable dashboards on this data using Superset or other BI tool Queries are going to be ```get me average of CPUTIME for PLAN Blah and PKG Blah for a time range get me top 10 PLANS which has highest CPUTIME usage get me list of PKGS where PLAN Blah``` I need to expose metrics vs dimensions to user so that with Superset they can easily create charts ```With Short Table and JSON PATH and JSON Index metric metricValue entityMap metricTime ====================================================================================== CPUTIME 2210 Plan myplan Pkg mypkg Server myDb2 1622076960000 WAITTIME 12 Plan myplan Pkg mypkg Server myDb2 1622076960000 Query is going to be SELECT AVG metricValue from metricTable where JSON PATH entityMap Plan= myplan and metric= CPUTIME ``` ```With Wide Table No JSON CPUTIME WAITTIME MQBPUSE 1400 metrics Plan Pkg Server MQBPool MqQmgr ZosWmSysSum metricTime ======================================================================================================================== 2210 myplan mypkg myDb2 1622076960000 12 myplan mypkg myDb2 1622076960000 Query is going to be SELECT AVG CPUTIME from metricTable where Plan= myplan ```

Neha Pawar · Answer

how many distinct metric do you expect and how many distinct fields inside entityMap

Neha Pawar · Answer

you can keep metric and metricValue as it is but still flatten the entityMap

Jackie · Answer

The query for short table should be `SELECT AVG metricValue FROM metricTable WHERE JSON MATCH entityMap Plan= myplan and metric= CPUTIME ` with JSON index

Machhindra · Answer

< Neha Pawar> JSON PATH works BUT look at the way users need to query The wide table query looks very natural Today there are 80 such EntityMap dimensions columns Note that not all dimentions will be filled for each incoming metric CPUTIME metric is for DB2 Plan and Package Where as MQBPUSE is meant for Buffers gt how many distinct metric do you expect and how many distinct fields inside entityMap Distinct Metrics = 1500 Distinct fields inside entityMap = 80

Neha Pawar · Answer

if that is more natural then you can extract them all or maybe the important ones

Neha Pawar · Answer

are there a set of fields in entityMap that are going to be more frequently called than others

Machhindra · Answer

gt The query for short table should be `SELECT AVG metricValue FROM metricTable WHERE JSON MATCH entityMap Plan= myplan and metric= CPUTIME ` with JSON index Yes That s what I am using now In Superset this is how user uses it

Xiang Fu · Answer

In superset you can create columns with pinot expressions

Xiang Fu · Answer

e g a column called `Plan` the expression is `JSON MATCH entityMap Plan= myplan `

Machhindra · Answer

< Neha Pawar> < Xiang Fu> gt if that is more natural then you can extract them all or maybe the important ones Following query works It allows me to filter the dimensions and metric and gets me the average metric value `SELECT AVG metricValue FROM metricTable WHERE JSON MATCH entityMap Plan= myplan and metric= CPUTIME ` However above method does not help if I want to create a new metric CPUTIME WAITTIME 10 What will be the query The wide table enables that possibility e g ```SELECT CPUTIME WAITTIME 1 AS newMetric FROM metricTable WHERE Plan= myplan OR SELECT CPUTIME WAITTIME 1 AS newMetric FROM metricTable WHERE JSON MATCH entityMap Plan= myplan ``` If we keep the dimensions in the JSON format then how would I list all the Plans available ```SELECT JSON MATCH entityMap Plan= myplan AS Plans FROM metricTable ```

Xiang Fu · Answer

from superset side you can create a new column and set the expression then reuse that

Xiang Fu · Answer

you can replace the column with the expression

Xiang Fu · Answer

for `CPUTIME WAITTIME` as they are not in the same record so you need to align the docs then do join to make them same line then multiplier You can use presto to achieve this

Xiang Fu · Answer

for extract unique values youcan do `SELECT distinct JSON MATCH entityMap Plan= myplan AS Plans FROM metricTable` or `SELECT JSON MATCH entityMap Plan= myplan AS Plans count as cnt FROM metricTable GROUP BY JSON MATCH entityMap Plan= myplan AS Plans`

Machhindra · Answer

< Xiang Fu> Above queries does not fetch any result

Xiang Fu · Answer

Hmm ok can you try to play around with the single quote