如何使用 azure 流分析基于分区键处理数据？答案

【问题标题】：how to process data based on the partition key using azure stream analytics?如何使用 azure 流分析基于分区键处理数据？
【发布时间】：2020-02-01 18:36:15
【问题描述】：

我正在从用节点 js 编写的 Web 作业向事件中心发送批处理数据，并且我可以在 Azure 流分析中接收数据。但我无法在 ASA 上看到分区键。我正在使用以下代码将数据发送到事件中心。

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);
});

response1['data']['data'] 和 response1['data']['data'] 中的数据是一个对象数据数组。例如 [{key1:value1,key2:value2},{{key1:value1,key2:value2}}]

如果我无法在 ASA 上看到分区（从输入获取样本数据时）如何使用 where 条件作为分区键并应用 trnsformation。

【问题讨论】：

标签： node.js azure-webjobs azure-eventhub azure-stream-analytics

【解决方案1】：

您需要编写一个提取数组数据的查询。以下是我的一个 POC 的示例：

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')

我的信息结构是

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

也许您可以使用此示例并针对您的数据结构进行调整。

【讨论】：