how to process data based on the partition key using azure stream analytics?

Question

I'm sending batch data to the event hub from web jobs written in node js and I can be able to receive the data at Azure stream analytics. But I cannot be able to see the partition key at ASA. I'm using the following code to send the data to the event hub.

const {EventHubClient} = require("@azure/event-hubs");
var axios = require('axios');
const connectionString = "Endpoint=connectionstring";
const eventHubsName = "eventhubname";
async function main(){
    const client = EventHubClient.createFromConnectionString(connectionString,eventHubsName);
    var axios = require('axios');
    var response = await axios.get('https://nseindia.com/live_market/dynaContent/live_watch/stock_watch/nifty500StockWatch.json');
    var response1 = await axios.get('https://www.nseindia.com/live_market/dynaContent/live_analysis/gainers/niftyGainers1.json');
    var eventData = [{body:response1['data']['data'],partitionKey:"pk12346"},{body:response['data']['data'],partitionKey:"pk12345"}];
    console.log(eventData);
    console.log("begin send...");
    await client.sendBatch(eventData);
    await client.close();
}
main().catch(err =>{
    console.log("Error occurred: ",err);
});

data inside response1['data']['data'] and response1['data']['data'] is an array of object data. for example [{key1:value1,key2:valu2},{{key1:value1,key2:valu2}}]

If I'm not able to see the partition at ASA(while getting the sample data from input) how can use where condition for a partitionkey and apply trnsformation.

kgalic kgalic · Accepted Answer · 2019-10-09T08:38:59

You need to write a query that extracts the array data. Below is the example from one of my POCs:

with deviceAndMessage as (Select IoTHub.ConnectionDeviceId as deviceId, * from inputData),
 unpackedmessages as (Select deviceAndMessage.Deviceid, message.ArrayValue.DisplayName, message.ArrayValue.Value.Value, message.ArrayValue.Value.SourceTimestamp from deviceAndMessage CROSS APPLY GetArrayElements(deviceAndMessage.message) AS message),
 inputWeatherData as (select deviceId, unpackedmessages.DisplayName, unpackedmessages.Value, unpackedmessages.SourceTimestamp from unpackedmessages),
 humidity as (select * from inputWeatherData where DisplayName = 'Humidity')

The structure of my message was

{
    "message" : [
       {
           "DisplayName" : "Test",
           "Value" : {
               "SourceTimestamp" : "whatever",
               "Value" : 1
           }
       },
       {
           "DisplayName" : "Test",
           "Value" : {
               "SourceTimestamp" : "whatever",
               "Value" : 1
           }
       }
    ]
}

Maybe you can use this example and adjust it for your data structure.

how to process data based on the partition key using azure stream analytics?

1 Answers