Spark構造化ストリーミングを使ったノートブックから、Microsoft Fabricのイベントストリームをクエリ

Microsoft FabricのEventstreamは、リアルタイムのデータ処理と分析を可能にする完全管理型イベントインジェスションおよびストリーミングサービスです。 Spark Structured Streamingを使ってFabricノートブックとイベントストリームを統合し、ストリーミングデータをリアルタイムで処理・分析できます。

この統合により、Real-Time ハブを通じてイベントストリームやその他のリアルタイムソースをFabricノート内から探索できます。 また、新しいイベントストリームを作成し、CDC対応のデータベース、メッセージブローカー、ストリーミングサービス、パブリックフィードなど約30(または増加中)のストリーミングソースからデータを取り込むこともできます。 新しいプログラミングモデルにより、接続文字列や認証情報なしで、慣れたSpark APIを使ってイベントストリームへの接続、ストリーミングデータの読み取り、リアルタイム分析が容易に行えるため、リアルタイムアプリケーション構築の安全性が向上しました

[前提条件]

開始する前に、次の前提条件が満たされていることを確認してください。

  • イベントストリームが有効化されたFabricワークスペース。
  • Fabricで設定したノートブック環境。
  • Spark 構造化ストリーミングと PySpark または Scala に関する基本的な知識。

ノートブックから Eventstream にクエリを実行する手順

  1. Eventstream を検出して Notebook に追加します。

    1. Microsoft Fabric ノートブックを開きます。

    2. 「データ項目」タブの [データ項目を追加] を選択し、ドロップダウンメニューから [Real-Time ハブから] を選択します。

    3. Real-Time ハブ ダイアログで、さまざまな検索オプションとフィルター オプションを使用してクエリを実行するターゲット イベントストリームを見つけます。

    4. ノートブック環境に eventstream を追加します。

      Real-Time ハブで eventstream を検出する方法を示すスクリーンショット。

      eventstream 項目が追加されると、ノートブック オブジェクト エクスプローラーの Real-Time ハブ ノードの下に項目名と共に表示されます。ノードを展開すると、eventstream 項目内に既定のストリームまたは派生ストリームが表示されます。

      ノートブック オブジェクト エクスプローラーで追加された eventstream を示すスクリーンショット。

  2. eventstream からストリーミング データを読み取ります。

    1. オブジェクト エクスプローラーの eventstream 項目のコンテキスト メニューから、[ Spark で読み取り ] を選択し、選択したイベントストリームから読み取る構成済みのコード セルを作成します。

      コンテキスト メニューから Spark で eventstream を読み取る方法を示すスクリーンショット。

    2. 新しく作成されたコード セルでは、次の PySpark コード スニペットを使用して、eventstream からストリーミング データを読み取ることができます。

      from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StringType
from pyspark.sql.dataframe import DataFrame

eventstream_options = {
    "eventstream.itemid": __in_eventstream_item_id,
    "eventstream.datasourceid": __in_eventstream_datasource_id
}

# Read from Kafka using the config map
df_raw = spark.readStream.format("kafka").options(**eventstream_options).load()

decoded_df = df_raw.select(
    col("key").cast(StringType()).alias("key"),
    col("value").cast(StringType()).alias("value"),
    col("partition"),
    col("offset")
)

def showDf(x:DataFrame, y:int):
    x.show()

# Print messages to the console
query = decoded_df.writeStream.foreachBatch(showDf).outputMode("append").start()
query.awaitTermination()

      eventstream.itemidオプションとeventstream.datasourceidオプションの値は、このコード セルの上にあるパラメーター コード セルに自動的に設定されます。

      注

      Spark を使用した 読み取り時の選択ごとに、必要なパラメーターが自動的に入力された新しいコード セルが生成され、Eventstream 項目に対して新しい Notebook 変換先が作成されます。ここで、 eventstream.datasourceid はノートブックの宛先ごとに一意です。 1 つのノートブック項目については、複数の宛先を作成しないように、1 つの Eventstream 項目の宛先として 1 回設定することをお勧めします。

      ジョブを実行する前に、Eventstream 項目のコンテキスト メニューの [データのプレビュー] ボタンをクリックして、ストリーミング データをプレビュー することもできます。 サイド パネルが開き、ノートブックの横にストリーミング データのスナップショットが表示されます。

      コンテキスト メニューから Eventstream データをプレビューする方法を示すスクリーンショット。

  3. 宛先としてノートブックを追加する: ノートブックを Eventstream の宛先として追加して、ノートブック内から Eventstream へのデータの取り込みを開始することもできます。 詳細については、イベントストリームに Spark Notebook の送信先を追加するを参照してください。

Conclusion

Spark構造化ストリーミングを用いてノートブックとFabricイベントストリームを統合することで、リアルタイムデータをシームレスに処理・分析できます。 この記事で示された手順に従うことで、イベントストリームに接続し、ストリーミングデータを読み、馴染みのあるSpark APIを使ってリアルタイム分析を行うことができます。 この統合により、リアルタイムアプリケーションの構築や、Fabricエコシステム内のストリーミングデータからの洞察獲得に新たな可能性が開かれます。

関連コンテンツ

Fabricでイベントストリームやノートブックを扱う方法について詳しく知りたい方は、以下のリソースをご覧ください

  • 最終更新: