Azure 流分析错误：无法反序列化来自 IOT 中心的输入事件

Posted 2023-05-08

【中文标题】Azure 流分析错误：无法反序列化来自 IOT 中心的输入事件

【英文标题】：Azure Stream Analytics Error : Could not deserialize the input event(s) from IOT hub

【发布时间】：2018-01-11 12:02:44

【问题描述】：

我创建了流分析作业以从 IOT 集线器读取数据作为输入并将数据存储到 SQL DB。

以下是为 Steam 分析作业配置的一些重要输入详细信息： 事件序列化格式：JSON 编码：utf-8

消息从 Dotnet 模拟代码发送到 IOT Hub。

当我运行我的工作时，我收到以下错误：

 Could not deserialize the input event as Json. Some possible reasons: 
 1) Malformed events 
 2) Input source configured with incorrect serialization format

这是我的 dotnet 代码。

       private static async void ReceiveC2dAsync()
    
        Console.WriteLine("\nReceiving cloud to device messages from service");
        while (true)
        
            Message receivedMessage = await deviceClient.ReceiveAsync();
            if (receivedMessage == null) continue;

            Console.ForegroundColor = ConsoleColor.Yellow;
            Console.WriteLine("Received message: 0", Encoding.ASCII.GetString(receivedMessage.GetBytes()));
            Console.ResetColor();

            await deviceClient.CompleteAsync(receivedMessage);
        
    
    private static async void SendDeviceToCloudMessagesAsync()
    
        double minTemperature = 20;
        double minHumidity = 60;
        int messageId = 1;
        Random rand = new Random();

        while (true)
        
            double currentTemperature = minTemperature + rand.NextDouble() * 15;
            double currentHumidity = minHumidity + rand.NextDouble() * 20;

            var telemetryDataPoint = new
            
                messageId = messageId++,
                deviceId = "myFirstDevice",
                temperature = currentTemperature,
                humidity = currentHumidity
            ;
            var messageString = JsonConvert.SerializeObject(telemetryDataPoint);
            string levelValue;
            string temperatureAlert = "false";


            if (rand.NextDouble() > 0.7)
            
                if (rand.NextDouble() > 0.5)
                
                    messageString = "This is a critical message";
                    levelValue = "critical";
                
                else
                
                    messageString = "This is a storage message";
                    levelValue = "storage";
                
            
            else
            
                levelValue = "normal";
            

            if(currentTemperature > 30)
            
                temperatureAlert = "true";
            

            var message = new Message(Encoding.UTF8.GetBytes(messageString));
            message.Properties.Add("level", levelValue);
            message.Properties.Add("temperatureAlert", temperatureAlert);

            await deviceClient.SendEventAsync(message);
            Console.WriteLine("0 > Sending message: 1", DateTime.Now, messageString);

            await Task.Delay(1000);
        
    

【问题讨论】：

您的模拟设备似乎生成了非 json 格式的消息，例如“这是一条关键消息”和“这是一条存储消息”。基本上，您必须选择：1. 在模拟代码中注释这部分，或者 2. 在 Azure IoT Hub Routes 中为这些消息添加过滤器。

@raghu456，您可以尝试将json数据保存到一个文件（命名为*.json），然后上传文件进行测试。请参考Azure Stream Analytics query testing and input-stream sampling。如果数据在json格式正确，查询成功。

感谢 Roman 指出错误。这解决了我的问题。

【参考方案1】：

您的模拟设备似乎生成了非 json 格式的消息，例如“这是一条关键消息”和“这是一条存储消息”。

基本上，您有两种选择来解决此问题： 1.在模拟代码中注释这部分或 2. 在 Azure IoT Hub Routes 中为这些消息添加过滤器