Real-Time Intelligence チュートリアルパート 2: デジタルツインビルダー (プレビュー) におけるストリーミングデータの取得と処理

2025-05-19

チュートリアルのこの部分では、別の種類のサンプルデータ (バスの場所に関する時系列情報を含むサンプルバスデータのリアルタイムデータストリーム) を設定します。サンプルデータをイベントハウスにストリーミングし、データに対していくつかの変換を実行してから、前のセクションで作成したサンプルデータ lakehouse にイベントハウスデータを取得するショートカットを作成します。デジタルツインビルダーでは、データをレイクハウスに格納する必要があります。

イベントハウスを作成する

チュートリアルリソースを作成するワークスペースを参照します。すべてのリソースを同じワークスペースに作成する必要があります。
+ 新規アイテム を選択します。
[項目の種類でフィルター処理する] 検索ボックスに「Eventhouse」と入力します。
Eventhouse 項目を選択します。
イベントハウス名として「Tutorial」と入力します。 KQL データベースは、同じ名前で同時に作成されます。
を選択してを作成します。プロビジョニングが完了すると、イベントハウスの [システムの概要] ページが表示されます。

eventstream を作成する

このセクションでは、サンプルのバスストリーミングデータを eventhouse に送信するイベントストリームを作成します。

ソースの追加

イベントストリームを作成し、 Buses サンプルデータをソースとして追加するには、次の手順に従います。

イベントハウスの [KQL データベース ] ウィンドウで、新しい Tutorial データベースを選択します。
メニューリボンから [データの取得 ] を選択し、[ Eventstream] > [新しいイベントストリーム] を選択します。
eventstream BusEventstream に名前を付けます。イベントストリームの作成が完了すると、イベントストリームが開きます。
[サンプルデータの使用] を選択します。
[ ソースの追加] ページで、ソース名として 「BusDataSource 」と入力します。 サンプルデータでバスを選択します。 [] を選択し、[] を追加します。

新しいイベントストリームの準備ができたら、作成キャンバスで開きます。

データを変換する

このセクションでは、受信サンプルデータに 1 つの変換を追加します。この手順では、文字列フィールドの ScheduleTime と Timestamp を DateTime 型にキャストし、わかりやすくするために Timestamp の名前を ArrivalTime に変更します。タイムスタンプフィールドを時系列データとして使用するには、デジタルツインビルダー (プレビュー) のタイムスタンプフィールドを DateTime 形式にする必要があります。

データ変換を追加するには、次の手順に従います。

[ イベントの変換] タイルまたは [変換先の追加 ] タイルの下矢印を選択し、[ フィールドの管理 ] の定義済み操作を選択します。タイルの名前が ManageFields に変更されます。
MangeFields タイルの編集アイコン (鉛筆のような形) を選択すると、[フィールドの管理] ウィンドウが開きます。
[ すべてのフィールドの追加] を選択します。このアクションにより、変換によってソースデータのすべてのフィールドが確実に存在します。
[タイムスタンプ] フィールドを選択します。 [種類の変更] を [はい] に切り替えます。 [ 変換された型] で、ドロップダウンリストから DateTime を選択します。 [名前] に、ActualTime の新しい名前を入力します。
保存せずに、[ ScheduleTime ] フィールドを選択します。 [種類の変更] を [はい] に切り替えます。 [ 変換された型] で、ドロップダウンリストから DateTime を選択します。名前は ScheduleTime のままにします。

[ 保存] を選択します。
[ フィールドの管理 ] ウィンドウが閉じます。 ManageFields タイルは、宛先に接続するまでエラーを表示し続けます。

宛先を追加する

メニューリボンから [宛先の追加] を選択し、[ Eventhouse] を選択します。

[Eventhouse]\(イベントハウス\) ウィンドウに次の情報を入力します。

フィールド	価値
データインジェストモード	インジェスト前イベント処理
宛先名	TutorialDestination
ワークスペース	リソースを作成したワークスペースを選択します。
Eventhouse	チュートリアル
KQL データベース	チュートリアル
KQL 宛先テーブル	新規作成 - テーブル名として bus_data_raw を入力します
入力データ形式	Json

[データソースの追加後にインジェストをアクティブ化する] ボックスがオンになっていることを確認します。
保存を選択します。
作成キャンバスで、[ ManageFields ] タイルを選択し、矢印を TutorialDestination タイルにドラッグして接続します。このアクションにより、フロー内のすべてのエラーメッセージが解決されます。
メニューリボンから [発行] を選択 します。これで、eventstream はサンプルストリーミングデータの eventhouse への送信を開始します。
数分後、Eventstream ビューの TutorialDestination カードには、[ データプレビュー ] タブにサンプルデータが表示されます。データが届くのを待っている間に、プレビューを数回更新することが必要になる場合があります。
イベントハウスでデータテーブルがアクティブであることを確認します。 チュートリアル KQL データベースに移動し、ビューを更新します。データを含む bus_data_raw というテーブルが含まれるようになりました。

更新ポリシーを使用してデータを変換する

バスストリーミングデータが KQL データベースに格納されたので、関数と Kusto 更新ポリシーを使用してデータをさらに変換できます。このセクションで実行する変換では、デジタルツインビルダー (プレビュー) で使用するデータを準備し、次のアクションを含めます。

JSON フィールド Properties を、含まれているデータ項目、 BusStatus 、 TimeToNextStationごとに個別の列に分割します。デジタルツインビルダーには JSON 解析機能がないため、データがデジタルツインビルダーに送信される前に、これらの値を分離する必要があります。
列 StopCodeを追加します。これは、各バス停を表す一意のキーです。この手順の目的は、このチュートリアルシナリオをサポートするためにサンプルデータセットを完了することです。個別のデータソースからの結合可能なエンティティインスタンスには、デジタルツインビルダーがそれらをリンクするために使用できる共通の列が含まれている必要があります。そのため、この手順では、静的バスストップデータセットの Stop_Code フィールドに一致するシミュレートされた int 値のセットを追加します。現実の世界では、関連するデータセットには既に何らかの共通点が含まれています。
変換されたバスデータを含む bus_data_processed という名前の新しいテーブルを作成します。
ショートカットを使用して Tutorial Lakehouse のデータにアクセスできるように、新しいテーブルに対して OneLake の可用性を有効にします。

変換クエリを実行するには、次の手順に従います。

イベントハウス内の Tutorial KQL データベースを選択します。メニューリボンから[ コードを含むクエリ]を選択すると、KQL クエリエディターが開きます。

次のコードをコピーしてクエリエディターに貼り付けます。各コードブロックを順番に実行します。

// Set columns
.create-or-alter function extractBusData ()
{
    bus_data_raw
    | extend BusState = tostring(todynamic(Properties).BusState)
        , TimeToNextStation = tostring(todynamic(Properties).TimeToNextStation)
        , StopCode = toint(10000 + abs(((toint(BusLine) * 100) + toint(StationNumber)) % 750))
    | project-away Properties
}

// Create table
.create table bus_data_processed (ActualTime:datetime, TripId:string, BusLine:string, StationNumber:string, ScheduleTime:datetime, BusState:string, TimeToNextStation:string, StopCode:int)

//Load data into table
.alter table bus_data_processed policy update
```
[{
    "IsEnabled": true,
    "Source": "bus_data_raw",
    "Query": "extractBusData",
    "IsTransactional": false,
    "PropagateIngestionProperties": true
}]
```

// Enable OneLake availability
.alter-merge table bus_data_processed policy mirroring dataformat=parquet with (IsEnabled=true, TargetLatencyInMinutes=5)

ヒント

コードを使用する代わりに、UI を使用して新しいテーブルの OneLake 可用性を有効にすることもできます。テーブルを選択し、 OneLake の可用性を切り替えます。

UI で OneLake の可用性を有効にするスクリーンショット。

UI オプションでは、既定の待機時間は、データの量に応じて 15 分から数時間です。待機時間を 5 分に短縮するには、前のコードブロックに示すように .alter-merge table コマンドを使用します。

必要に応じて、[クエリ] タブを Bus データ処理 として保存して、後で識別できるようにします。
bus_data_processedという名前の新しいテーブルがデータベースに作成されます。しばらく待つと、処理されたバスデータの設定が開始されます。

Lakehouse ショートカットを作成する

最後に、デジタルツインビルダー (プレビュー) のサンプルデータを保持する Tutorial lakehouse で処理されたバスデータを公開するショートカットを作成します。デジタルツインビルダーではデータソースがレイクハウスである必要があるため、この手順が必要です。

チュートリアルの lakehouse に移動してください (パート1「コンテキストデータをアップロードする」で以前に作成したものです)。メニューリボンから、[データの取得] >[新しいショートカット] を選択します。
[内部ソース] で、[Microsoft OneLake] を選びます。次に、 チュートリアル KQL データベースを選択します。
テーブルの一覧を展開し、bus_data_processedの横にあるチェックボックスをオンにします。 [次へ] を選択します。
ショートカットの詳細を確認し、[ 作成] を選択します。
bus_data_processedテーブルがレイクハウスで利用できるようになりました。データが含まれていることを確認します (これには数分かかる場合があります)。