Azure stream analytics JSON Flattening

I was working on an azure stream analytics project which processed device telemetry data from event hub.The input has multiple JSON Arrays and I needed to convert my output to a wide format csv file.In this blog I will walk you through how to do this. Azure Stream Analytics offers a SQL query language for performing transformations and computations over streams of events. This is a subset of T-SQL syntax.For my current requirement this fitted well.One can directly write the query in the azure portal and test it there.I also found Visual studio code extensions for Azure stream analytics allows us to craft the query locally with sample mock input data . The problem-I need to flatten the Input json which has temperature readings and pollution readings from a sensor.These are arrays which may be of differing length The sample input JSON

  
  {
    "sensor_readings": {
        "readings":{
        "temperature_readings": [
            {
                "date": "02-02-2020",
                "hour": "12",
                "second": "00",
                "temperature": "1.12"
            },
            {
                "date": "02-02-2020",
                "hour": "13",
                "second": "00",
                "temperature": "2.12"
            }
        ]
    
,
    "air_pollution_readings": 
        [
            {
                "date": "02-02-2020",
                "hour": "12",
                "second": "00",
                "element":"o3",
                "particulate": "2.2"
            },
            {
                "date": "02-02-2020",
                "hour": "13",
                "second": "00",
                "element":"o3",
                "particulate": "2.1"
            },
            {
                "date": "02-02-2020",
                "hour": "14",
                "second": "00",
                "element":"o3",
                "particulate": "1.1"
            }
            
            
        ]}
    }
    ,
    "siteid": "a1234566",
    "deviceid": "2343434"
}

For the first step we need to iterate through all the array elements in each of the arrays This can be done with the GetArrayElements array function

with 
temp_q as(
SELECT 
    temp_sensor.ArrayValue as av,
    temp_sensor.ArrayIndex as ax,
    e.siteid,
    e.deviceid,
    'temp' as type 
FROM 
    Input1 e 
    outer apply 
    GetArrayElements(e.sensor_readings.readings.temperature_readings) as temp_sensor 
)

This will flatten the temperature_readings JSON Array Same goes with the air pollution sensor data

  air_q as(
SELECT 
    temp_sensor.ArrayValue as av,
    temp_sensor.ArrayIndex as ax,
    e.siteid,
    e.deviceid,
    'air' as type 
FROM 
    Input1 e 
    outer apply 
    GetArrayElements(e.sensor_readings.readings.air_pollution_readings) as temp_sensor 
), 

  tot_q as(
select 
    * 
from 
    temp_q 
union 
SELECT 
    * 
from 
    air_q 
)
 

All array elements in long format for both temperature and air_pollution sensor Output will look like

{"av":{"date":"02-02-2020","hour":"12","second":"00","temperature":"1.12"},"ax":0,"siteid":"a1234566","deviceid":"2343434","type":"temp"}
{"av":{"date":"02-02-2020","hour":"13","second":"00","temperature":"2.12"},"ax":1,"siteid":"a1234566","deviceid":"2343434","type":"temp"}
{"av":{"date":"02-02-2020","hour":"12","second":"00","element":"o3","particulate":"2.2"},"ax":0,"siteid":"a1234566","deviceid":"2343434","type":"air"}
{"av":{"date":"02-02-2020","hour":"13","second":"00","element":"o3","particulate":"2.1"},"ax":1,"siteid":"a1234566","deviceid":"2343434","type":"air"}
{"av":{"date":"02-02-2020","hour":"14","second":"00","element":"o3","particulate":"1.1"},"ax":2,"siteid":"a1234566","deviceid":"2343434","type":"air"}

Now we want to convert long format to wide .This can be done with the WITH syntax in addition to the JOIN operation. JOINs only allow LEFT OUTER and INNER joins. We will first need to know which of the arrays are larger and which are smaller .We will then join first using the larger query

 processed_q as(
select 
    case when p.temp_count > = p.air_count then 'temp' else 'air' end larger,
    case when p.temp_count < p.air_count then 'temp' else 'air' end smaller,
    * 
from 
    (
    SELECT 
        t.deviceid,
        tot_q.ax,
        tot_q.type,
        tot_q.av,
        GetArrayLength(t.sensor_readings.readings.temperature_readings) as temp_count,
        GetArrayLength(t.sensor_readings.readings.air_pollution_readings) as air_count 
    from 
        Input1 t 
        left outer join tot_q 
        on tot_q.deviceid = t.deviceid AND DATEDIFF(minute, t, tot_q) BETWEEN 0 AND 0 
    ) p 
), 
first_join as(
select 
    * 
from 
    processed_q 
where 
    type = larger 
),

To the output obtained we join the smaller and filter out rows where the array pos are not same since its a left outer it will pick up all the rows from the larger array.This will give us the required output

  reqd_join as(
select 
    n.smaller,
    n.larger,
    n2.type,
    n.deviceid,
    n.ax,
    n.av,
    n2.ax as smaller_ax,
    n2.av as smaller_av 
from 
    first_join n 
    left outer join tot_q n2 
    on n.deviceid = n2.deviceid and n2.type = n.smaller AND DATEDIFF(minute, n2, n) BETWEEN 0 AND 0 and n.ax = n2.ax 
) 

  
  

Output Test Same number of rows in both arrays

  
{"smaller":"air","larger":"temp","type":"air","deviceid":"2343434","ax":0,"av":{"date":"02-02-2020","hour":"12","second":"00","temperature":"1.12"},"smaller_ax":0,"smaller_av":{"date":"02-02-2020","hour":"12","second":"00","element":"o3","particulate":"2.2"}}
{"smaller":"air","larger":"temp","type":"air","deviceid":"2343434","ax":1,"av":{"date":"02-02-2020","hour":"13","second":"00","temperature":"2.12"},"smaller_ax":1,"smaller_av":{"date":"02-02-2020","hour":"13","second":"00","element":"o3","particulate":"2.1"}}
  
  

Temperature array has more rows than air pollution array .Note that the air pollution smaller_ax and smaller_av is null

  
{"smaller":"air","larger":"temp","type":"air","deviceid":"2343434","ax":0,"av":{"date":"02-02-2020","hour":"12","second":"00","temperature":"1.12"},"smaller_ax":0,"smaller_av":{"date":"02-02-2020","hour":"12","second":"00","element":"o3","particulate":"2.2"}}
{"smaller":"air","larger":"temp","type":"air","deviceid":"2343434","ax":1,"av":{"date":"02-02-2020","hour":"13","second":"00","temperature":"2.12"},"smaller_ax":1,"smaller_av":{"date":"02-02-2020","hour":"13","second":"00","element":"o3","particulate":"2.1"}}
{"smaller":"air","larger":"temp","type":null,"deviceid":"2343434","ax":2,"av":{"date":"02-02-2020","hour":"14","second":"00","temperature":"3.12"},"smaller_ax":null,"smaller_av":null}
  
  

Air pollution array has more rows.

  
{"smaller":"temp","larger":"air","type":"temp","deviceid":"2343434","ax":0,"av":{"date":"02-02-2020","hour":"12","second":"00","element":"o3","particulate":"2.2"},"smaller_ax":0,"smaller_av":{"date":"02-02-2020","hour":"12","second":"00","temperature":"1.12"}}
{"smaller":"temp","larger":"air","type":"temp","deviceid":"2343434","ax":1,"av":{"date":"02-02-2020","hour":"13","second":"00","element":"o3","particulate":"2.1"},"smaller_ax":1,"smaller_av":{"date":"02-02-2020","hour":"13","second":"00","temperature":"2.12"}}
{"smaller":"temp","larger":"air","type":null,"deviceid":"2343434","ax":2,"av":{"date":"02-02-2020","hour":"14","second":"00","element":"o3","particulate":"3.1"},"smaller_ax":null,"smaller_av":null}
  

Github