Azure SQL Edge でデータストリーミングジョブを作成する

[アーティクル]
09/27/2022

この記事では、Azure SQL Edge で T-SQL ストリーミングジョブを作成する方法を説明します。外部ストリームの入力オブジェクトと出力オブジェクトを作成してから、ストリーミングジョブの作成の一部としてストリーミングジョブクエリを定義します。

外部ストリームの入力オブジェクトと出力オブジェクトを構成する

T-SQL ストリーミングでは、SQL Server の外部データソース機能を使用して、ストリーミングジョブの外部ストリームの入力と出力に関連付けるデータソースを定義します。外部ストリームの入力オブジェクトまたは出力オブジェクトを作成するには、次の T-SQL コマンドを使用します。

さらに、出力ストリームとして Azure SQL Edge、SQL Server、または Azure SQL Database が使用されている場合は、CREATE DATABASE SCOPED CREDENTIAL (Transact-SQL) が必要です。この T-SQL コマンドでは、データベースにアクセスするための資格情報を定義します。

サポートされている入力ストリームと出力ストリームのデータソース

現在、Azure SQL Edge でストリーム入力および出力としてサポートされているのは、次のデータソースのみです。

[データソースの種類]	入力	出力	説明
Azure IoT Edge ハブ	Y	Y	Azure IoT Edge ハブに対するストリーミングデータの読み書きを行うためのデータソース。詳細については、IoT Edge ハブに関するページを参照してください。
SQL Database	N	Y	SQL Database にストリーミングデータを書き込むためのデータソース接続。データベースは、Azure SQL Edge のローカルデータベースでも、SQL Server または Azure SQL Database のリモートデータベースでもかまいません。
Kafka	Y	N	Kafka トピックからストリーミングデータを読み取るためのデータソース。現在、このアダプターは、Azure SQL Edge の Intel または AMD バージョンでのみ使用できます。 ARM64 バージョンの Azure SQL Edge では使用できません。

例:Azure IoT Edge ハブ用の外部ストリームの入力オブジェクトと出力オブジェクトを作成する

次の例では、Azure IoT Edge ハブ用の外部ストリームオブジェクトを作成します。 Azure IoT Edge ハブ用の外部ストリームの入力および出力データソースを作成するには、最初に読み書きされるデータのレイアウトに対する外部ファイル形式を作成する必要もあります。

JSON 型の外部ファイル形式を作成します。

Create External file format InputFileFormat
WITH 
(  
   format_type = JSON,
)
go

Azure IoT Edge ハブに対する外部データソースを作成します。次の T-SQL スクリプトでは、Azure SQL Edge と同じ Docker ホスト上で実行される IoT Edge ハブへのデータソース接続が作成されます。
```
CREATE EXTERNAL DATA SOURCE EdgeHubInput 
WITH 
(
    LOCATION = 'edgehub://'
)
go
```
Azure IoT Edge ハブに対する外部ストリームオブジェクトを作成します。次の T-SQL スクリプトでは、IoT Edge ハブに対するストリームオブジェクトが作成されます。 IoT Edge ハブのストリームオブジェクトの場合、LOCATION パラメーターは、読み取り元または書き込み先である IoT Edge ハブのトピックおよびチャネルの名前です。
```
CREATE EXTERNAL STREAM MyTempSensors 
WITH 
(
    DATA_SOURCE = EdgeHubInput,
    FILE_FORMAT = InputFileFormat,
    LOCATION = N'TemperatureSensors',
    INPUT_OPTIONS = N'',
    OUTPUT_OPTIONS = N''
);
go
```

例:Azure SQL Database への外部ストリームオブジェクトを作成する

次の例では、Azure SQL Edge のローカルデータベースに対する外部ストリームオブジェクトを作成します。

データベースにマスターキーを作成します。これは、資格情報のシークレットの暗号化に必須です。
```
CREATE MASTER KEY ENCRYPTION BY PASSWORD = '<<Strong_Password_For_Master_Key_Encryption>>';
```
SQL Server ソースにアクセスするために、データベーススコープの資格情報を作成します。次の例では、IDENTITY = username と SECRET = password を使用して、外部データソースに対する資格情報を作成します。
```
CREATE DATABASE SCOPED CREDENTIAL SQLCredential
WITH IDENTITY = '<SQL_Login>', SECRET = '<SQL_Login_PASSWORD>'
go
```
CREATE EXTERNAL DATA SOURCE を使用して、外部データソースを作成します。次に例を示します。
- LocalSQLOutput という名前の外部データソースを作成します。
- 外部のデータソースを識別します (LOCATION = '<vendor>://<server>[:<port>]')。この例では、Azure SQL Edge のローカルインスタンスを指しています。
- 前に作成した資格情報を使用します。
```
CREATE EXTERNAL DATA SOURCE LocalSQLOutput 
WITH 
(
    LOCATION = 'sqlserver://tcp:.,1433',
    CREDENTIAL = SQLCredential
)
go
```
外部ストリームオブジェクトを作成します。次の例では、データベース MySQLDatabase の dbo.TemperatureMeasurements テーブルを指す外部ストリームオブジェクトを作成します。
```
CREATE EXTERNAL STREAM TemperatureMeasurements 
WITH 
(
    DATA_SOURCE = LocalSQLOutput,
    LOCATION = N'MySQLDatabase.dbo.TemperatureMeasurements',
    INPUT_OPTIONS = N'',
    OUTPUT_OPTIONS = N''
);
```

例:Kafka 用の外部ストリームオブジェクトを作成する

次の例では、Azure SQL Edge のローカルデータベースに対する外部ストリームオブジェクトを作成します。この例では、Kafka サーバーが匿名アクセス用に構成されていることを前提としています。

CREATE EXTERNAL DATA SOURCE を使用して、外部データソースを作成します。次に例を示します。

Create EXTERNAL DATA SOURCE [KafkaInput] 
With
(
    LOCATION = N'kafka://<kafka_bootstrap_server_name_ip>:<port_number>'
)
GO

Kafka 入力用の外部ファイル形式を作成します。次の例では、Gzip 圧縮を使用して JSON ファイル形式を作成しました。
```
CREATE EXTERNAL FILE FORMAT JsonGzipped  
 WITH 
 (  
     FORMAT_TYPE = JSON , 
     DATA_COMPRESSION = 'org.apache.hadoop.io.compress.GzipCodec' 
 )
```

外部ストリームオブジェクトを作成します。次の例では、Kafka トピック *TemperatureMeasurement* を指す外部ストリームオブジェクトを作成します。

CREATE EXTERNAL STREAM TemperatureMeasurement 
WITH 
(  
    DATA_SOURCE = KafkaInput, 
    FILE_FORMAT = JsonGzipped,
    LOCATION = 'TemperatureMeasurement',
    INPUT_OPTIONS = 'PARTITIONS: 10' 
);

ストリーミングジョブとストリーミングクエリを作成する

sys.sp_create_streaming_job システムストアドプロシージャを使用して、ストリーミングクエリを定義し、ストリーミングジョブを作成します。 sp_create_streaming_job ストアドプロシージャは、次のパラメーターを受け取ります。

job_name:ストリーミングジョブの名前。ストリーミングジョブの名前は、インスタンス全体で一意です。
statement:Stream Analytics クエリ言語ベースのストリーミングクエリステートメント。

次の例では、1 つのストリーミングクエリを使用して簡単なストリーミングジョブを作成します。このクエリでは、IoT Edge ハブから入力を読み取り、データベースの dbo.TemperatureMeasurements に書き込みます。

EXEC sys.sp_create_streaming_job @name=N'StreamingJob1',
@statement= N'Select * INTO TemperatureMeasurements from MyEdgeHubInput'

次の例では、複数の異なるクエリを使用して、さらに複雑なストリーミングジョブを作成します。これらのクエリには、組み込みの AnomalyDetection_ChangePoint 関数を使用して温度データの異常を特定するものが含まれます。

EXEC sys.sp_create_streaming_job @name=N'StreamingJob2', @statement=
N' Select * INTO TemperatureMeasurements1 from MyEdgeHubInput1

Select * Into TemperatureMeasurements2 from MyEdgeHubInput2

Select * Into TemperatureMeasurements3 from MyEdgeHubInput3

SELECT
Timestamp as [Time],
[Temperature] As [Temperature],
GetRecordPropertyValue(AnomalyDetection_ChangePoint(Temperature, 80, 1200) OVER(LIMIT DURATION(minute, 20)), ''Score'') as ChangePointScore,
GetRecordPropertyValue(AnomalyDetection_ChangePoint(Temperature, 80, 1200) OVER(LIMIT DURATION(minute, 20)), ''IsAnomaly'') as IsChangePointAnomaly
INTO TemperatureAnomalies FROM MyEdgeHubInput2;
'
go

ストリーミングジョブを開始、停止、削除、監視する

Azure SQL Edge でストリーミングジョブを開始するには、sys.sp_start_streaming_job ストアドプロシージャを実行します。このストアドプロシージャでは、開始するストリーミングジョブの名前が入力として必要です。

exec sys.sp_start_streaming_job @name=N'StreamingJob1'
go

ストリーミングジョブを停止するには、sys.sp_stop_streaming_job ストアドプロシージャを実行します。このストアドプロシージャでは、入力として停止するストリーミングジョブの名前が必要です。

exec sys.sp_stop_streaming_job @name=N'StreamingJob1'
go

ストリーミングジョブをドロップ (削除) するには、sys.sp_drop_streaming_job ストアドプロシージャを実行します。このストアドプロシージャでは、削除するストリーミングジョブの名前が入力として必要です。

exec sys.sp_drop_streaming_job @name=N'StreamingJob1'
go

ストリーミングジョブの現在の状態を取得するには、sys.sp_get_streaming_job ストアドプロシージャを実行します。このストアドプロシージャでは、削除するストリーミングジョブの名前が入力として必要です。ストリーミングジョブの名前と現在の状態が出力されます。

exec sys.sp_get_streaming_job @name=N'StreamingJob1'
        WITH RESULT SETS
(
       (
       name nvarchar(256),
       status nvarchar(256),
       error nvarchar(256)
       )
)

ストリーミングジョブの状態は次のいずれかです。

Status	説明
作成済み	ストリーミングジョブは作成されましたが、まだ開始されていません。
開始中	ストリーミングジョブは開始フェーズになっています。
アイドル	ストリーミングジョブは実行中ですが、処理する入力がありません。
処理中	ストリーミングジョブは実行中であり、入力を処理しています。この状態は、ストリーミングジョブが正常な状態であることを示します。
低下しています	ストリーミングジョブは実行中ですが、入力処理の間に致命的ではないエラーが発生しました。入力ジョブは実行を続けますが、エラーが発生する入力は削除されます。
停止済み	ストリーミングジョブは停止されました。
失敗	ストリーミングジョブは失敗しました。これは、通常、処理中に致命的なエラーが発生したことを示します。

注意

ストリーミングジョブは非同期的に実行されるため、実行時にジョブでエラーが発生する場合があります。ストリーミングジョブの失敗をトラブルシューティングするには、sys.sp_get_streaming_job ストアドプロシージャを使用するか、Azure SQL Edge コンテナーの Docker ログを調べて、ストリーミングジョブのエラーの詳細を確認します。

Azure SQL Edge でデータストリーミングジョブを作成する

外部ストリームの入力オブジェクトと出力オブジェクトを構成する

サポートされている入力ストリームと出力ストリームのデータソース

例:Azure IoT Edge ハブ用の外部ストリームの入力オブジェクトと出力オブジェクトを作成する

例:Azure SQL Database への外部ストリームオブジェクトを作成する

例:Kafka 用の外部ストリームオブジェクトを作成する

ストリーミングジョブとストリーミングクエリを作成する

ストリーミングジョブを開始、停止、削除、監視する

次のステップ

その他のリソース

その他のリソース

Azure SQL Edge でデータ ストリーミング ジョブを作成する

外部ストリームの入力オブジェクトと出力オブジェクトを構成する

サポートされている入力ストリームと出力ストリームのデータ ソース

例:Azure IoT Edge ハブ用の外部ストリームの入力オブジェクトと出力オブジェクトを作成する

例:Azure SQL Database への外部ストリーム オブジェクトを作成する

例:Kafka 用の外部ストリーム オブジェクトを作成する

ストリーミング ジョブとストリーミング クエリを作成する

ストリーミング ジョブを開始、停止、削除、監視する

次のステップ

その他のリソース

その他のリソース

Azure SQL Edge でデータストリーミングジョブを作成する

サポートされている入力ストリームと出力ストリームのデータソース

例:Azure SQL Database への外部ストリームオブジェクトを作成する

例:Kafka 用の外部ストリームオブジェクトを作成する

ストリーミングジョブとストリーミングクエリを作成する

ストリーミングジョブを開始、停止、削除、監視する