Stream Analytics のノーコード エディターを使用してフィルター処理と Azure Synapse SQL への取り込みを行う

  • [アーティクル]

この記事では、ノーコード エディターを使用して Stream Analytics ジョブを簡単に作成する方法について説明します。 Event Hubs から継続的に読み取り、受信データをフィルター処理してから、Synapse SQL テーブルに結果を継続的に書き込みます。

前提条件

  • Azure Event Hubs リソースはパブリックにアクセスできる必要があり、ファイアウォールの内側に置いたり、Azure Virtual Network でセキュリティ保護したりすることはできません。
  • Event Hubs のデータは、JSON、CSV、または Avro 形式でシリアル化される必要があります。

データをフィルター処理して取り込む Stream Analytics ジョブを開発する

リアルタイム データをフィルター処理して Synapse SQL のテーブルに取り込む Stream Analytics ジョブを開発するには、次の手順のようにします。

  1. Azure portal で、Azure Event Hubs インスタンスを見つけて選択します。

  2. [機能]>[データの処理] を選択し、[Filter and ingest to Synapse SQL](フィルター処理と Synapse SQL への取り込み) カードで [開始] を選択します。
    Event Hubs のデータの処理でカードの開始を示すスクリーンショット。

  3. Stream Analytics ジョブを識別するための名前を入力して、[作成] を選択します。
    ジョブ名を入力する [New Stream Analytics job] (新しい Stream Analytics ジョブ) ウィンドウを示すスクリーンショット。

  4. [Event Hubs] ウィンドウでデータの種類として [シリアル化] を指定し、ジョブが Event Hubs に接続するのに使用する [認証方法] を指定します。 次に、 [接続](Connect) を選択します。
    Event Hubs 接続構成を示すスクリーンショット。

  5. 接続が正常に確立されて、Event Hubs のインスタンスにデータ ストリームが流れるようになると、次の 2 つのことがすぐに表示されます。

    • 入力データに存在するフィールド。 [フィールドの追加] を選択するか、フィールドの横にある 3 つのドット記号を選択して削除、名前の変更、または型の変更を行うことができます。
      フィールドの種類を削除、名前変更、または変更できる Event Hubs のフィールドの一覧を示すスクリーンショット。
    • ダイアグラム ビューの [データ プレビュー] テーブルでの受信データのライブ サンプル。 それは定期的に自動更新されます。 [ストリーミング プレビューの一時停止] を選択すると、サンプル入力データの静的ビューを見ることができます。
      [Data Preview] (データのプレビュー) の下にサンプル データが示されているスクリーンショット。

  6. [フィルター] 領域で、入力データを条件でフィルター処理するフィールドを選択します。
    条件を使用して受信データをフィルター処理できる [フィルター] 領域を示すスクリーンショット。

  7. フィルター処理されたデータを送信する Synapse SQL テーブルを選択します。

    1. ドロップダウン メニューから、サブスクリプション、データベース (専用SQL プール名)、認証方法を選択します。
    2. フィルター処理されたデータが取り込まれるテーブル名を入力します。 [接続] を選択します。
      Synapse SQL テーブルの接続の詳細を示すスクリーンショット。

    注意

    テーブルのスキーマが、データ プレビューによって生成されるフィールドの数およびその型と、正確に一致している必要があります。

  8. 必要に応じて、[静的プレビューの取得]/[静的プレビューの更新] を選択して、選択した Synapse SQL テーブルに取り込まれるデータ プレビューを表示します。
    [静的プレビューの取得] と [静的プレビューの更新] のオプションを示すスクリーンショット。

  9. [保存] を選択してから、Stream Analytics ジョブの [開始] を選択します。
    [保存] と [開始] のオプションを示すスクリーンショット。

  10. ジョブを開始するには、次のものを指定します。

    • ジョブを実行するストリーミング ユニット (SU) の数。 SU は、ジョブに割り当てられるコンピューティングとメモリの量を表します。 3 から始めて、必要に応じて調整することをお勧めします。
    • [出力データのエラー処理] – データ エラーが原因でジョブの送信先への出力が失敗した場合に必要な動作を指定できます。 既定では、ジョブは書き込み操作が成功するまで再試行します。 出力イベントを削除することもできます。
      [Stream Analytics ジョブの開始] のオプションを示すスクリーンショット。このオプションでは、出力時刻を変更し、ストリーミング ユニットの数を設定し、[出力データのエラー処理] のオプションを選択できます。

  11. [開始] を選ぶと、2 分以内にジョブの実行が開始され、下のタブ セクションにメトリックが表示されます。

    [Stream Analytics ジョブ] タブの [データの処理] セクションでジョブを確認することもできます。必要に応じて、[メトリックを開く] を選んで監視するか、停止して再起動します。

    実行中のジョブの状態が表示されている [Stream Analytics ジョブ] タブのスクリーンショット。

Event Hubs の geo レプリケーション機能を使用する場合の考慮事項

Azure Event Hubs では最近、geo レプリケーション機能のプレビューがローンチされました。 この機能は、Azure Event Hubs の Geo ディザスター リカバリー 機能とは異なります。

フェールオーバーの種類が [適用]、レプリケーションの整合性が [非同期]である場合、Stream Analytics ジョブでは、Azure Event Hubs 出力への出力が 1 回のみ行われることは保証されません。

Event Hubs を出力とするプロデューサーである Azure Stream Analytics が、フェールオーバー期間中および Event Hubs によるスロットリング中に、プライマリとセカンダリの間のレプリケーションのラグが最大構成ラグに達すると、ジョブでウォーターマーク遅延を検出する場合があります。

Event Hubs を入力とするコンシューマーである Azure Stream Analytics が、フェールオーバー期間中にウォーターマーク遅延を検出し、フェールオーバーの完了後、データをスキップするか、重複データを見つけようとする場合があります。

これらの注意事項により、Event Hubs のフェールオーバーが完了した直後に、適切な開始時刻で Stream Analytics ジョブを再起動することをお勧めします。 また、Event Hubs の geo レプリケーション機能はパブリック プレビュー段階であるため、現時点で運用環境の Stream Analytics ジョブにこのパターンを使用することはお勧めしません。 現在の Stream Analytics の動作は、Event Hubs の geo レプリケーション機能が一般公開される前に改善され、Stream Analytics の運用ジョブで使用できるようになります。

次のステップ

Azure Stream Analytics の詳細と、作成したジョブを監視する方法を理解してください。