Azure Stream Analytics Querying Basics

00_Setup.sh

#!/bin/bash
AZ_RESOURCE_GROUP_DEFAULT=MyTollBooth
AZ_COSMOSDB_ACCOUNT_NAME_DEFAULT=

read -p "Azure Resource Group [$AZ_RESOURCE_GROUP_DEFAULT]: " AZ_RESOURCE_GROUP
AZ_RESOURCE_GROUP=${AZ_RESOURCE_GROUP:-$AZ_RESOURCE_GROUP_DEFAULT}

read -p "Azure Cosmos DB Account Name [$AZ_COSMOSDB_ACCOUNT_NAME_DEFAULT]: " AZ_COSMOSDB_ACCOUNT_NAME
AZ_COSMOSDB_ACCOUNT_NAME=${AZ_COSMOSDB_ACCOUNT_NAME:-$AZ_COSMOSDB_ACCOUNT_NAME_DEFAULT}

export cosmos_sb_account_name=''
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c tollAppCollection
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c filtered --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c transit_duration --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c missing_exits --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c tumbling_window --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c hopping_window --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c high_performers --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c rush_hour --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600
az cosmosdb collection create -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c expired_registrations --partition-key-path '/toll_id' --throughput 400 --default-ttl 3600

01_FilteringAndProjection.sql

SELECT  TollId AS toll_id, 
        LicensePlate AS license_plate, 
        State, 
        CarModel.Make AS vehicle_make,
        CarModel.Model AS vehicle_model,
        CASE CarModel.VehicleType
            WHEN 1 THEN 'Passenger'
            WHEN 2 THEN 'Commercial'
            ELSE 'Other' 
        END AS vehicle_category,
        CarModel.VehicleWeight AS vehicle_weight,
        DATEPART(mi,EntryTime) AS minutes,
        DATEPART(ss,EntryTime) AS seconds,
        DATEPART(ms,EntryTime) AS milleseconds
INTO    FilteredOutput
FROM    EntryStream TIMESTAMP BY EntryTime
WHERE   (State = 'CA' OR State = 'WA') 
AND     CarModel.VehicleWeight < 3000 
AND     CHARINDEX ('T',  CarModel.Model) = 1

02_TemporalJoin.sql

SELECT  ES.TollId AS toll_id, 
        ES.EntryTime AS entry_time, 
        EX.ExitTime AS exit_time, 
        ES.LicensePlate AS license_plate, 
        DATEDIFF(Second, ES.EntryTime, EX.ExitTime) AS transit_duration
INTO    TransitDurationOutput
FROM    EntryStream ES TIMESTAMP BY EntryTime
JOIN    ExitStream EX TIMESTAMP BY ExitTime
ON      (ES.TollId=EX.TollId AND ES.LicensePlate=EX.LicensePlate)
AND     DATEDIFF(Minute, ES, EX) BETWEEN 0 AND 15

03_TemporalLeftOuterJoin.sql

SELECT  ES.TollId AS toll_id, 
        ES.EntryTime AS entry_time, 
        ES.LicensePlate AS license_plate
INTO    MissingExitsOutput
FROM    EntryStream ES TIMESTAMP BY EntryTime
LEFT    OUTER JOIN ExitStream EX TIMESTAMP BY ExitTime
ON      (ES.TollId=EX.TollId AND ES.LicensePlate=EX.LicensePlate)
AND     DATEDIFF(Minute, ES, EX) BETWEEN 0 AND 5
WHERE   EX.ExitTime IS NULL

04_TumblingWindow.sql

SELECT  TollId AS toll_id, 
        System.Timestamp AS window_end, 
        COUNT(*) AS count
INTO    TumblingWindowOutput
FROM    EntryStream TIMESTAMP BY EntryTime
GROUP   BY TollId, TumblingWindow(second, 10)

05_HoppingWindow.sql

SELECT  TollId AS toll_id, 
        DateAdd(second,-10,System.Timestamp()) AS window_start,
        System.Timestamp AS window_end, 
        COUNT(*) AS count
INTO    HoppingWindowOutput
FROM    EntryStream TIMESTAMP BY EntryTime
GROUP   BY TollId, HoppingWindow(second, 10, 5)

06_SlidingWindow.sql

SELECT  TollId AS toll_id, 
        DateAdd(second,-20,System.Timestamp()) AS window_start,
        System.Timestamp AS window_end, 
        COUNT(*) AS count
INTO    HighPerformersOutput
FROM    EntryStream TIMESTAMP BY EntryTime
GROUP   BY TollId, SlidingWindow(second, 20)
HAVING  COUNT(*) > 10

07_SessionWindow.sql

SELECT  TollId AS toll_id, 
        MIN(EntryTime) AS window_start,
        System.Timestamp AS window_end, 
        COUNT(*) AS count,
        DATEDIFF(s, MIN(EntryTime), System.Timestamp()) AS duration_in_seconds
INTO    RushHourOutput
FROM    EntryStream TIMESTAMP BY EntryTime
GROUP   BY TollId, SessionWindow(minute, 5, 60) OVER (PARTITION BY toll_id)

08_Lookup.sql

SELECT  ES.TollId AS toll_id, 
        ES.EntryTime AS entry_time, 
        ES.LicensePlate AS license_plate,
        R.RegistrationId AS registration_id
INTO    ExpiredRegistrationsOutput
FROM    EntryStream ES TIMESTAMP BY EntryTime
JOIN    Registration R
ON      ES.LicensePlate = R.LicensePlate
WHERE   R.Expired = '1'

99_CleanUp.sh

#!/bin/bash
AZ_RESOURCE_GROUP_DEFAULT=MyTollBooth
AZ_COSMOSDB_ACCOUNT_NAME_DEFAULT=

read -p "Azure Resource Group [$AZ_RESOURCE_GROUP_DEFAULT]: " AZ_RESOURCE_GROUP
AZ_RESOURCE_GROUP=${AZ_RESOURCE_GROUP:-$AZ_RESOURCE_GROUP_DEFAULT}

read -p "Azure Cosmos DB Account Name [$AZ_COSMOSDB_ACCOUNT_NAME_DEFAULT]: " AZ_COSMOSDB_ACCOUNT_NAME
AZ_COSMOSDB_ACCOUNT_NAME=${AZ_COSMOSDB_ACCOUNT_NAME:-$AZ_COSMOSDB_ACCOUNT_NAME_DEFAULT}

az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c filtered
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c transit_duration
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c missing_exits
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c tumbling_window
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c hopping_window
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c high_performers
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c rush_hour
az cosmosdb collection delete -g $AZ_RESOURCE_GROUP -n $AZ_COSMOSDB_ACCOUNT_NAME -d tollAppDatabase -c expired_registrations