Fabricを手動で立ち上げて、PythonからEventstream経由でEventhouseにデータを送ってみた

概要

こんにちは、クラスメソッド製造ビジネステクノロジー部の田中聖也です。
Microsoft Fabricを初めて触ることになって、まず「外部のプログラムからFabricにデータを送れるのか？」を手を動かして確かめてみました。
Eventstreamで受けて、Eventhouseのテーブルに貯めて、KQLで確認するところまでを手動で立ち上げたのですが、画面が多くてめっちゃ迷ったので、手順を画像つきで残すことにしました。

今回のゴールは、Pythonからテストイベントを1件送って、KQLでselectできるところまでです。
完成形はこちら

今回作るもの

PCからJSONを1件送って、Eventstreamで受けて、Eventhouseのテーブルに貯めるところまでを作ります。

PC(Python) ──JSON──▶ Eventstream ──▶ Eventhouse(KQLデータベース)

送るデータ

1イベント = 1つのJSONで、7項目を送ります。

{
  "event_id": "6f1c2b9e-...",
  "event_time": "2026-10-12T01:23:45.000Z",
  "person_id": "P001",
  "room_id": "ROOM_A",
  "event_type": "ENTER",
  "receiver_id": "RX_B",
  "sent_time": "2026-10-12T01:23:45.300Z"
}

テーブルの列と型は次のとおりです

時刻はUTC(協定世界時)のISO 8601(日時の標準的な書き方)形式で、末尾にZをつけて送ります
datetime列に取り込むときに型が崩れるのを防ぐためです

準備

今回の検証環境↓

作成するアイテムの名前↓

ワークスペース以外はアンダースコア区切りにしています

やってみた

手順は全部で8つです。

1. ワークスペースを作る
2. Eventhouseを作る
3. テーブル・マッピング・ポリシーをKQLで作る
4. Eventstreamを作ってソース(カスタムエンドポイント)を追加する
5. 宛先(Eventhouse)を追加して発行する
6. 接続文字列を.envに書く
7. 取り込み設定(データの取得ウィザード)を完了する
8. 動作確認

手順1: ワークスペースを作る

app.fabric.microsoft.com にサインインして、まず試用版を開始します
右上のアカウントアイコンから「試用版を開始」を選べば開始できます(60日間無料)

試用版が始まったら、左メニューの「ワークスペース」→「新しいワークスペース」をクリックします
名前に ble-room-tracker-poc と入力して、「この名前は使用できます。」と出ればOKです↓

下の「詳細」を開いて、ライセンスモードで「Fabric 評価版」を選びます。
※英語UIや公式ドキュメントでは「Fabric Trial」と表記されています。
容量(Trial-...)は自動で入るので、それ以外は既定のまま「適用」をクリックします↓

手順2: Eventhouseを作る

作成したワークスペースで「新しい項目」をクリックして、一覧から「イベントハウス」を選びます。
右上の検索欄に「イベント」と入れると見つけやすいかもです。

名前に ble_room_tracker と入力して「Create」をクリックします。

Eventhouseの「システムの概要」画面が開きます。
左側の「KQLデータベース」を見ると、同じ名前の ble_room_tracker が自動で作られています。

手順3: テーブル・マッピング・ポリシーをKQLで作る

左側の「KQLデータベース」にある ble_room_tracker をクリックします。

ツリーに表示される ble_room_tracker_queryset(クエリを書く画面)をクリックします。

エディターに、次のKQL(Kusto Query Language、Eventhouseで使うクエリ言語)を貼り付けます。
やっているのは次の4つです。

• テーブルの作成
• JSONマッピング(JSONのどの項目をテーブルのどの列に入れるかの対応表)の作成
• 保持ポリシー(データを何日間残すかの設定)の指定
• ingestion_time()(Eventhouseに取り込まれた時刻を返す関数)が使えるかの確認

.create table presence_events (
    event_id: guid, event_time: datetime, person_id: string, room_id: string,
    event_type: string, receiver_id: string, sent_time: datetime)

// JSONマッピング(取り込みウィザードで選ぶ)
.create table presence_events ingestion json mapping "presence_events_mapping"
'[{"column":"event_id","path":"$.event_id","datatype":"guid"},{"column":"event_time","path":"$.event_time","datatype":"datetime"},{"column":"person_id","path":"$.person_id","datatype":"string"},{"column":"room_id","path":"$.room_id","datatype":"string"},{"column":"event_type","path":"$.event_type","datatype":"string"},{"column":"receiver_id","path":"$.receiver_id","datatype":"string"},{"column":"sent_time","path":"$.sent_time","datatype":"datetime"}]'

// 保持ポリシーを明示する(PoC中は無期限扱い)
.alter-merge table presence_events policy retention softdelete = 36500d

// ingestion_time() が有効か確認する
.show table presence_events policy ingestiontime

貼り付けるとこうなります↓

ここからが大事で、段落(空行で区切られたまとまり)ごとにカーソルをその中に置いて「実行」(またはShift + Enter)を押します。
上から順に4回実行します。

実行結果はそれぞれこうなりました↓

左側のツリーのTablesに presence_events が表示されていれば完了です。

ハマりどころ1: KQLの「実行」はカーソルのある段落だけを実行する

正直、一番最初にハマったのがここでした
4つのKQLをまとめて貼り付けて実行したところ、次のエラーが返ってきました↓

Entity ID 'presence_events' of kind 'Table' was not found.

原因は、「実行」がエディター全体ではなく、カーソルがある段落だけを実行する仕様でした。
カーソルが最後の .show の行にあったので、テーブルを作る前に確認だけが実行されていました。
また、. で始まる管理コマンドは1回に1つしか実行できません。
段落ごとにカーソルを置いて、上から順に実行し直せば解決できました。

4つ目の結果が IsEnabled: true でない(または空)ときは、次を実行してからもう一度確認します

.alter table presence_events policy ingestiontime true

手順4: Eventstreamを作ってソースを追加する

ワークスペースに戻って「新しい項目」から「Eventstream」を選びます。

名前に ble_room_tracker_es と入力して「作成」をクリックします
「Enable schema-aware eventstream (Preview)」というチェックがありますが、今回は使わないのでオフのままにしました。

編集画面が開いたら、「カスタム エンドポイントを使用する」を選びます。

ソース名に receiver_a と入力して「追加」をクリックします
ここにも「スキーマの関連付け(プレビュー)」がありますが、こちらもアクティブ化せずに進めます。

キャンバスに receiver_a → ble_room_tracker_es のノードが並べばOKです。
右側の「イベントの変換または変換先の追加」から「宛先の追加」に進みます。

手順5: 宛先(Eventhouse)を追加して発行する

ツールバーの「宛先の追加」をクリックして、右側パネルの「イベントハウス」にある「構成」をクリックします。

次の内容を設定して「保存」をクリックします

インジェストモードは2つ選べます。

今回の送信側はテーブルと同じ形のJSONを送るので、「直接インジェスト」になります。
画面にも書いてありますが、取り込みが始まると後から変更できないので注意です。

キャンバスが receiver_a → ble_room_tracker_es → presence_events とつながったら、右上の緑の「発行」をクリックします↓

ハマりどころ2: 保存しただけでは反映されない

画面上部に「編集モード 公開すると、変更は有効になります。」と出ている間は、まだ下書きの状態です。
「保存」を押しただけでは動かなくて、「発行」を押して初めて有効になります。

発行が終わると、宛先ノードが「未構成」の状態で表示されます。

手順6: 接続文字列を.envに書く

次の手順7のウィザードでは、届いたデータの見本が必要になります。
なので、先にPCからサンプル用のデータを送る必要があります。

Eventstreamの画面で receiver_a ノードをクリックして、下の詳細欄で「SAS キー認証」→「Event Hub」タブを開きます
SASキー(Shared Access Signature、接続用の署名つきの鍵)の画面に、イベントハブ名と接続文字列が並んでいます。

コピーするのは次の2つです(目のアイコンで表示してからコピーします)

1. イベント ハブ名
2. 接続文字列 - 主キー

.envファイルを作って、値を書き込みます。

EVENTSTREAM_CONNECTION_STRING=Endpoint=sb://xxxx.servicebus.windows.net/;SharedAccessKeyName=key_xxxx;SharedAccessKey=xxxx;EntityPath=es_xxxx
EVENTSTREAM_NAME=es_xxxx

送信コードの要点

送信側のコードは、Azure Event Hubsのサンプルとほぼ同じです。
カスタムエンドポイントはEvent Hubs互換なので、azure-eventhubのSDKがそのまま使えます。

import asyncio
import json
import os
import uuid
from datetime import datetime, timezone

from azure.eventhub import EventData
from azure.eventhub.aio import EventHubProducerClient
from dotenv import load_dotenv

def utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z"

async def main() -> None:
    event = {
        "event_id": str(uuid.uuid4()),
        "event_time": utc_now(),
        "person_id": "P_TEST",
        "room_id": "ROOM_A",
        "event_type": "ENTER",
        "receiver_id": "RX_A",
        "sent_time": utc_now(),
    }
    client = EventHubProducerClient.from_connection_string(
        os.environ["EVENTSTREAM_CONNECTION_STRING"],
        eventhub_name=os.environ["EVENTSTREAM_NAME"],
    )
    async with client:
        batch = await client.create_batch()
        batch.add(EventData(json.dumps(event)))
        await client.send_batch(batch)

load_dotenv()
asyncio.run(main())

Fabric向けに書いたコードはほぼありません
実際のPoCでは、BLEやMQTTを使わずにテストイベントを1件だけ送る send_test_event.py を用意しました

手順7: 取り込み設定(データの取得ウィザード)を完了する

直接インジェストの場合、発行した後にEventhouse側で「どのテーブルに、どのマッピングで書き込むか」を設定します。

発行前: Eventstreamの宛先で「直接インジェスト」を選ぶ(テーブルはまだ選べない)
   ↓ 発行
発行後: 宛先ノードの「構成」→ データの取得ウィザード
        Configure(テーブルを選ぶ) → Inspect(見本データで列の対応を確認) → Summary(完了)

宛先ノードの「構成」をクリックすると、「データの取得」ウィザードが開きます
Configure画面では、左側で既存のテーブル presence_events を選びます(「新しいテーブル」は選びません)
右側は自動で入った値のまま「次へ」をクリックします

Inspect画面では、形式が「JSON」、マッピングが presence_events_mapping になっていることを確認します。

ハマりどころ3: 見本データがないと「終了」が押せない

この時点ではまだ1件も送っていないので、次のように表示されて「終了」が押せません↓

イベント ストリームからデータが読み取られませんでした。

設定の誤りではなくて、ウィザードが届いたデータの見本を使って列の対応を確認するための仕様です。
ウィザードは開いたままにして、別のターミナルでテストイベントを1件送ります(device/ フォルダで実行)

.venv\Scripts\python -m ble_room_tracker.send_test_event

次のように表示されれば送信成功です。

送信しました：event_id=... person_id=P_TEST room_id=ROOM_A

ウィザードに戻って「再試行」をクリックします。
しばらく「イベント ストリームからのデータを待機しています」と表示されます↓

「データ サンプルが見つかりました。」と表示されて、送ったイベントが7列に分かれて見えれば成功です↓

「終了」をクリックすると、Summary画面で既存のテーブル・マッピング・データ接続の3つが完了と表示されます↓

Eventstreamの画面で、receiver_a と presence_events がどちらも「アクティブ」になっていれば完了です↓

ここで一番効いたのは、送る側を先に用意してからウィザードを開く、という順番でした!!
手順6で.envを先に書いておいたのは、このためです

なお、ウィザード用に送った1件目は、データ接続ができる前のものなのでテーブルには入りません
想定どおりの動作なので問題ありません

手順8: 動作確認

データ接続ができた後のデータを確認したいので、テストイベントをもう1件送ります

.venv\Scripts\python -m ble_room_tracker.send_test_event

1分ほど待ってから、ble_room_tracker_queryset で次のKQLを実行します
エディターの末尾に空行を空けて貼り付けて、カーソルをその中に置いて実行します

presence_events
| extend ingest_time = ingestion_time()
| order by event_time desc
| take 10

送ったときに表示された event_id の行があって、person_id = P_TEST、room_id = ROOM_A などの値がそのまま入っていればOKです↓

ちゃんと入っていることを確認できました。

続けて、列の型が定義どおりかを getschema で確認します

presence_events | getschema

KQLデータベースの画面でも、presence_events に1行取り込まれていることが確認できます。

送信時刻(sent_time)と取り込み時刻(ingestion_time())の差は約44.6秒でした。
※データ接続を作った直後の1件だけの値なので、あくまで参考値です。

ワークスペースのHomeを見ると、Eventhouse、KQLデータベース、Eventstreamの3つのアイテムが並んでいます↓

まとめ

初めてのFabricでしたが、画面操作だけで、PythonからEventstreamを経由してEventhouseにデータを貯める受け口が作れました！
画面操作だけで必要なリソースが立ち上がるので、結構楽でいいなと思いました。