eventstream-authoring-cli

更新チェック — セッションごとに1回（必須） このスキルをセッションで初めて使用する際は、続行する前に check-updates スキルを実行してください。

• GitHub Copilot CLI / VS Code：check-updates スキルを呼び出してください。
• Claude Code / Cowork / Cursor / Windsurf / Codex：ローカルとリモートの package.json のバージョンを比較してください。
• このセッション中にすでにチェックを実行済みの場合は、省略してください。

重要な注意事項

1. ワークスペース名から詳細（ID を含む）を確認するには、すべてのワークスペースを一覧表示してから、JMESPath でフィルタリングしてください。
2. ワークスペース ID、項目の種類、項目名から項目の詳細（ID を含む）を確認するには、そのワークスペース内の該当する種類の項目をすべて一覧表示してから、JMESPath でフィルタリングしてください。
3. Eventstream と Eventhouse は異なります。Eventstream は、リアルタイムのイベント取り込みおよびルーティングのパイプラインです。KQL データベースの操作には、eventhouse-authoring-cli または eventhouse-consumption-cli を使用してください。

Eventstream 作成 — CLI スキル

目次

タスク	参照先	備考
Fabric のワークスペースとアイテムの検索	COMMON-CLI.md § Fabric のワークスペースとアイテムの検索	必須 — 最初にリンクを読む［名前からワークスペース ID を、または名前、アイテムの種類、ワークスペース ID からアイテム ID を検索する際に必要］
Fabric のトポロジと主要な概念	COMMON-CORE.md § Fabric のトポロジと主要な概念
環境 URL	COMMON-CORE.md § 環境 URL
認証とトークンの取得	COMMON-CORE.md § 認証とトークンの取得	audience が間違っていると 401 が発生します。認証に関する問題が起きた場合は、まずお読みください
コアのコントロールプレーン REST API	COMMON-CORE.md § コアのコントロールプレーン REST API	ページネーション、LRO のポーリング、レート制限のパターンを含みます
注意点、ベストプラクティス、トラブルシューティング	COMMON-CORE.md § 注意点、ベストプラクティス、トラブルシューティング
ツール選択の根拠	COMMON-CLI.md § ツール選択の根拠
認証のレシピ	COMMON-CLI.md § 認証のレシピ	az login のフローとトークンの取得
az rest による Fabric コントロールプレーン API	COMMON-CLI.md § az rest による Fabric コントロールプレーン API	必ず --resource を渡してください。ページネーションと LRO のヘルパーを含みます
注意点とトラブルシューティング（CLI 固有）	COMMON-CLI.md § 注意点とトラブルシューティング（CLI 固有）	az rest の audience、シェルのエスケープ、トークンの有効期限
クイックリファレンス	COMMON-CLI.md § クイックリファレンス	az rest のテンプレートとトークン audience／ツールの対応表
Eventstream リソースモデル	EVENTSTREAM-AUTHORING-CORE.md § Eventstream リソースモデル	最初に読む — ソース、オペレーター、ストリーム、宛先で構成されるグラフベースのトポロジ
ソースの構成	EVENTSTREAM-AUTHORING-CORE.md § ソースの構成	API がサポートする 25 種類のソースと、ソースごとのプロパティ
変換オペレーター	EVENTSTREAM-AUTHORING-CORE.md § 変換オペレーター	8 種類のオペレーター：Filter、Aggregate、GroupBy、Join、ManageFields、Union、Expand、SQL
宛先の構成	EVENTSTREAM-AUTHORING-CORE.md § 宛先の構成	API がサポートする 4 種類の宛先とノードスキーマ
ストリームの種類	EVENTSTREAM-AUTHORING-CORE.md § ストリームの種類	DefaultStream（自動）と DerivedStream（オペレーターから生成）
Eventstream のライフサイクル（REST API）	EVENTSTREAM-AUTHORING-CORE.md § Eventstream のライフサイクル（REST API）	CRUD と Definition エンドポイント
アイテム定義とデプロイ	EVENTSTREAM-AUTHORING-CORE.md § アイテム定義とデプロイ	eventstream.json の Base64 エンコードパターン
注意点と制限事項	EVENTSTREAM-AUTHORING-CORE.md § 注意点と制限事項	カスタムエンドポイントは最大 11 個、Base64 エンコード、名前の制約
Eventstream の作成	SKILL.md § Eventstream の作成
トポロジ全体のデプロイ	SKILL.md § トポロジ全体のデプロイ	エンドツーエンド：トポロジ JSON の作成 → Base64 エンコード → 定義の送信
Eventstream トポロジの更新	SKILL.md § Eventstream トポロジの更新
Eventstream の削除	SKILL.md § Eventstream の削除
注意点、ルール、トラブルシューティング	SKILL.md § 注意点、ルール、トラブルシューティング	必ず実施／回避／優先チェックリスト

---

Eventstream の作成

空の Eventstream 項目を作成し、定義 API を介してソース、送信先、オペレーターを構成します。

ステップ 1: 項目の作成

az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com" \
  --headers "Content-Type=application/json" \
  --body '{"displayName": "my-eventstream", "description": "IoT sensor pipeline"}'

返された id を EVENTSTREAM_ID として保存します。

ステップ 2: トポロジーの構築

ソース、ストリーム、オペレーター、送信先を含む eventstream.json トポロジーを構築します。各ノードは inputNodes を介して上流を参照します。

シリアル化エラーを避けるため、JSON はプログラムで構築することを推奨します。主なルール:

• トポロジーには DefaultStream がちょうど 1 つ必要です — すべてのソースは inputNodes を介してそこに供給されます
• オペレーターは inputNodes[].name を介して入力を参照します
• DerivedStream は properties に inputSerialization が必要です
• 送信先は入力ストリームまたはオペレーターを参照します

ステップ 3: 定義のデプロイ

トポロジー JSON を Base64 エンコードし、定義 API を介して送信します。ペイロード構造の全体については、項目の定義とデプロイ を参照してください。

---

完全なトポロジーのデプロイ

トポロジーを含む完全な Eventstream を 1 回の API 呼び出しでデプロイするには、Create Item with Definition エンドポイントを使用します:

# 1. Build eventstream.json content (topology)
TOPOLOGY_JSON='{"compatibilityLevel":"1.1","sources":[...],"streams":[...],"operators":[...],"destinations":[...]}'

# 2. Build eventstreamProperties.json (optional — controls retention and throughput)
PROPERTIES_JSON='{"retentionTimeInDays":1,"eventThroughputLevel":"Low"}'

# 3. Base64-encode both (no line wraps)
TOPOLOGY_B64=$(echo -n "$TOPOLOGY_JSON" | base64 -w 0)
PROPERTIES_B64=$(echo -n "$PROPERTIES_JSON" | base64 -w 0)

# 4. Submit via Items API
az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/items" \
  --resource "https://api.fabric.microsoft.com" \
  --headers "Content-Type=application/json" \
  --body "{
    \"displayName\": \"my-eventstream\",
    \"type\": \"Eventstream\",
    \"definition\": {
      \"parts\": [
        {
          \"path\": \"eventstream.json\",
          \"payload\": \"${TOPOLOGY_B64}\",
          \"payloadType\": \"InlineBase64\"
        },
        {
          \"path\": \"eventstreamProperties.json\",
          \"payload\": \"${PROPERTIES_B64}\",
          \"payloadType\": \"InlineBase64\"
        }
      ]
    }
  }"

Note: eventstreamProperties.json を省略した場合、API はデフォルト値を適用します: retentionTimeInDays: 1、eventThroughputLevel: "Low"。保持期間 (1～90 日) とスループットを制御するには、明示的に含めてください。

Windows (PowerShell) では、base64 エンコードに [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($json)) を使用します。

---

Eventstream トポロジーの更新

1. 現在の定義を取得: POST /v1/workspaces/{wsId}/eventstreams/{esId}/getDefinition
2. base64 から eventstream.json ペイロードをデコードします
3. トポロジーを変更します (ノードの追加/削除/更新)
4. base64 に再エンコードします
5. 送信: POST /v1/workspaces/{wsId}/eventstreams/{esId}/updateDefinition

API Note: Eventstream Definition API は、/definition リソースに対する GET/PUT ではなく、アクション動詞 (getDefinition、updateDefinition) を伴う POST を使用します。これは Fabric Items Definition パターンに従っています。詳細は公式ドキュメントを参照してください。

Update Definition API は、長時間実行操作に対して 202 Accepted を返します。完了するまで Location ヘッダーの URL をポーリングしてください。

フィルターオペレーターの追加

⚠️ CRITICAL: フィルターオペレーターの条件は、column と value にネストされたオブジェクトを使用します — 裸の文字列ではありません。以下に示すオブジェクト形式ではなく "column": "temperature" を使用すると、サイレントに API に拒否されます。

{
  "name": "FilterHighTemp",
  "type": "Filter",
  "inputNodes": [{"name": "my-stream"}],
  "properties": {
    "conditions": [{
      "column": {
        "node": null,
        "columnName": "temperature",
        "columnPath": null,
        "expressionType": "ColumnReference"
      },
      "operatorType": "GreaterThan",
      "value": {
        "dataType": "Float",
        "value": "30.0",
        "expressionType": "Literal"
      }
    }]
  }
}

すべてのオペレーター条件フィールドに必要な構造:

• column → オブジェクト: {node, columnName, columnPath, expressionType: "ColumnReference"}
• value → オブジェクト: {dataType, value, expressionType: "Literal"}
• operatorType → 文字列: Equals、NotEquals、GreaterThan、GreaterThanOrEquals、LessThan、LessThanOrEquals、Contains、DoesNotContain、StartsWith、DoesNotStartWith、EndsWith、DoesNotEndWith、IsEmpty、IsNull、IsNotNull、IsNotNullOrEmpty
• dataType → BigInt、Float、Nvarchar(max)、DateTime、Bit

この同じネストされたオブジェクトのパターンは、列を参照するすべてのオペレーター (Filter、Aggregate、GroupBy、Join、ManageFields) に適用されます。

---

Eventstream の削除

az rest --method DELETE \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}" \
  --resource "https://api.fabric.microsoft.com"

成功すると 200 OK を返します。

---

注意点、ルール、トラブルシューティング

必ず行うこと

• 定義を送信する前に、eventstream.json のペイロードを必ず Base64 エンコードします
• az rest を呼び出す際は、必ず --resource https://api.fabric.microsoft.com を指定します
• ワークスペース名 → ID、アイテム名 → ID の解決には、必ず JMESPath フィルタリングを使用します
• 演算子の列／値参照には、必ずネストされたオブジェクトを使用します — "column": {"columnName": "x", "expressionType": "ColumnReference", ...} を使用し、"column": "x" は使用しません（API は文字列のみの値を黙って拒否します）
• トポロジあたりの DefaultStream は必ず 1 つだけにします — すべてのソースをそこに接続します（API は複数の DefaultStream を拒否します）
• LRO の応答をポーリングします — Update Definition は Location ヘッダーとともに 202 Accepted を返します

推奨事項

• 手動で文字列を組み立てるのではなく、トポロジ JSON をプログラムで生成します
• テストやプロトタイピングには SampleData ソースタイプを使用します
• デフォルト値に依存せず、retentionTimeInDays を明示的に設定します
• ソース設定でクラウド接続を参照する前に、接続を検証します
• 演算子の出力を Real-Time Hub で利用できるように、DerivedStreams を使用します

避けること

• 定義ペイロードに生の JSON を使用しないでください — Base64 エンコードが必要です
• Eventstream の表示名にアンダースコアやドットを使用しないでください（SQL 演算子が動作しなくなります）
• ユーザー定義のトポロジノード名（ソース、演算子、DerivedStreams、宛先）にハイフン、アンダースコア、ドット、スペースを使用しないでください — 英数字の PascalCase のみ使用できます（例：filter-temperature や filter_temperature ではなく、FilterTemperature を使用します）。例外：DefaultStream 名はプラットフォームによって {eventstreamName}-stream として自動生成され、ハイフンを含む場合があります — 名前を変更しないでください
• CustomEndpoint ソースと CustomEndpoint／Eventhouse 直接取り込み先の合計数を 11 個を超えないようにしてください
• Eventstream と Eventhouse を混同しないでください — これらは別個の Fabric ワークロードです
• ワークスペース ID やアイテム ID をハードコードしないでください — 必ず API 経由で検出します

---

例

プラットフォームに関する注意 — 例では PowerShell を使用します。JSON 本文は必ず [IO.File]::WriteAllText() を使って一時ファイルに書き込み（BOM なし）、インラインの --body "..." は cmd.exe によって破損する可能性があるため使用せず、 az rest に --body "@$file" を渡します。改行の問題を避けるため、 ConvertTo-Json では -Compress を使用します。安全にインラインで指定できる唯一の例外は、空の本文に対する --body '{}' です。

例 1：ソースを含む Eventstream を作成する

プロンプト：「dev ワークスペースに、サンプルデータソースを含む SensorIngestion という Eventstream を作成してください。」

# 1. Discover workspace ID
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

# 2. Create empty Eventstream
$esBody = @{ displayName = "SensorIngestion"; description = "IoT sensor pipeline" } | ConvertTo-Json -Compress
$bodyFile = Join-Path ([IO.Path]::GetTempPath()) "es_create.json"
[IO.File]::WriteAllText($bodyFile, $esBody, [System.Text.UTF8Encoding]::new($false))
$created = az rest --method post `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --headers "Content-Type=application/json" `
  --body "@$bodyFile" | ConvertFrom-Json

# 3. Get the created Eventstream ID from response
$esId = $created.id
if (-not $esId) { throw "Eventstream creation did not return an ID" }

# 4. Build topology — DefaultStream uses inputNodes (not parentName)
$topology = @{
    compatibilityLevel = "1.0"
    sources = @(@{
        name = "SampleSource"
        type = "SampleData"
        properties = @{ type = "Bicycles" }
    })
    streams = @(@{
        name = "SensorIngestion-stream"
        type = "DefaultStream"
        properties = @{}
        inputNodes = @(@{ name = "SampleSource" })
    })
    operators = @()
    destinations = @()
}
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))

# 5. Deploy definition
$defBody = @{
    definition = @{
        parts = @(@{
            path = "eventstream.json"
            payload = $topologyB64
            payloadType = "InlineBase64"
        })
    }
} | ConvertTo-Json -Depth 5 -Compress
$defFile = Join-Path ([IO.Path]::GetTempPath()) "es_def.json"
[IO.File]::WriteAllText($defFile, $defBody, [System.Text.UTF8Encoding]::new($false))
# updateDefinition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/updateDefinition" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $defFile -Raw -Encoding UTF8)

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $succeeded = $false
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') { $succeeded = $true; Write-Host "Update succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "updateDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "updateDefinition LRO timed out" }
}

例 2：DerivedStream を含むフィルター演算子を追加する

プロンプト：「SensorIngestion Eventstream に、No_Bikes > 5 のイベントだけを残すフィルターを追加し、フィルター後の出力を DerivedStream として公開してください。」

重要：Filter 演算子ノードを追加しただけでは、DefaultStream の出力先は切り替わりません。 フィルター後の出力を利用できるようにするには、inputNodes を介して DerivedStream（または宛先）をフィルターの出力に接続します。

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }

# 2. Get current definition (handles LRO via Location header)
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/getDefinition" `
  -Headers @{ Authorization = "Bearer $token"; "Content-Type" = "application/json" } `
  -Body '{}'

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $def = $null
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') {
            $def = Invoke-RestMethod -Uri "$location/result" `
              -Headers @{ Authorization = "Bearer $token" }
            break
        } elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "getDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $def -or -not $def.definition) { throw "getDefinition LRO timed out (last status: $(if ($poll) { $poll.status } else { 'unknown' }))" }
} else {
    $def = $response.Content | ConvertFrom-Json
}

# 3. Decode existing topology
$esPart = $def.definition.parts | Where-Object { $_.path -eq 'eventstream.json' } | Select-Object -First 1
if (-not $esPart) { throw "eventstream.json part not found in definition" }
$topology = [Text.Encoding]::UTF8.GetString(
  [Convert]::FromBase64String($esPart.payload)) | ConvertFrom-Json

# 4. Add Filter operator (PascalCase name — no underscores or hyphens)
#    Column: expressionType + columnName; Value: expressionType + dataType + value
$filter = @{
    name = "FilterLowBikes"
    type = "Filter"
    inputNodes = @(@{ name = "SensorIngestion-stream" })
    properties = @{
        conditions = @(@{
            operatorType = "GreaterThan"
            column = @{
                expressionType = "ColumnReference"
                node = $null
                columnName = "No_Bikes"
                columnPath = $null
            }
            value = @{
                expressionType = "Literal"
                dataType = "BigInt"
                value = "5"
            }
        })
    }
}
$existingOps = @($topology.operators | Where-Object { $_ -ne $null })
$topology.operators = $existingOps + @($filter)

# 5. Add DerivedStream wired to filter output (makes filtered data available)
$derivedStream = @{
    name = "FilteredOutput"
    type = "DerivedStream"
    properties = @{
        inputSerialization = @{ type = "Json"; properties = @{ encoding = "UTF8" } }
    }
    inputNodes = @(@{ name = "FilterLowBikes" })
}
$existingStreams = @($topology.streams | Where-Object { $_ -ne $null })
$topology.streams = $existingStreams + @($derivedStream)

# 6. Re-encode and update
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))
$esPart.payload = $topologyB64

$defBody = @{ definition = @{ parts = $def.definition.parts } } | ConvertTo-Json -Depth 5 -Compress
$defFile = Join-Path ([IO.Path]::GetTempPath()) "es_def.json"
[IO.File]::WriteAllText($defFile, $defBody, [System.Text.UTF8Encoding]::new($false))
# updateDefinition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/updateDefinition" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $defFile -Raw -Encoding UTF8)

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $succeeded = $false
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') { $succeeded = $true; Write-Host "Update succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "updateDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "updateDefinition LRO timed out" }
}

例 3：トポロジ全体をデプロイする（インライン定義を使用して作成）

プロンプト：「Custom Endpoint ソース、高価値イベント用のフィルター、フィルター後の出力用の DerivedStream を含む、EventPipeline という完全な Eventstream を作成してください。」

注：この例では、作成後に更新するのではなく、Fabric Items API（POST /items）を使用して 1 回の呼び出しで定義を含む Eventstream を作成します。

# 1. Discover workspace ID
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

# 2. Build complete topology with filter + DerivedStream
$topology = @{
    compatibilityLevel = "1.0"
    sources = @(@{
        name = "CustomSource"
        type = "CustomEndpoint"
        properties = @{}
    })
    streams = @(
        @{
            name = "EventPipeline-stream"
            type = "DefaultStream"
            properties = @{}
            inputNodes = @(@{ name = "CustomSource" })
        }
        @{
            name = "FilteredEvents"
            type = "DerivedStream"
            properties = @{
                inputSerialization = @{ type = "Json"; properties = @{ encoding = "UTF8" } }
            }
            inputNodes = @(@{ name = "FilterPremium" })
        }
    )
    operators = @(@{
        name = "FilterPremium"
        type = "Filter"
        inputNodes = @(@{ name = "EventPipeline-stream" })
        properties = @{
            conditions = @(@{
                operatorType = "GreaterThan"
                column = @{
                    expressionType = "ColumnReference"
                        node = $null
                        columnName = "Amount"
                        columnPath = $null
                    }
                value = @{
                    expressionType = "Literal"
                    dataType = "BigInt"
                    value = "100"
                }
            })
        }
    })
    destinations = @()
}

# 3. Create with inline definition (single API call)
$topologyJson = $topology | ConvertTo-Json -Depth 10 -Compress
$topologyB64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($topologyJson))

$body = @{
    displayName = "EventPipeline"
    type = "Eventstream"
    definition = @{
        parts = @(@{
            path = "eventstream.json"
            payload = $topologyB64
            payloadType = "InlineBase64"
        })
    }
} | ConvertTo-Json -Depth 5 -Compress
$bodyFile = Join-Path ([IO.Path]::GetTempPath()) "es_create_full.json"
[IO.File]::WriteAllText($bodyFile, $body, [System.Text.UTF8Encoding]::new($false))

# Create-with-definition returns 202 Accepted (LRO). Use Invoke-WebRequest to capture headers.
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/items" `
  -Headers @{ Authorization = "Bearer $token" } `
  -ContentType "application/json; charset=utf-8" `
  -Body (Get-Content $bodyFile -Raw -Encoding UTF8)

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 10 }
    $succeeded = $false
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') { $succeeded = $true; Write-Host "Create succeeded"; break }
        elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "Create LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $succeeded) { throw "Create LRO timed out" }
} else {
    Write-Host "Created: $(($response.Content | ConvertFrom-Json).displayName)"
}

例 4：Eventstream を削除する

プロンプト：「dev ワークスペースから SensorIngestion Eventstream を削除してください。」

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='dev'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'dev' not found" }

$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }

# 2. Delete
az rest --method delete `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId" `
  --resource "https://api.fabric.microsoft.com"