Azure Cosmos DB の Azure Stream Analytics の出力は、ストリーム処理結果を JSON ドキュメントとして Azure Cosmos DB コンテナに書き込みます。 非構造化JSONデータに対するデータアーカイブや低遅延クエリをサポートしています。 この出力がどのように振る舞うかを理解することで、シナリオに必要なスループット、一貫性、パーティション設定を整えます。
出力ターゲットとしての Azure Cosmos DB の基礎
Stream AnalyticsのAzure Cosmos DB出力は、ストリーム処理結果をJSON出力としてAzure Cosmos DBコンテナに書き込みます。 Azure Cosmos DB に馴染みがない場合は、「Azure Cosmos DB のドキュメント」を参照して、作業を開始してください。
Stream AnalyticsはSQL APIを通じてのみAzure Cosmos DBに接続します。 他のAzure Cosmos DB APIはまだサポートされていません。 Stream Analytics を、その他の API で作成した Azure Cosmos DB アカウントへ接続する場合は、データが正しく格納されない可能性があります。 Azure Cosmos DBを出力として使う場合は、ジョブを互換性レベル1.2に設定してください。
Stream Analytics によって、ご利用のデータベース内にコンテナーは作成されません。 代わりに、ユーザーはそれらを事前に作成するように求められます。 これにより、Azure Cosmos DB コンテナーの課金コストを制御できるようになります。 また、Azure Cosmos DB API を使用して、ご利用のコンテナーのパフォーマンス、整合性、および容量を直接調整することもできます。 以下のセクションでは、Azure Cosmos DB 用のコンテナー オプションのいくつかを詳しく説明します。
整合性、可用性、および待機時間の調整
アプリケーションの要件に合わせて、Azure Cosmos DBでデータベースとコンテナを微調整し、一貫性、可用性、遅延、スループットのトレードオフを行ってください。
読み取り・書き込み遅延に対して必要とする読み取り一貫性レベルに応じて、データベースアカウントで一貫性レベルを選択してください。 スループットを向上させるために、コンテナ上でリクエストユニット(RU)を拡大しましょう。 また Azure Cosmos DB では、コンテナーへの各 CRUD 操作に対する同期インデックス作成も、既定で有効になっています。 このオプションはAzure Cosmos DBでの読み書き性能を制御するもう一つの便利な方法です。 詳細については、データベースとクエリの一貫性レベルの変更に関する記事をご覧ください。
Stream Analytics からのアップサート
Azure Cosmos DBとのStream Analytics連携を用いることで、特定のドキュメントID列に基づいてコンテナ内のレコードを挿入または更新できます。 この操作は、upsert とも呼ばれます。 Stream Analytics ではオプティミスティック upsert 手法が使用されます。 ドキュメント ID の競合により挿入が失敗した場合のみ、更新が行われます。
互換性レベル1.0を使用することで、Stream Analyticsはこの更新をPATCH操作として実行し、文書の部分的な更新をサポートします。 Stream Analytics では、新しいプロパティの追加、または既存のプロパティの置き換えが段階的に行われます。 ただし、JSON ドキュメント内の配列プロパティの値を変更すると、配列全体が上書きされます。 つまり、配列はマージされません。
互換性レベル1.2を使用すると、アップサートの動作が文書の挿入または置き換えに変更されます。 互換性レベル 1.2 に関する以降のセクションでは、この動作について詳しく説明します。
もし入ってくるJSON文書に既存のIDフィールドがある場合、Azure Cosmos DBは自動的にそのフィールドをDocument ID列として使用します。 Stream Analyticsはその後の書き込みをそのように処理するため、以下のいずれかの状況が生じます。
- 一意なIDの場合は挿入が行われる。
- ID が重複していて、ドキュメント ID が ID に設定されていると upsert となる。
- ID が重複していて、ドキュメント ID が設定されていないと、最初のドキュメントの後でエラーになる。
重複した ID を持つものも含め、"すべての" ドキュメントを保存する場合は、(AS キーワードを使用して) クエリ内の ID フィールドの名前を変更します。 Azure Cosmos DB により ID フィールドを作成するか、または ID を別の列の値に置き換えます (AS キーワードを使用するか、またはドキュメント ID 設定を使用)。
Azure Cosmos DB でのデータ パーティション分割
Azure Cosmos DB では、パーティションがご利用のワークロードに基づいて自動的にスケーリングされます。 データを分割するには 無制限 のコンテナを使ってください。 Stream Analytics では、無制限コンテナーに書き込む場合、以前のクエリ手順または入力のパーティション分割スキームと同数の並列ライターが使用されます。
注意
Azure Stream Analytics では最上位のパーティション キーを使用した無制限コンテナーのみがサポートされています。 たとえば、/region がサポートされています。 ネストされたパーティションキー(例えば /region/name)はサポートされていません。
選択したパーティション キーによっては、次の "警告" が出力される場合があります。
CosmosDB Output contains multiple rows and just one row per partition key. If the output latency is higher than expected, consider choosing a partition key that contains at least several hundred records per partition key.
多くの異なる値を持ち、作業負荷を均等に分配するパーティションキープロパティを選びましょう。 パーティショニングの自然な現象として、単一のパーティションの最大スループットは同じパーティションキーに関わるリクエストを制限します。
同一パーティション キー値に属するドキュメントのストレージ サイズは 20 GB に制限されます (物理的なパーティション サイズの制限は 50 GB です)。 理想的なパーティションキーとは、クエリ内で頻繁にフィルターとして現れ、十分なカージナリティを持ち、ソリューションのスケーラブルさを保証するものです。
Stream Analytics クエリと Azure Cosmos DB に使用されるパーティション キーは、同一である必要はありません。 完全並列トポロジーの場合は、Stream AnalyticsクエリのパーティションキーとしてInput PartitionキーPartitionIdを使用しますが、Azure Cosmos DBコンテナのパーティションキーとしては推奨されない場合があります。
また、パーティション キーは Azure Cosmos DB 用のストアド プロシージャやトリガーでのトランザクションの境界でもあります。 トランザクションで同時に発生する文書が同じパーティションキー値を共有するようにパーティションキーを選択してください。 パーティション キーの選択については、記事「Azure Cosmos DB でのパーティション分割」に詳しく説明されています。
固定されたAzure Cosmos DBコンテナの場合、Stream Analyticsは満杯後にスケールアップやスケールアウトを行う手段を提供しません。 それらの上限は 10 GB と 10,000 RU/秒のスループットです。 固定コンテナーから無制限コンテナー (1,000 RU/秒以上のスループットとパーティション キーを備えたコンテナーなど) にデータを移行するには、データ移行ツールまたは変更フィード ライブラリを使用します。
複数の固定コンテナーに書き込む機能は非推奨とされています。 Stream Analyticsの仕事を拡大するために使わないでください。
互換性レベル 1.2 でのスループットの向上
互換性レベル1.2を使用することで、Stream AnalyticsはAzure Cosmos DBへの一括書き込みをネイティブに統合します。 この統合を用いることで、Stream AnalyticsはAzure Cosmos DBに効果的に書き込みを行い、スループットを最大化し、スロットリングリクエストを効率的に処理します。
この強化された書き込みメカニズムは、upsert の動作が異なるため、新しい互換性レベルで利用できます 1.2以前のレベルを使うことで、アップサートの動作は文書の挿入やマージを行うことになります。 1.2を使うことで、アップサートの動作が文書の挿入や置き換えに変更されます。
1.2以前のレベルを使うことで、Stream Analyticsはカスタムストアドプロシージャを使ってパーティションキーごとに文書をAzure Cosmos DBにまとめてアップセルします。 そこでStream Analyticsはトランザクションとしてバッチを書き込みます。 たとえ1つのレコードが一時的なエラー(スロットリング)に遭っても、Stream Analyticsはバッチ全体を再試行しなければなりません。 この挙動は、合理的なスロットリングが行われるシナリオでも遅くなります。
次の例は、同じ Azure Event Hubs 入力から読み取られるまったく同じ 2 つの Stream Analytics ジョブを示しています。 どちらの Stream Analytics ジョブもパススルー クエリにより完全にパーティション分割されており、同じ Azure Cosmos DB コンテナーに書き込みます。 左側のメトリックは、互換性レベル 1.0 で構成されたジョブからのものです。 右側のメトリクスは1.2で設定されたジョブからのものです。 Azure Cosmos DB コンテナーのパーティション キーは、入力イベントから取得される一意の GUID です。
Event Hubs の受信イベント レートは、Azure Cosmos DB コンテナー (20,000 RU) が取り込めるように構成されている量の 2 倍であるため、Azure Cosmos DB ではスロットリングが発生すると考えられます。 しかし、1.2 のジョブは、一貫してより高いスループット (1 分あたりの出力イベント) とより低い平均 SU 使用率 (%) で書き込みを行います。 あなたの環境では、この違いはさらにいくつかの要因に依存します。 これらの要因には、選択しているイベント形式、入力イベントまたはメッセージのサイズ、パーティション キー、クエリが含まれます。
1.2を使うことで、Stream AnalyticsはAzure Cosmos DBの利用可能なスループットの100%をより賢く活用し、スロットリングやレート制限による再送信がほとんどありません。 この動作により、コンテナーで同時に実行されるクエリなどの他のワークロードにも優れたエクスペリエンスが提供されます。 1 秒あたり 1,000 から 10,000 のメッセージに対応するシンクとして Azure Cosmos DB を使用する場合、Stream Analytics でどのようにスケールアウトが行われるのかを確認したい場合は、この Azure サンプル プロジェクトを試してください。
Azure Cosmos DBの出力スループットは1.0と1.1で同一です。 Azure Cosmos DB を使用する Stream Analytics では、互換性レベル 1.2 を使用することを "強くお勧めします"。
JSON 出力の Azure Cosmos DB 設定
Stream AnalyticsでAzure Cosmos DBを出力として設定すると、以下のプロパティが出力を定義します。
| フィールド | 説明 |
|---|---|
| 出力エイリアス | ご利用の Stream Analytics クエリ内でこの出力を参照するエイリアス。 |
| サブスクリプション | Azure サブスクリプション。 |
| アカウント識別子 | Azure Cosmos DB アカウントの名前またはエンドポイント URI。 |
| アカウント キー | Azure Cosmos DB アカウントの共有アクセス キー。 |
| データベース | Azure Cosmos DB データベース名。 |
| コンテナー名 |
MyContainer などのコンテナー名。
MyContainer という名前のコンテナーが 1 つ存在する必要があります。 |
| ドキュメント ID | 省略可能。 出力イベントのカラム名は、挿入または更新操作の一意キーとして機能します。 空にすると、Stream Analyticsはすべてのイベントを更新オプションなしで挿入します。 |
Azure Cosmos DB 出力を構成したら、それをクエリ内で INTO ステートメントのターゲットとして使用できます。 Azure Cosmos DBの出力をその方法で使う場合は、パーティションキーを明示的に設定する必要があります。
出力レコードには、Azure Cosmos DB のパーティション キーの後に名前が付けられた大文字と小文字が区別される列が含まれている必要があります。 より多くの並列処理を実現するには、同じ列を使用する PARTITION BY 句がステートメントで必要になることがあります。
クエリの例を次に示します。
SELECT TollBoothId, PartitionId
INTO CosmosDBOutput
FROM Input1 PARTITION BY PartitionId
エラー処理と再試行
Azure Cosmos DB へのイベントの送信中に一時的な障害、サービスの利用不可、またはスロットリングが発生した場合、Stream Analytics では、操作を正常に終了するために無期限に再試行が行われます。 しかし、Unauthorized(HTTPエラーコード401)、NotFound(HTTPエラーコード404)、Forbidden(HTTPエラーコード403)、BadRequest(HTTPエラーコード400)の失敗に対しては再試行は試みません。
Azure Cosmos DBの出力が失敗する一般的な問題
いくつかの条件でAzure Cosmos DBの出力が失敗することがあります。 Stream Analyticsの出力データがコンテナの一意なインデックス制約に違反したり、 PartitionKey 列が存在しなかったり、 Id 列が存在しない場合があります。 一意インデックス制約の詳細については、Azure Cosmos DBの「一意キー制約」を参照してください。