Stream Analytics Job Output Stuck

Lior Bahat 1

I have a stream analytics job with an MQTT JSON input and a postgreSQL output.
I noticed that if a certain JSON parameter that is part of the message no output rows are generated. I couldn't find any related info in the logs. Restarting the stream analytics job fixes the issue until the next message with this parameter.
The only difference I see between this parameter and the previous ones is that it is an INT while the previous ones are VARCHAR and FLOAT.

This is the JSON (the problematic parameter is "tsen1":
{\"ppb\":117.35,\"pid\":155.40,\"temp\":22.50,\"humid\":29.10,\"pres\":1002.00,\"wspd\":0.00,\"wdir\":359.60,\"sflag\":0,\"tflag\":0,\"pstat\":1,\"estat\":0,\"batt\":13.31,\"chrg\":0.00,\"run\":113.86,\"tsen1\":2440,\"heat1\":15,\"tset1\":2492,\"sd\":1000,\"tcrh\":0,\"r232\":0,\"csq\":-92,\"snr\":5,\"brdv\":5,\"fwv\":594,\"misc\":166,\"cell\":115000,\"samp\":2,\"trig\":10,\"tstatic\":250,\"btime\":30,\"tl1\":250,\"tt1\":45,\"ds1\":0,\"df1\":360,\"ws1\":0,\"ev1\":129,\"ev2\":0,\"ev3\":0,\"ev4\":0}}"
}
}

Sander van de Velde 28,386 Reputation points MVP

2021-11-12T17:54:01.527+00:00

Hello @Lior Bahat ,

so you are ingesting telemetry but the Azure Stream Analytics is not able to cope with the incoming messages.

How do you ingest the data? Do you use Azure resources (eg IoT hub or event hub) or a custom deserializer?

Have you checked the logging of your service already?

Do you trust your query to support the (new) format? Have you tested this in eg. VS Code or the test page in the portal?
Lior Bahat 1 Reputation point

2021-11-12T21:26:42.81+00:00

Hi,
I'm ingesting the data thought the IoT hub.
I couldn't find information regarding the issue in the logs. I couldn't find data in the logs. I am new to this environment and tried to follow the documentation but I couldn't configure the diagnostic settings although I created a log analytics workspace.

I am not sure what you mean be "new format". There is one object but in some cases only a subset of the parameters is sent from the device (The other parameters are slow-changing and are sent at larger intervals). I did test the query on the input in the portal test page and it works. 148994-azurestreamanalyticsquery.txt

Thanks for your help!
Sander van de Velde 28,386 Reputation points MVP

2021-11-13T00:28:03.347+00:00

Hello @Lior Bahat ,

The query is just passing through the incoming type/format of the messages. That should not cause that much of a problem...

I could think of a number of reasons why rows are missing...

Because you ingest from an IoT Hub, did you use a separate consumer group for Azure Stream Analytics?

When message arrive at the IoT Hub, these message are routed to every connected service (eg. an Azure Function, Time Series Insight or Stream Analytics).

Each service wants a copy of the same message, hence you have to define multiple consumer groups in the IoT Hub:

If you do not add a separate consumer group, other services, including the Azure IoT Explorer tool could 'steal' messages...

Next to that, are your sure the service behind the output will accept all messages, even when column types change?
Lior Bahat 1 Reputation point

2021-11-13T09:26:34.927+00:00

Hi,
I will follow up on your advice regarding a separate consumer group. Nonetheless it is strange that things work until a message with certain parameters is passed.
Moreover, I had a look inside the logs of the postgreSQL server and found the following messages:

2021-11-13 09:09:31 UTC-618f80b2.b810-STATEMENT: COPY public.spod_tbl ("device_id","sample_time","ppb","pid","temp","humid","pres","wspd","wdir","sflag","tflag","pstat","estat","batt","chrg","run","tsen1","heat1","tset1","sd","tcrh","r232","csq","snr","brdv","fwv","misc","cell","samp","trig","tstatic","btime","tl1","tt1","ds1","df1","ws1","ev1","ev2","ev3","ev4") FROM STDIN (FORMAT BINARY)
2021-11-13 09:09:31 UTC-618f80b2.b810-ERROR: incorrect binary data format
2021-11-13 09:09:31 UTC-618f80b2.b810-CONTEXT: COPY spod_tbl, line 1, column tsen1

This is referring to the exact parameter causing the problem. From this point the error is repeated.
Any ideas on how to debug this?

Thanks!

2 answers

Sander van de Velde 28,386 Reputation points MVP

2021-11-13T10:14:29.077+00:00
Hello @Lior Bahat ,

We see this postgreSQL error:

2021-11-13 09:09:31 UTC-618f80b2.b810-ERROR: incorrect binary data format

The The Stream Analytics query is passing through all messages, it seems specific incoming data is rejected by PostgreSQL.

Using the query language of Stream Analytics, you can change the output format of individual columns by eg. casting to an int or double.

the PostgreSQL output is just release and still in public preview.
Please sign in to rate this answer.
Lior Bahat 1 Reputation point

2021-11-13T10:23:58.897+00:00

Great!

To speed up my debugging:

Can I view the messages sent from the stream analytics to the postgres server?

Can I input the data from file (that contains sampled input data) instead of working with the device? I used it for testing but can I have it as input to the stream analytics?

Thanks!

Sander van de Velde 28,386 Reputation points MVP

2021-11-13T15:17:56.303+00:00

Hello @Lior Bahat ,

Azure Stream Analytics is a service that passes through events, it has no internal storage. Therefor, historical messages are not stored.

Because you are using Azure IoT Hub (with a retention time (with a standard retention time of 1 day), you could try to rerun older messages by adjusting the start time:

If you create a second output to an event hub or blob storage account and you send the same query output to these resources, you can check out what is sent. The services accept every message so no exceptions there.

Azure Storage explorer:

There are several ways to test your Stream Analytics job.

You can record live data on an IoT Hub input:

You can then upload and test with that data in the portal.

A more professional way of working with Stream Analytics is using the VS Code Stream Analytics extension.

This also supports multiple testing scenarios.
Sign in to comment
67391162 0 Reputation points

2024-02-01T14:41:54.27+00:00

Bonjour, est-ce que je peux configurer un stream analytics job avec un input « iot hub » et un output « mqtt » pour envoyer les données traitées à mon app NodeJS ?
Please sign in to rate this answer.
Sander van de Velde 28,386 Reputation points MVP

2024-02-01T17:09:42.71+00:00

Auto translation:

Hello, can I configure a stream analytics job with an input “iot hub” and an output “mqtt” to send the processed data to my NodeJS app?

Please start a new thread with your own question instead of adding them to an existing one created for another question.

Thanks.

On topic: Azure Stream Analytics is an engine for filtering, routing, enriching, joining, etc. data.

If you want to send messages to a MQTT client, you need something like this:

data -> stream analytics job -> some logic (like an Azure function) -> The new MQTT support in EventGrid <- devices listening to that broker.
Sign in to comment