Azure Stream Analytics で JSON データと Avro データを解析する

[アーティクル]
06/02/2023

Azure Stream Analytics では、CSV、JSON、および Avro データ形式のイベントの処理をサポートしています。 JSON データと Avro データのどちらも、入れ子になったオブジェクト (レコード) や配列などの複合型を含む構造にすることができます。

Note

Event Hub Capture によって作成される AVRO ファイルは、カスタム逆シリアライザー 機能を使用する必要がある特定の形式を使用します。詳細については、.NET カスタム逆シリアライザーを使用して任意の形式の入力を読み取るを参照してください。

レコードデータ型

レコードデータ型は、対応する形式が入力データストリームで使用される場合に、JSON 配列と Avro の配列を表すために使用されます。これらの例は、JSON 形式の入力イベントを読み取るサンプルセンサーを示しています。 1 つのイベントの例を以下に示します。

{
    "DeviceId" : "12345",
    "Location" :
    {
        "Lat": 47,
        "Long": 122
    },
    "SensorReadings" :
    {
        "Temperature" : 80,
        "Humidity" : 70,
        "CustomSensor01" : 5,
        "CustomSensor02" : 99,
        "SensorMetadata" : 
        {
        "Manufacturer":"ABC",
        "Version":"1.2.45"
        }
    }
}

既知のスキーマの入れ子のフィールドにアクセスする

クエリから直接、入れ子になったフィールドに簡単にアクセスするには、ドット表記 (.) を使用します。たとえば、このクエリでは、上記の JSON データの Location プロパティの緯度と経度の座標が選択されます。次に示したように、ドット表記を使用してさまざまなレベルに移動できます。

SELECT
    DeviceID,
    Location.Lat,
    Location.Long,
    SensorReadings.Temperature,
    SensorReadings.SensorMetadata.Version
FROM input

結果は次のとおりです。

DeviceID	Lat	Long	気温	Version
12345	47	122	80	1.2.45

すべてのプロパティを選択する

'*' ワイルドカードを使用すると、入れ子になったレコードのすべてのプロパティを選択できます。次の例を確認してください。

SELECT
    DeviceID,
    Location.*
FROM input

結果は次のとおりです。

DeviceID	Lat	Long
12345	47	122

プロパティ名が変数であるときに入れ子のフィールドにアクセスする

プロパティ名が変数の場合は、GetRecordPropertyValue 関数を使用します。これにより、プロパティ名をハードコーディングすることなく、動的なクエリを作成できます。

たとえば、サンプルデータストリームを、各デバイスセンサーのしきい値を含む参照データと結合する必要があるとします。そのような参照データのスニペットを次に示します。

{
    "DeviceId" : "12345",
    "SensorName" : "Temperature",
    "Value" : 85
},
{
    "DeviceId" : "12345",
    "SensorName" : "Humidity",
    "Value" : 65
}

ここでの目的は、記事の先頭にあるサンプルデータセットをその参照データに結合し、各センサーメジャーに対してしきい値を上回る 1 つのイベントを出力することです。これは、結合により、複数のセンサーがそれぞれのしきい値を超えた場合に、上記の 1 つのイベントによって複数の出力イベントが生成される可能性があることを意味します。結合せずに同様の結果を得るには、以下のセクションを参照してください。

SELECT
    input.DeviceID,
    thresholds.SensorName,
    "Alert: Sensor above threshold" AS AlertMessage
FROM input      -- stream input
JOIN thresholds -- reference data input
ON
    input.DeviceId = thresholds.DeviceId
WHERE
    GetRecordPropertyValue(input.SensorReadings, thresholds.SensorName) > thresholds.Value

GetRecordPropertyValueにより、 SensorReadingsのプロパティを選択します。この名前は参照データから取得したプロパティ名と一致します。次に、SensorReadings の関連する値が抽出されます。

結果は次のとおりです。

DeviceID	SensorName	AlertMessage
12345	湿度	注意：しきい値を超えたセンサー

レコードフィールドを個々のイベントに変換する

レコードフィールドを個々のイベントに変換するには、APPLY 演算子を GetRecordProperties 関数と組み合わせて使用します。

元のサンプルデータでは、次のクエリを使用して、さまざまなイベントにプロパティを抽出できます。

SELECT
    event.DeviceID,
    sensorReading.PropertyName,
    sensorReading.PropertyValue
FROM input as event
CROSS APPLY GetRecordProperties(event.SensorReadings) AS sensorReading

結果は次のとおりです。

DeviceID	SensorName	AlertMessage
12345	気温	80
12345	湿度	70
12345	CustomSensor01	5
12345	CustomSensor02	99
12345	SensorMetadata	[object Object]

WITHを使用すると、これらのイベントを異なる宛先にルーティングできます。

WITH Stage0 AS
(
    SELECT
        event.DeviceID,
        sensorReading.PropertyName,
        sensorReading.PropertyValue
    FROM input as event
    CROSS APPLY GetRecordProperties(event.SensorReadings) AS sensorReading
)

SELECT DeviceID, PropertyValue AS Temperature INTO TemperatureOutput FROM Stage0 WHERE PropertyName = 'Temperature'
SELECT DeviceID, PropertyValue AS Humidity INTO HumidityOutput FROM Stage0 WHERE PropertyName = 'Humidity'

SQL 参照データの JSON レコードを解析する

ジョブで参照データとして Azure SQL Database を使用する場合、JSON 形式のデータを含む列を持つことができます。次に例を示します。

DeviceID	Data
12345	{"key": "value1"}
54321	{"key": "value2"}

単純な JavaScript ユーザー定義関数を記述することで、Data 列の JSON レコードを解析できます。

function parseJson(string) {
return JSON.parse(string);
}

次に、以下に示すように Stream Analytics クエリのステップを作成して、JSON レコードのフィールドにアクセスできます。

WITH parseJson as
(
SELECT DeviceID, udf.parseJson(sqlRefInput.Data) as metadata,
FROM sqlRefInput
)

SELECT metadata.key
INTO output
FROM streamInput
JOIN parseJson 
ON streamInput.DeviceID = parseJson.DeviceID

配列データ型

配列データ型は、順序が付けられた値のコレクションです。配列値の一般的な操作の詳細を以下に示します。これらの例では、関数 GetArrayElement、GetArrayElements、GetArrayLength、および APPLY 演算子を使用しています。

以下にイベントの例を示します。 CustomSensor03 と SensorMetadata のどちらも配列型です。

{
    "DeviceId" : "12345",
    "SensorReadings" :
    {
        "Temperature" : 80,
        "Humidity" : 70,
        "CustomSensor01" : 5,
        "CustomSensor02" : 99,
        "CustomSensor03": [12,-5,0]
     },
    "SensorMetadata":[
        {          
            "smKey":"Manufacturer",
            "smValue":"ABC"                
        },
        {
            "smKey":"Version",
            "smValue":"1.2.45"
        }
    ]
}

特定の配列要素を操作する

指定したインデックス位置にある配列の要素を選択します (配列の最初の要素を選択します)。

SELECT
    GetArrayElement(SensorReadings.CustomSensor03, 0) AS firstElement
FROM input

結果は次のとおりです。

firstElement
12

配列の長さを選択する

SELECT
    GetArrayLength(SensorReadings.CustomSensor03) AS arrayLength
FROM input

結果は次のとおりです。

arrayLength
3

配列要素を個々のイベントに変換する

配列のすべての要素を個々のイベントとして選択します。 APPLY 演算子が GetArrayElements 組み込み関数と組み合わされて、配列のすべての要素を個々のイベントとして抽出します。

SELECT
    DeviceId,
	CustomSensor03Record.ArrayIndex,
	CustomSensor03Record.ArrayValue
FROM input
CROSS APPLY GetArrayElements(SensorReadings.CustomSensor03) AS CustomSensor03Record

結果は次のとおりです。

deviceId	ArrayIndex	ArrayValue
12345	0	12
12345	1	-5
12345	2	0

SELECT   
    i.DeviceId,	
    SensorMetadataRecords.ArrayValue.smKey as smKey,
    SensorMetadataRecords.ArrayValue.smValue as smValue
FROM input i
CROSS APPLY GetArrayElements(SensorMetadata) AS SensorMetadataRecords

結果は次のとおりです。

deviceId	smKey	smValue
12345	Manufacturer	ABC
12345	Version	1.2.45

抽出されたフィールドを列に表示する必要がある場合は、JOIN 操作に加えて、WITH構文を使用してデータセットをピボットすることができます。この結合を行うには、重複を防ぐ時間的境界の条件が必要です。

WITH DynamicCTE AS (
	SELECT   
		i.DeviceId,
		SensorMetadataRecords.ArrayValue.smKey as smKey,
		SensorMetadataRecords.ArrayValue.smValue as smValue
	FROM input i
	CROSS APPLY GetArrayElements(SensorMetadata) AS SensorMetadataRecords 
)

SELECT
	i.DeviceId,
	i.Location.*,
	V.smValue AS 'smVersion',
	M.smValue AS 'smManufacturer'
FROM input i
LEFT JOIN DynamicCTE V ON V.smKey = 'Version' and V.DeviceId = i.DeviceId AND DATEDIFF(minute,i,V) BETWEEN 0 AND 0 
LEFT JOIN DynamicCTE M ON M.smKey = 'Manufacturer' and M.DeviceId = i.DeviceId AND DATEDIFF(minute,i,M) BETWEEN 0 AND 0

結果は次のとおりです。

deviceId	Lat	Long	smVersion	smManufacturer
12345	47	122	1.2.45	ABC

参照

Azure Stream Analytics でのデータ型

Azure Stream Analytics で JSON データと Avro データを解析する

レコード データ型

既知のスキーマの入れ子のフィールドにアクセスする

すべてのプロパティを選択する

プロパティ名が変数であるときに入れ子のフィールドにアクセスする

レコード フィールドを個々のイベントに変換する

SQL 参照データの JSON レコードを解析する

配列データ型

特定の配列要素を操作する

配列の長さを選択する

配列要素を個々のイベントに変換する

参照

その他のリソース

レコードデータ型

レコードフィールドを個々のイベントに変換する