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

Databricks unit testing

 A simple way to unit test databricks notebooks is to make the notebooks parameterized.

To test if contiguous months are generated we run the contiguous notebook with test input and specify an output location.We then assert to see if the output is expected

Step 1

To run blog_contiguous notebook from test notebook

outputTable = "test_output"
dbutils.notebook.run("blog_contiguous",
timeout_seconds=180, arguments={
"product": "dbfs:/FileStore/tables/test_sales-1.csv",
"output": outputTable,
})

Step 2

Assert values in the test notebook

spark.catalog.refreshTable(outputTable)
output = table(outputTable).cache()
assert output.count() == 12
assert output.groupby(["Year"]).count().collect()[0]['count']==12
assert [output.select("Month").collect()[i]['Month'] for i in range(0,output.count())]==[1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]







Git hub unit test

Databricks hands on create contiguous rows for missing data

Create contiguous rows for missing data

As a data engineer creating ETL pipelines I often have to transform my data to create contiguous data.For instance in this toy example I have data for cornflakes sales for 4 months in 2003 January,February,March and September.I will then have to augment to my data and create rows for the rest of the month for every year. In this article I will show you how to using a community databricks notebook

Consider Corn flake sales data with missing months.

My aim is to generate rows for the missing months per year.To generate contiguous data for all months we first find the months that are missing .We will then create rows for each of those missing months and set the value of the sales to the average per year.

In code speak

df_1=productDF.groupby(["Product","Year"]).agg(F.collect_set("Month"),F.mean("Sales")) df_1=df_1.withColumnRenamed("collect_set(Month)","Months_Sold") df_1=df_1.withColumnRenamed("avg(Sales)","Average_per_year") df_1=df_1.withColumn("Year_Months",F.array([F.lit(i) for i in range(1,13)])) df_1=df_1.withColumn("Left_Months",F.array_except(F.col("Year_Months"),F.col("Months_Sold"))) df_1=(df_1.withColumn("Month",F.explode(F.col("Left_Months")))) df_1=df_1.withColumn("Sales",F.col("Average_per_year")) df_2=df_1.select("Month","Year","Product","Sales") df_all=productDF.union(df_2).orderBy([F.col("Year"),F.col("Month")]) 

Git hub source :Contiguous data