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

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

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

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

eventhouse システムの概要 ページが表示されます。

eventstream を作成する

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

ソースの追加

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

イベントハウスの [システム概要] ページで、KQL データベースの一覧から新しい Tutorial データベースを選択します。
メニューリボンから [データの取得 ] を選択し、[ Eventstream] > [新しいイベントストリーム] を選択します。
名前として 「BusEventstream 」と入力し、[ 作成] を選択します。 eventstream の準備ができたら、イベントストリームが開きます。
[サンプルデータの使用] を選択します。
[ソースの追加] ページで、[サンプルデータ] で [バス] を選択します。ソース名として 「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 という名前の新しいテーブルを作成します。
ショートカットを使用して TutorialLH lakehouse 内のデータにアクセスできるように、新しいテーブルに対して OneLake を利用できるようにします。

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

KQL データベース のTutorial_querysetを 選択して、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:long, StationNumber:long, 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 ショートカットを作成する

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

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