Azure Stream Analytics での時間の処理について
この記事では、Azure Stream Analytics ジョブで時間の処理の現実的な問題を解決するために、設計上の選択を行う方法について説明します。 時間の処理の設計上の決定は、イベント順序要素に密接に関連しています。
バック グラウンド時間の概念
説明をより良く構成するため、背景の概念のいくつか定義しましょう。
イベント時間:元のイベントが発生した時刻です。 たとえば、高速道路を移動している車が料金所に近づくときです。
処理時間:イベントが処理システムに到達し、観測された時刻です。 たとえば、料金所のセンサーが車を検知して、コンピューター システムがデータを処理するためにかかるわずかな時間です。
基準値:イベントがどのポイントまでストリーミング プロセッサで受信されているかを示すイベント時間マーカーです。 基準値により、システムはイベント取り込みの明確な進行状況を示すことができます。 ストリームの性質上、受信イベント データが停止することはないため、基準値で示されるのは、ストリーム内の特定の時点までの進行状況です。
基準値の概念は重要です。 基準値により、Stream Analytics は、取り消す必要がない完全かつ正確で繰り返し可能な結果をシステムが生成できるタイミングを判断することができます。 この処理は、予測可能で繰り返し可能な方法で実行できます。 たとえば、なんらかのエラー処理条件のために再計算を実行する必要がある場合、基準値は安全な開始点と終了点となります。
このテーマに関する追加リソースについては、Tyler Akidau 氏のブログ記事「Streaming 101」 (ストリーミング 101) と「Streaming 102」 (ストリーミング 102) を参照してください。
最適な開始時刻の選択
Stream Analytics では、イベント時間を選択するための 2 つの選択肢 (到着時間とアプリケーション時間) をユーザーに提供します。
到着時間
イベントがソースに到達すると、入力ソースに到着時刻が割り当てられます。 到着時間にアクセスするには、Event Hubs 入力の場合は EventEnqueuedUtcTime プロパティを、IoT Hub 入力の場合は IoTHub.EnqueuedTime プロパティを、BLOB 入力の場合は BlobProperties.LastModified プロパティを使用します。
既定では到着時間が使用され、テンポラル ロジックが不要なデータ アーカイブのシナリオに最適です。
アプリケーション時間 (イベント時間とも呼ばれます)
アプリケーション時間は、イベントの生成時に割り当てられ、イベント ペイロードの一部になります。 アプリケーション時間でイベントを処理するには、SELECT クエリで Timestamp by 句を使用します。 Timestamp by がない場合は、イベントは到着時間で処理されます。
ソース システムまたはネットワークの遅延を考慮するためにテンポラル ロジックが関与する場合は、ペイロードでタイムスタンプを使用することが重要です。 イベントに割り当てられた時間は SYSTEM.TIMESTAMP で使用できます。
Azure Stream Analytics での時間の進行
アプリケーション時間を使用すると、時間の進行は受信イベントに基づきます。 ストリーム処理システムでは、イベントがないかどうか、またはイベントが遅延しているかどうかを把握するのは困難です。 このため、Azure Stream Analytics では、各入力パーティションに対し、次の方法でヒューリスティック基準値が生成されます。
受信イベントがある場合は、基準値 = Stream Analytics でこれまでに確認された最長のイベント時間 - 誤順序の許容値ウィンドウ サイズです。
受信イベントがない場合は、基準値 = 現在の推定到着時間 - 到着遅延許容値ウィンドウです。 推定到着時間は、最後に入力イベントが確認された時点からの経過時間 + その入力イベントの到着時間です。
実際の到着時間は、イベントを処理している Azure Stream Analytics VM ではなく、Event Hubs などの入力イベント ブローカーで生成されるため、到着時間は見積もることしかできません。
設計には、基準値の生成のほかに、次の 2 つの目的があります。
受信イベントの有無に関係なく、システムが適切なタイミングで結果を生成する。
出力結果を表示するタイミングを制御できます。 Azure portal で、Stream Analytics ジョブの [イベント順序] ページで、 [順不同のイベント] 設定を構成できます。 その設定を構成するときに、イベント ストリーム内で順不同のイベントの許容値による適時性のトレードオフを検討してください。
受信イベントがない場合でも基準値を生成し続けるには、到着遅延許容値ウィンドウが必要です。 ときには、イベント入力ストリームがスパースの場合など、受信イベントが到着しない期間がある場合があります。 入力イベント ブローカーで複数のパーティションを使用することで、その問題はさらに悪化します。
入力がスパースで、複数のパーティションが使用されている場合、到着遅延許容値ウィンドウがないストリーミング データ処理システムは、出力の遅延に悩まされる可能性があります。
このシステム動作は、反復可能である必要があります。 反復可能性は、ストリーミング データ処理システムの重要なプロパティです。
基準値は、到着時間とアプリケーション時間から派生します。 どちらもイベント ブローカーで永続化されるため、反復可能です。 イベントなしで到着時間を見積もる場合、Azure Stream Analytics では、障害復旧での再生時の反復可能性のために、推定到着時間が記録されます。
イベント時間として到着時間の使用を選択した場合、順不同の許容値と到着遅延の許容値を構成する必要はありません。 到着時間は入力イベント ブローカーで増加することが保証されているため、Azure Stream Analytics では単に構成が無視されます。
到着遅延イベント
到着遅延許容期間の定義により、Azure Stream Analytics では、受信イベントごとにイベント時間が到着時間と比較されます。 イベント時間が許容期間外の場合は、イベントを削除するようにシステムを構成するか、イベントの時間を許容期間内になるように調整することができます。
基準値が生成されると、サービスは基準値よりも低いイベント時間でイベントを受信する可能性があります。 これらのイベントを破棄するか、イベントの時間を基準値に調整するようにサービスを構成することができます。
調整の一環として、イベントの System.Timestamp が新しい値に設定されますが、イベント時間フィールド自体は変更されません。 この調整は、イベントの System.Timestamp がイベント時間フィールドの値と異なる可能性があり、予期しない結果が生成される可能性がある場合にのみ行います。
サブストリームによる時間変化の処理
ここで説明されているヒューリスティック基準値の生成メカニズムは、さまざまなイベントの送信元間で時間がほぼ同期されているほとんどのケースで問題なく機能します。 ただし、実際には、特に多くの IoT シナリオでは、システムでイベント送信元の時計を制御することはほとんどできません。 イベント送信元は、フィールド内のあらゆるデバイスが考えられ、ハードウェアおよびソフトウェアのバージョンも異なる可能性があります。
入力パーティション内のすべてのイベントに基準値をグローバルに使用する代わりに、Stream Analytics にはサブストリームと呼ばれる別のメカニズムがあります。 TIMESTAMP BY 句とキーワード OVER を使用するジョブ クエリを記述することで、ジョブでサブストリームを利用することができます。 サブストリームを指定するには、キーワード OVER の後にキー列の名前 (deviceid など) を指定して、システムがその列で時間ポリシーを適用するようにします。 各サブストリームは、独自の独立した基準値を取得します。 このメカニズムは、イベントの送信元の間での大幅な時刻のずれやネットワークの遅延に対処するときに、タイムリーな出力の生成を可能にするのに便利です。
サブストリームは、Azure Stream Analytics によって提供される独自のソリューションで、他のストリーミング データ処理システムでは提供されていません。
サブストリームを使用する場合、Stream Analytics では、到着遅延許容値ウィンドウを受信イベントに適用します。 到着遅延許容期間によって、異なるサブストリームを相互に分離できる最大量が決定されます。 たとえば、デバイス 1 がタイムスタンプ 1 に、デバイス 2 がタイムスタンプ 2 にある場合、到着遅延許容期間の最大値はタイムスタンプ 2 からタイムスタンプ 1 を引いた値になります。 既定の設定 (5 秒) は、異なるタイムスタンプを持つデバイスには小さすぎる場合があります。 5 分から開始して、デバイスの時刻のずれのパターンに従って調整することをお勧めします。
早期到着イベント
到着遅延許容値ウィンドウの反対のような、早期到着ウィンドウと呼ばれるもう 1 つの概念にお気づきかもしれません。 このウィンドウは 5 分に固定されており、到着遅延許容値ウィンドウとは異なる目的を果たします。
Azure Stream Analytics は完全な結果を保証しているため、ジョブの最初の出力時刻 (入力時刻ではなく) として、ジョブの開始時刻を指定するだけです。 ジョブの開始時刻は、完全なウィンドウ (ウィンドウの途中からだけでなく) が処理されるようにするために必要です。
Stream Analytics は、クエリ仕様から開始時刻を派生します。 ただし、入力イベント ブローカーは到着時間でインデックス作成されるだけなので、システムでイベントの開始時刻を到着時刻に変換する必要があります。 システムは、入力イベント ブローカーでその時点からイベントの処理を開始できます。 早期到着ウィンドウの制限を使用すると、イベントの開始時刻から 5 分の早期到着ウィンドウを引くことで、簡単に変換できます。 この計算は、イベント時間が到着時間より 5 分早いことが確認されたすべてのイベントがシステムによってドロップされることも意味します。 イベントがドロップされると、早期入力イベント メトリックがインクリメントされます。
この概念は、出力をどこから開始するかに関わらず、処理が繰り返し可能になることを保証するために使用されます。 このようなメカニズムがなければ、他の多くのストリーミング システムが要求しているように、反復可能性を保証することはできません。
イベント順序時間の許容値の副作用
Stream Analytics ジョブには、複数のイベント順序オプションがあります。 順不同のイベント設定 (誤順序の許容値) と到着が遅れるイベント設定 (到着遅延許容値) の 2 つは、Azure portal で構成できます。 早期到着許容値は固定されており、調整することはできません。 これらの時間ポリシーは、強力な保証を提供するために Stream Analytics によって使用されます。 ただし、これらの設定には、次のようないくつかの予期しない影響を及ぼす場合があります。
早すぎるイベントが誤って送信される。
早期イベントは、通常に出力されるべきではありません。 しかし、送信元の時計が早すぎる場合、早期イベントが出力に送信される可能性があります。 すべての早期到着イベントはドロップされるため、出力にはこれらは一切表示されません。
Azure Stream Analytics によって処理される Event Hubs に古いイベントが送信される。
古いイベントは最初は無害に見えるかもしれませんが、到着遅延許容値の適用により、古いイベントはドロップされる可能性があります。 イベントが古すぎる場合、イベント取り込み中に System.Timestamp 値が変更されます。 この動作により、現在 Azure Stream Analytics は、履歴イベント処理のシナリオよりも、ほぼリアルタイムのイベント処理シナリオにより適しています。 一部のケースでは、到着が遅れるイベントの時間を可能な最大値 (20 日) に設定して、この動作を回避することができます。
出力が遅延しているように見える。
最初の基準値は、次の算出時間で生成されます。システムがこれまでに監視してきたイベントの最長時間 - 誤順序の許容値ウィンドウ サイズ。 既定では、誤順序の許容値はゼロ (00 分 00 秒) に構成されます。 この値をより高い値 (ゼロ以外の時間値) に設定すると、計算される最初の基準値の時間により、ストリーミング ジョブの最初の出力がその値の時間分 (またはそれ以上) 遅延します。
入力がスパース。
特定のパーティション内に入力がない場合、基準値時間 = 到着時間 - 到着遅延許容値ウィンドウとなります。 その結果、入力イベントがスパースで頻度が低い場合、出力がその時間分遅延する可能性があります。 既定の到着が遅れるイベント値は 5 秒です。 たとえば、1 つずつ入力イベントを送信すると、多少の遅延が見られます。 到着が遅れるイベント ウィンドウを大きい値に設定すると、遅延が悪化する可能性があります。
System.Timestamp 値が、イベント時間フィールドの時間と異なる。
前述のように、システムでは、誤順序の許容値または到着遅延許容値ウィンドウでイベント時間が調整されます。 イベントの System.Timestamp 値は調整されますが、イベント時間フィールドは調整されません。 これは、タイムスタンプが調整されたイベントを識別するために使用できます。 許容値の 1 つによってタイムスタンプが変更された場合、通常はこれらは同じです。
監視するメトリック
Azure Stream Analytics ジョブのメトリックを通じて、イベント順序時間許容値の複数の影響を監視できます。 関係するメトリックを次に示します。
| メトリック | 説明 |
|---|---|
| 順不同のイベント | ドロップされた、または調整されたタイムスタンプが付与された、順不同で受信したイベントの数を示します。 このメトリックは、Azure portal でジョブの [イベント順序] ページで [順不同のイベント] 設定を構成することで直接の影響を受けます。 |
| 遅延入力イベント | ソースからの到着が遅延しているイベントの数を示します。 このメトリックには、ドロップされたか、タイムスタンプが調整されたイベントが含まれます。 このメトリックは、Azure portal でジョブの [イベント順序] ページで [到着が遅れるイベント] の設定を構成することで直接の影響を受けます。 |
| 初期入力イベント | ドロップされたか、または 5 分を超えて早く到着した場合に調整されているタイムスタンプを持つ、ソースから早期に到着したイベントの数を示します。 |
| 基準値の遅延 | ストリーミング データ処理ジョブの遅延を示します。 詳細については、次のセクションを参照してください。 |
基準値の遅延の詳細
基準値の遅延メトリックは、処理ノードの実時間 - これまでに観測された最大の基準値として計算されます。 詳細については、基準値の遅延のブログ記事を参照してください。
通常の操作でこのメトリックの値が 0 より大きくなるのには、いくつかの理由が考えられます。
ストリーミング パイプラインの固有の処理の遅延。 通常この遅延はわずかです。
誤順序の許容値ウィンドウのサイズの分だけ基準値が減るため、このウィンドウにより遅延が発生します。
到着遅延許容値ウィンドウのサイズの分だけ基準値が減るため、このウィンドウにより遅延が発生します。
メトリックを生成している処理ノードの時刻のずれ。
ストリーミング パイプラインを遅くする原因となる可能性があるリソースの制約は、他にもたくさんあります。 基準値の遅延メトリックは、次の原因で増加する場合があります。
入力イベントのボリュームを処理するために十分な処理リソースが Stream Analytics にない。 リソースをスケールアップするには、「ストリーミング ユニットの理解と調整」を参照してください。
入力イベント ブローカーが、スループット不足により調整されている。 考えられる解決策については、「Azure Event Hubs のスループット単位を自動的にスケールアップする」を参照してください。
出力シンクが十分な容量でプロビジョニングされていないため、調整されている。 考えられる解決策は、使用されている出力サービスの種類によって大きく異なります。
出力イベントの頻度
Azure Stream Analytics では、基準値の進行状況を唯一のトリガーとして使用して、出力イベントを生成します。 基準値は入力データから派生されるため、障害復旧時、またはユーザーが開始した再処理でも反復可能です。 ウィンドウ集計を使用すると、サービスではウィンドウの終了時にのみ出力が生成されます。 場合によっては、ウィンドウから生成される部分的な集計を確認したい場合があります。 部分的な集計は、Azure Stream Analytics では現在サポートされていません。
他のストリーミング ソリューションでは、外部の状況に応じて、さまざまなトリガー ポイントで出力イベントを具体化できる場合があります。 ソリューションの中には、指定された時間枠で出力イベントを複数回生成できるものもあります。 入力値が調整されると、集計結果はより正確になります。 イベントは、最初に推測して、徐々に変更することができます。 たとえば、特定のデバイスがネットワークからオフラインになると、システムでは推定値を使用することができます。 後で、同じデバイスがネットワークでオンラインになったら、 実際のイベント データを入力ストリームに含めることができます。 その時間枠の処理からの出力結果は、より正確な出力になります。
図を使った基準値の例の説明
以下の図は、さまざまな状況での基準値の進行状況を示しています。
次の表は、以下でグラフ化されているサンプル データを示しています。 イベント時間と到着時間が異なっていることに注目してください。一致しているものもあれば、一致していないものもあります。
| イベント時間 | 到着時間 | deviceId |
|---|---|---|
| 12:07 | 12:07 | device1 |
| 12:08 | 12:08 | device2 |
| 12:17 | 12:11 | device1 |
| 12:08 | 12:13 | device3 |
| 12:19 | 12:16 | device1 |
| 12:12 | 12:17 | device3 |
| 12:17 | 12:18 | device2 |
| 12:20 | 12:19 | device2 |
| 12:16 | 12:21 | device3 |
| 12:23 | 12:22 | device2 |
| 12:22 | 12:24 | device2 |
| 12:21 | 12:27 | device3 |
この図では、次の許容値が使用されています。
- 早期到着ウィンドウは 5 分
- 到着遅延ウィンドウは 5 分
- 並べ替えウィンドウは 2 分
これらのイベントを進行する基準値の図:
上の図で示されている注目すべきプロセス:
最初のイベント (device1) と 2 番目のイベント (device2) の時間は一致していて、調整なしで処理されています。 各イベントで基準値が進行します。
3 番目のイベント (device1) が処理されるとき、到着時間 (12:11) がイベント時間 (12:17) より先行しています。 イベントは 6 分早く到着したため、5 分間の早期到着許容値によりこのイベントはドロップされます。
早期イベントのこのケースでは、基準値は進行しません。
4 番目のイベント (device3) と 5 番目のイベント (device1) の時間は一致していて、調整なしで処理されます。 各イベントで基準値が進行します。
6 番目のイベント (device3) が処理されるとき、到着時間 (12:17) とイベント時間 (12:12) は基準値レベルを下回っています。 イベント時間が基準値レベル (12:17) に調整されます。
12 番目のイベント (device3) が処理されるとき、到着時間 (12:27) はイベント時間 (12:21) よりも 6 分早くなっています。 到着遅延ポリシーが適用されます。 イベント時間が基準値 (12:21) を上回る (12:22) に調整されるため、さらなる調整は適用されません。
早期到着ポリシーなしで進行する基準値の 2 番目の図:
この例では、早期到着ポリシーは適用されていません。 早期に到着する外れ値イベントにより、基準値が大幅に増加します。 このシナリオでは、3 番目のイベント (時間 12:11 の deviceId1) がドロップされず、基準値が 12:15 に上昇していることに注目してください。 4 番目のイベント時間は、結果として 7 分進めて調整されています (12:08 から 12:15)。
最後の図では、サブストリームが使用されています (DeviceId 経由で)。 複数の基準値が追跡されます (ストリームごとに 1 つ)。 結果として、時間が調整されたイベントが少なくなります。
