บทช่วยสอน: สร้างแผนที่แบบเรียลไทม์โดยใช้ REST API และ Python

Fabric Maps สามารถแสดงภาพ ข้อมูลเชิงพื้นที่แบบเรียลไทม์ โดยเชื่อมต่อกับ ชุดข้อมูลอาคารเหตุการณ์ ที่อัปเดตอย่างต่อเนื่องผ่านการนําเข้า Eventstream

บทช่วยสอนนี้แตกต่างจากสถานการณ์แบบคงที่ที่ใช้ไฟล์ที่จัดเก็บไว้ใน Lakehouse บทช่วยสอนนี้สาธิต สถาปัตยกรรมที่ขับเคลื่อนด้วยเหตุการณ์แบบสตรีมมิ่ง ซึ่ง:

• ระบบจะนําเข้าเหตุการณ์ไปยังอีเวนต์เฮาส์
• ข้อมูลถูกสืบค้นโดยใช้ Kusto Query Language (KQL)
• แผนที่จะรีเฟรชแบบไดนามิกเมื่อมีข้อมูลใหม่มาถึง

บทช่วยสอนนี้มุ่งเน้นไปที่ ทําให้เวิร์กโฟลว์แบบ end-to-end เป็นแบบอัตโนมัติ โดยใช้ Fabric REST API และ Python เพื่อให้คุณสามารถจัดเตรียมทรัพยากรและกําหนดค่าประสบการณ์แผนที่แบบเรียลไทม์โดยทางโปรแกรมได้ สําหรับสถานการณ์ข้อมูลแบบคงที่โดยใช้ไฟล์ Lakehouse โปรดดู สร้างแผนที่แบบคงที่โดยใช้ REST API และ Python

ในบทช่วยสอนนี้ คุณจะได้เรียนรู้วิธีสร้างและทําให้โซลูชันเชิงพื้นที่แบบเรียลไทม์ใน Microsoft Fabric เป็นแบบอัตโนมัติโดยใช้ Eventstream, Eventhouse และ KQL

หากต้องการจําลองการสตรีมอย่างต่อเนื่องและดูการอัปเดตแผนที่แบบเกือบเรียลไทม์ ให้ทําบทแนะนําสอนการใช้งานนี้ให้เสร็จก่อน จากนั้นจึงดําเนินการติดตามผล บทช่วยสอน: จําลองการนําเข้าข้อมูลแบบเรียลไทม์สําหรับแผนที่โดยใช้ REST API และ Python ซึ่งสร้างขึ้นโดยตรงบน eventhouse, eventstream, ฟังก์ชัน KQL และแผนที่ที่คุณสร้างที่นี่

ภาพรวมสถานการณ์: การติดตามสินทรัพย์แบบเรียลไทม์

บทช่วยสอนนี้อิงตามสถานการณ์การติดตามสินทรัพย์แบบเรียลไทม์ คล้ายกับสถานการณ์การติดตามฟลีตที่ใช้ใน Fabric Maps ดั้งเดิม บทช่วยสอน: สร้างการกําหนดเส้นทางใบสั่งงานแบบเรียลไทม์ด้วย Fabric Maps

ในสถานการณ์สมมตินี้:

• พาหนะจะปล่อยการอัปเดตตําแหน่งเป็นระยะ
• เหตุการณ์สถานที่จะถูกนําเข้าไปยังบ้านงานอีเวนต์
• แผนที่แสดงตําแหน่งยานพาหนะล่าสุดและอัปเดตโดยอัตโนมัติเมื่อมีกิจกรรมใหม่มาถึง

รูปแบบนี้เป็นตัวแทนของกรณีการใช้งานแบบเรียลไทม์ทั่วไป เช่น:

• การติดตามยานพาหนะ
• การจัดส่งใบสั่งงาน
• การตรวจสอบทรัพย์สินและอุปกรณ์

Microsoft Fabric ใช้ Eventstream และ Eventhouse เพื่อนําเข้า ประมวลผล และวิเคราะห์ข้อมูลการสตรีมแบบเกือบเรียลไทม์ ทําให้สามารถแสดงภาพข้อมูลการดําเนินงานแบบสดได้โดยตรงบนแผนที่

บทช่วยสอนนี้เป็นไปตามรูปแบบการทํางานอัตโนมัติทั่วไปใน Fabric: สร้างโครงสร้างพื้นฐาน→นําเข้าสตรีม→ตรวจสอบการนําเข้า→แผนที่การแสดงผล

ข้อกำหนดเบื้องต้น

• Python 3.10 หรือใหม่กว่า
• Azure CLI
• รหัสพื้นที่ทํางาน Fabric
• สิทธิ์ในการเรียก Fabric REST API เช่น:
  • Item.ReadWrite.All

การพิสูจน์ตัวตน

บทช่วยสอนนี้ใช้ DefaultAzureCredentialซึ่งสามารถรับรองความถูกต้องโดยใช้แหล่งข้อมูลข้อมูลประจําตัวภายใน/ผู้พัฒนาหลายแหล่ง สําหรับผู้อ่านครั้งแรก วิธีที่ง่ายที่สุดคือการลงชื่อเข้าใช้ Azure CLI

ตรวจสอบสิทธิ์ภายในเครื่อง (แนะนําสําหรับการทํางานครั้งแรก)

1. เปิดเทอร์มินัล
2. วิ่ง:

az login

DefaultAzureCredential สามารถใช้ข้อมูลประจําตัวที่ลงชื่อเข้าใช้เพื่อรับโทเค็นเพื่อการเข้าถึงสําหรับสิ่งต่อไปนี้

• Fabric REST API (ทรัพยากร: https://api.fabric.microsoft.com/.default)
• การสืบค้นระนาบข้อมูล Kusto (KQL) กับบ้านเหตุการณ์ (ทรัพยากร: https://api.kusto.windows.net/.default)
• จุดสิ้นสุด Power BI / Fabric REST ที่ใช้สําหรับการสํารวจการดําเนินการที่ยาวนาน (ทรัพยากร: https://analysis.windows.net/powerbi/api/.default)

ลงชื่อเข้าใช้ Microsoft Fabric (แนะนํา)

ก่อนที่คุณจะเรียกใช้บทช่วยสอนนี้ เราขอแนะนําให้ลงชื่อเข้าใช้ Microsoft Fabric อย่างน้อยหนึ่งครั้ง:

https://app.fabric.microsoft.com

การลงชื่อเข้าใช้ช่วยให้แน่ใจว่าข้อมูลประจําตัว Fabric การเป็นสมาชิกบทบาทพื้นที่ทํางาน และการกําหนดความจุของคุณได้รับการจัดเตรียมอย่างสมบูรณ์ก่อนที่จะได้รับโทเค็นการเข้าถึง Microsoft Entra โดยทางโปรแกรม

ขั้นตอนนี้มีประโยชน์อย่างยิ่งในกรณีต่อไปนี้

• คุณเพิ่งเริ่มใช้ Microsoft Fabric
• พื้นที่ทํางานเพิ่งสร้างขึ้น
• เพิ่มการมอบหมายบทบาทของคุณเมื่อเร็วๆ นี้

สร้างไฟล์ข้อมูลเมล็ดพันธุ์ (ข้อมูลแผนที่เริ่มต้น)

เพื่อให้แน่ใจว่าแผนที่แสดงข้อมูลทันทีหลังจากการจัดสรร สคริปต์จะส่งเหตุการณ์เริ่มต้นชุดเล็กๆ ไปยังสตรีมเหตุการณ์

1. สร้างไฟล์ใหม่ในไดเรกทอรีเดียวกับสคริปต์ Python ของคุณ: vehicle_locations_seed.csv
2. วางเนื้อหาต่อไปนี้:

VehicleId,Latitude,Longitude,EventTime
V-001,47.6101,-122.3344,2026-01-01T10:00:00Z
V-002,47.6150,-122.3200,2026-01-01T10:00:00Z
V-003,47.6205,-122.3493,2026-01-01T10:00:00Z
V-004,47.6050,-122.3300,2026-01-01T10:00:00Z

ขั้นตอนที่ 1 - สร้างไฟล์โปรเจ็กต์ Python ใหม่

ในขั้นตอนนี้ คุณจะสร้างไฟล์ Python เปล่าที่คุณสร้างขึ้นทีละส่วน

สร้างไฟล์ใหม่ที่มีชื่อว่า

create_realtime_map.py

เปิดไฟล์ในเครื่องมือแก้ไข

ขั้นตอนที่ 2 - ติดตั้งไลบรารีที่จําเป็นและเพิ่มคําสั่งนําเข้าที่จําเป็น

ในขั้นตอนนี้ คุณติดตั้งการขึ้นต่อกัน และเพิ่มการนําเข้าที่สคริปต์ของคุณใช้

ติดตั้งไลบรารีที่จําเป็น

ในหน้าต่างเทอร์มินัลที่คุณเพิ่งเปิด ให้เรียกใช้คําสั่งต่อไปนี้:

pip install httpx azure-identity azure-eventhub

แต่ละห้องสมุดมีไว้เพื่ออะไร

• httpx: ส่งคําขอ HTTP ไปยัง Fabric REST API
• azure-identity: ให้ DefaultAzureCredential สําหรับการรับรองความถูกต้อง Microsoft Entra
• azure-eventhub: ส่งเหตุการณ์ Seed ไปยังจุดสิ้นสุดที่เข้ากันได้กับฮับเหตุการณ์ของ Eventstream เพื่อเติมข้อมูลในบ้านเหตุการณ์

เพิ่มคําสั่งนําเข้าลงในไฟล์ .py ของคุณ

ที่ด้านบนของ create_realtime_map.py ให้เพิ่มสิ่งต่อไปนี้

import base64
import csv
import json
import os
import time
import uuid

import httpx
from azure.eventhub import EventData, EventHubProducerClient
from azure.eventhub.exceptions import EventHubError
from azure.identity import DefaultAzureCredential

ขั้นตอนที่ 3 - เพิ่มส่วนการกําหนดค่า

ในขั้นตอนนี้ คุณกําหนดตัวแปรที่แอปพลิเคชันของคุณใช้ รวมถึงรหัสพื้นที่ทํางานและชื่อทรัพยากร

การรวมศูนย์การกําหนดค่าในคลาสเดียว Config — แทนที่จะกระจายค่าฮาร์ดโค้ดข้ามฟังก์ชัน — ให้ข้อได้เปรียบที่เป็นรูปธรรมสามประการแก่คุณ:

• การเคลื่อนย้ายสภาพแวดล้อม: รหัสพื้นที่ทํางาน ชื่อทรัพยากร และการตั้งค่าอื่นๆ อยู่ในที่เดียว ดังนั้นคุณจึงสามารถเรียกใช้สคริปต์ซ้ํากับพื้นที่ทํางานหรือเครื่องอื่นได้โดยการเปลี่ยนสองสามบรรทัด (หรือตัวแปรสภาพแวดล้อม) แทนการตามล่าผ่านโค้ด
• ลายเซ็นฟังก์ชันที่สะอาดขึ้น: ฟังก์ชันขั้นตอนยอมรับวัตถุเดียว cfg แทนรายการพารามิเตอร์ที่ยาว ซึ่งช่วยให้การประสานงานอ่าน main() ง่าย
• การจัดการข้อมูลลับที่ปลอดภัยยิ่งขึ้น: ค่าที่ละเอียดอ่อน เช่น รหัสพื้นที่ทํางาน ถูกโหลดจากตัวแปรสภาพแวดล้อม ดังนั้นจึงไม่เคยถูกส่งไปพร้อมกับสคริปต์

เพิ่มข้อความต่อไปนี้ด้านล่างคําสั่ง นําเข้า :

# =========================================================
# Configuration (centralized)
# =========================================================

class Config:
    """
    Central configuration: workspace ID, resource display names, and
    ingestion settings. A single instance is built in main() and passed
    to each step function.
    """
    def __init__(self):
        # Workspace
        self.workspace_id = os.environ.get("FABRIC_WORKSPACE_ID", "")
        if not self.workspace_id:
            raise RuntimeError("Set FABRIC_WORKSPACE_ID environment variable before running the script.")

        # Resource display names / descriptions
        self.eventhouse_display_name = "eh_realtime_locations"
        self.eventhouse_description = "Stores streaming location events for a Fabric Maps real-time tutorial"

        self.eventhouse_table_name = "VehicleLocations"
        self.kql_function_name = "LatestVehicleLocations"

        self.eventstream_display_name = "es_realtime_locations"
        self.eventstream_description = "Streams events into an eventhouse table (created via Eventstream REST API)"

        self.map_display_name = "My Real-Time Fabric Map"
        self.map_description = "Created using Fabric Maps REST API (Eventhouse + Eventstream + Kusto function)"

        # Map refresh
        self.refresh_interval_ms = 5000

        # Seed data (initial map data)
        self.seed_csv_path = os.path.join(os.path.dirname(__file__), "vehicle_locations_seed.csv")
        
        # Will be provided interactively after eventstream is created
        self.eventhub_connection_string = os.environ.get("EVENTHUB_CONNECTION_STRING", "")

ตั้งค่ารหัสพื้นที่ทํางานโดยใช้ตัวแปรสภาพแวดล้อม

แทนที่จะฮาร์ดโค้ดรหัสพื้นที่ทํางานโดยตรงในสคริปต์บทช่วยสอนนี้จะอ่านจากตัวแปรสภาพแวดล้อม วิธีนี้จะเก็บค่าเฉพาะสภาพแวดล้อมออกจากซอร์สโค้ด และช่วยให้คุณสามารถนําสคริปต์กลับมาใช้ใหม่ในพื้นที่ทํางานหรือเครื่องได้โดยไม่ต้องแก้ไข

ก่อนเรียกใช้สคริปต์ ให้สร้างตัวแปรสภาพแวดล้อมชื่อ FABRIC_WORKSPACE_ID.

ตั้งค่าตัวแปรสภาพแวดล้อมบน Windows

บน Windows คุณสามารถตั้งค่าตัวแปรจากเทอร์มินัลใดก็ได้ที่รองรับตัวแปรสภาพแวดล้อม เช่น PowerShell, Windows PowerShell, PowerShell หรือ Command Prompt windows ที่สร้างขึ้นใน Visual Studio และ Visual Studio Code, เทอร์มินัล Windows และส่วนใหญ่ เปลือกหอยอื่น ๆ

เรียกใช้สิ่งต่อไปนี้ใน PowerShell หรือเทอร์มินัลแบบรวม VS Code:

$env:FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

เพื่อยืนยันว่าตัวแปรถูกตั้งค่า:

echo $env:FABRIC_WORKSPACE_ID

สิ่งนี้จะตั้งค่าตัวแปรสําหรับเซสชันเทอร์มินัลปัจจุบันเท่านั้น

ตั้งค่าตัวแปรสภาพแวดล้อมถาวร (Windows)

หากต้องการให้ตัวแปรพร้อมใช้งานในเซสชันในอนาคต ให้ใช้วิธีใดวิธีหนึ่งต่อไปนี้

• PowerShell (บรรทัดเดียว): เรียกใช้setx FABRIC_WORKSPACE_ID "<WORKSPACE_ID>" setxคําสั่งจะเขียนไปยังสภาพแวดล้อมของผู้ใช้ แต่จะไม่อัปเดตเทอร์มินัลปัจจุบัน ให้ปิดและเปิดเทอร์มินัลอีกครั้ง (หรือเปิดเทอร์มินัลใหม่) ก่อนเรียกใช้สคริปต์
• จีโอ:
  1. เปิด คุณสมบัติของระบบ
  2. เลือก การตั้งค่าระบบขั้นสูง
  3. เลือกตัวแปรสภาพแวดล้อม
  4. ภายใต้ ตัวแปรผู้ใช้ ให้เลือก สร้าง
  5. ป้อน:
     • ชื่อ: FABRIC_WORKSPACE_ID
     • ค่า: รหัสพื้นที่ทํางานของคุณ
  6. เลือก ตกลง เพื่อบันทึก
  7. ปิดและเปิดเทอร์มินัลของคุณอีกครั้งก่อนที่จะเรียกใช้สคริปต์อีกครั้ง

ตั้งค่าตัวแปรสภาพแวดล้อมบน macOS หรือ Linux

บน macOS และ Linux คุณสามารถตั้งค่าตัวแปรจากเชลล์ใดก็ได้ที่รองรับ export—Bash, Zsh (ค่าเริ่มต้นใน macOS สมัยใหม่), Fish (ด้วยไวยากรณ์ที่แตกต่างกันเล็กน้อย) และเทอร์มินัลในตัวใน Visual Studio Code และตัวแก้ไขอื่นๆ

วิ่ง:

export FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

เพื่อยืนยันว่าตัวแปรถูกตั้งค่า:

echo $FABRIC_WORKSPACE_ID

สิ่งนี้ตั้งค่าตัวแปรสําหรับเซสชันเชลล์ปัจจุบันเท่านั้น

ตั้งค่าตัวแปรสภาพแวดล้อมถาวร (macOS หรือ Linux)

หากต้องการให้ตัวแปรพร้อมใช้งานในเซสชันในอนาคต ให้เพิ่มบรรทัดลงใน export โปรไฟล์เชลล์ของคุณ:

• Zsh (ค่าเริ่มต้นบน macOS): ~/.zshrc
• ทุบตี: ~/.bashrc (Linux) หรือ ~/.bash_profile (macOS)
• ปลา: เรียกใช้ set -Ux FABRIC_WORKSPACE_ID "<WORKSPACE_ID>" แทนการแก้ไขไฟล์

หลังจากอัปเดตโปรไฟล์แล้ว ให้เปิดเทอร์มินัลใหม่หรือเรียกใช้ source ~/.zshrc (หรือไฟล์ที่เหมาะสม) เพื่อให้การเปลี่ยนแปลงมีผล

ขั้นตอนที่ 4 - เพิ่มฟังก์ชันตัวช่วย

ในขั้นตอนนี้ คุณจะแยกข้อกังวลข้ามมิติ เช่น การรับรองความถูกต้อง การสร้างส่วนหัว การสํารวจการดําเนินการที่ทํางานเป็นเวลานาน และตรรกะการลองใหม่ลงในชุดตัวช่วยที่นํากลับมาใช้ใหม่ได้ขนาดเล็ก ซึ่งทุกขั้นตอนสามารถเรียกใช้ฟังก์ชันได้

การรวมศูนย์ข้อกังวลเหล่านี้ไว้ในผู้ช่วย แทนที่จะรวมไว้ในทุกไซต์การโทร ให้ข้อดีที่เป็นรูปธรรมสามประการแก่คุณ:

• แหล่งข้อมูลเดียวสําหรับข้อกังวลข้ามด้าน: การรับรองความถูกต้อง ส่วนหัว และการสํารวจ LRO เป็นสิ่งจําเป็นสําหรับการเรียก API เกือบทุกครั้ง การรวมศูนย์ทําให้ฟังก์ชันแต่ละขั้นตอนมุ่งเน้นไปที่ทรัพยากรของตัวเองแทนที่จะใช้ตรรกะการรับโทเค็นและลองใหม่อีกครั้ง
• ความยืดหยุ่นโดยไม่ยุ่งเหยิง: ตัวช่วยดูดซับเงื่อนไขชั่วคราว เช่น การจัดเตรียมแบบอะซิงโครนัส ความล่าช้าในการเผยแพร่แบ็กเอนด์ ความล้มเหลวในการส่งที่ลองใหม่ได้ ดังนั้นฟังก์ชันขั้นตอนจึงสั้นและอ่านเหมือนรายการตรวจสอบ
• สอนและแก้ไขได้ง่ายขึ้น: ตัวช่วยแต่ละคนจะถูกนํามาใช้เพียงครั้งเดียวและนํากลับมาใช้ใหม่ หาก Fabric เปลี่ยนรูปแบบ LRO หรือขอบเขตการรับรองความถูกต้อง คุณจะแก้ไขได้ในที่เดียว

ตัวช่วยที่คุณเพิ่มในขั้นตอนนี้คือ:

• ตัวช่วยการรับรองความถูกต้อง: สร้างส่วนหัวสําหรับ REST API Fabric (และปลายทาง LRO ของคลัสเตอร์ Power BI)
• FabricClient: ตัวห่อหุ้มน้ําหนักเบาสําหรับการเรียก API ที่สอดคล้องกัน
• ตัวจัดการ LRO: สํารวจการดําเนินการที่ทํางานเป็นเวลานานโดยใช้ Location / x-ms-operation-id / Retry-After รวมถึงการตอบสนอง 200-with-Running จุดสิ้นสุดคลัสเตอร์ Power BI และเพย์โหลดการเสร็จสมบูรณ์ตามสถานะเท่านั้น (แก้ไขโดย displayName)
• ตัวช่วยเพย์โหลดคําจํากัดความ: base64-encode map.json สําหรับคําจํากัดความแบบอินไลน์
• ตัวช่วยการเชื่อมต่อ Eventstream: พร้อมท์สําหรับ สายอักขระการเชื่อมต่อ ปลายทางแบบกําหนดเอง
• ตัวช่วยเมล็ดพันธุ์: ส่งเหตุการณ์เริ่มต้นพร้อมตรรกะการลองใหม่เพื่อให้แน่ใจว่าการนําเข้าสําเร็จ
• <ตัวช่วยความพร้อมของฐานข้อมูล c0>KQL: รอให้ฐานข้อมูล KQL พร้อมใช้งานสําหรับ Fabric Maps

สร้างฟังก์ชันตัวช่วยการรับรองความถูกต้อง

ทุก Fabric REST ที่บทช่วยสอนนี้ทําจะมีโทเค็นการเข้าถึง Microsoft Entra (โทเค็นผู้ถือ) ในส่วนหัว Authorization แทนที่จะได้มาซึ่งโทเค็นเฉพาะกิจขั้นตอนนี้จะห่อหุ้ม DefaultAzureCredential ด้วยขนาดเล็ก TokenProvider และเปิดเผยตัวสร้างส่วนหัวเฉพาะกลุ่มเป้าหมายสําหรับแต่ละตระกูลปลายทางที่สคริปต์เรียกใช้

การรวมศูนย์การได้มาซึ่งโทเค็นและการสร้างส่วนหัวในตัวช่วย — แทนที่จะรับโทเค็นในทุกไซต์การโทร — ให้ข้อได้เปรียบที่เป็นรูปธรรมสามประการแก่คุณ:

• ข้อมูลประจําตัวแบบรวมศูนย์: DefaultAzureCredential เดียวถูกห่อไว้ใน TokenProvider และนํามาใช้ซ้ําสําหรับการเรียก API ทุกครั้ง ดังนั้นการค้นพบข้อมูลประจําตัว (Azure CLI, VS Code, ข้อมูลประจําตัวที่มีการจัดการ ฯลฯ) จะเกิดขึ้นเพียงครั้งเดียว
• โทเค็นที่รับรู้ผู้ชม: Fabric, Kusto และ Power BI ตําแหน่งข้อมูลคลัสเตอร์ปฏิเสธโทเค็นที่ออกสําหรับผู้ชมที่ไม่ถูกต้อง เครื่องมือสร้างส่วนหัวแยกต่างหากต่อกลุ่มเป้าหมายจะเก็บขอบเขตที่ถูกต้องไว้ถัดจากไซต์การโทร ดังนั้นจึงเห็นได้ชัดว่าแต่ละฟังก์ชันกําหนดเป้าหมายปลายทางใด
• สดใหม่ในทุกคําขอ: ผู้ Authorization สร้างส่วนหัวสร้างส่วนหัวตามความต้องการแทนที่จะแคชโทเค็นเอง ข้อมูลประจําตัวพื้นฐานจะรีเฟรชอย่างโปร่งใส ดังนั้นไซต์การโทรจึงไม่ต้องคิดถึงวันหมดอายุ

บทช่วยสอนนี้เรียกใช้ REST API Fabric โดยใช้ขอบเขตที่ได้รับมอบหมาย เช่น Item.ReadWrite.All

เพิ่มสิ่งต่อไปนี้หลัง Config ชั้นเรียน

# =========================================================
# Auth helpers
#
# Authentication utilities built on `DefaultAzureCredential` that acquire and
# construct Authorization headers for calling Fabric REST APIs.
# =========================================================

class TokenProvider:
    """
    Thin wrapper around `DefaultAzureCredential` that acquires Entra access
    tokens. `_fabric_headers()` and `_pbi_headers()` call `get()` per
    request so the Authorization header is always fresh; the underlying
    credential refreshes transparently.
    """
    def __init__(self):
        self._cred = DefaultAzureCredential()

    def get(self, scope: str) -> str:
        return self._cred.get_token(scope).token


_tokens = TokenProvider()


def _fabric_headers() -> dict[str, str]:
    """
    Build headers for Fabric REST API calls.

    This function is called each time we make a Fabric REST call so the token is fresh.
    """
    return {
        "Authorization": f"Bearer {_tokens.get('https://api.fabric.microsoft.com/.default')}",
        "Content-Type": "application/json"
    }


def _kusto_headers() -> dict[str, str]:
    """
    Build headers for Kusto (Eventhouse `queryServiceUri`) management and query calls.
    """
    return {
        "Authorization": f"Bearer {_tokens.get('https://api.kusto.windows.net/.default')}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }


def _pbi_headers() -> dict[str, str]:
    """
    Build headers for polling Power BI cluster LRO endpoints
    (e.g., df-*.analysis.windows.net) that require a Power BI audience token.
    """
    return {
        "Authorization": f"Bearer {_tokens.get('https://analysis.windows.net/powerbi/api/.default')}",
        "Content-Type": "application/json"
    }

สร้าง Fabric Client Wrapper

การเรียก REST Fabric ส่วนใหญ่ในบทช่วยสอนนี้ส่งส่วนหัว Authorization และ Content-Type เดียวกัน แทนที่จะทําซ้ําในทุกไซต์การโทรบทช่วยสอนนี้จะห่อหุ้มhttpx.ClientFabricClientในส่วนหัวขนาดเล็กที่แนบส่วนหัวโดยอัตโนมัติในขณะที่ยังคงส่งคืนดิบhttpx.Responseเพื่อให้ผู้โทรแต่ละคนสามารถตรวจสอบรหัสสถานะได้ (ตัวอย่างเช่นเพื่อแยกความแตกต่าง201จาก 202)

การห่อแบบ httpx.Client นี้ — แทนที่จะส่งผ่าน headers=_fabric_headers() ทุกไซต์การโทร — ให้ข้อได้เปรียบที่เป็นรูปธรรมสองประการแก่คุณ:

• ส่วนหัวในที่เดียว: ทุกไซต์การโทรจะรับคําขอล่าสุด _fabric_headers() โดยอัตโนมัติ ดังนั้นจึงไม่สามารถส่งคําขอใหม่โดยไม่ได้ตั้งใจหากไม่มี Authorization ส่วนหัว
• รหัสสถานะยังคงมองเห็นได้: request() ส่งคืน JSON ดิบ httpx.Response แทน JSON ที่ถอดรหัส ดังนั้นไซต์การโทรจึงยังคงสามารถแยกสาขาตามสถานะ (201 เทียบกับ 202) และตรวจสอบส่วนหัว เช่น Location หรือ Retry-After สําหรับการจัดการ LRO

เพิ่มสิ่งต่อไปนี้หลังจากฟังก์ชันตัวช่วยการรับรองความถูกต้อง:

# =========================================================
# FabricClient (minimal wrapper so call sites stay clean)
# =========================================================

class FabricClient:
    """
    Small wrapper around httpx.Client so we don't repeat headers everywhere.

    Keeps the tutorial behavior:
    - request() returns the raw httpx.Response so the caller can handle 201 vs 202.
    """
    def __init__(self, http_client: httpx.Client):
        self._http = http_client

    def request(self, method: str, url: str, *, json_body=None) -> httpx.Response:
        return self._http.request(method, url, headers=_fabric_headers(), json=json_body)

สร้างฟังก์ชันตัวช่วย LRO

Fabric REST API หลายตัวที่ใช้ในบทช่วยสอนนี้ เช่น Create Eventhouse, Create Eventstream และ Create Map รองรับ long-running operations (LRO)

API เหล่านี้สามารถส่งคืนการตอบกลับได้หลายรูปแบบ:

• 201 Created ด้วยเนื้อหาทรัพยากรแบบอินไลน์ (ซิงโครนัส)
• 202 Accepted โดยมี Location ส่วนหัวชี้ไปที่ URL สถานะการทํางาน (อะซิงโครนัส)
• 202 Accepted ด้วย x-ms-operation-id ส่วนหัวแทน Location (อะซิงโครนัส, รูปแบบอื่น)
• 200 OK มี status: "Running" หรือ status: "NotStarted" ขณะทําโพล (ยังอยู่ระหว่างดําเนินการ)
• 200 OK โดยมี status: "Succeeded" แต่ไม่มีรหัสทรัพยากรในเนื้อหา (สําเร็จ แก้ไขโดยการแสดงรายการและการจับคู่ displayName)

เมื่อต้องการจัดการกับสิ่งเหล่านี้อย่างสม่ําเสมอ ให้สร้างฟังก์ชันตัวช่วยเดียวที่:

1. ส่งคืนรหัสทรัพยากรทันทีหากการตอบกลับเริ่มต้นมีอยู่แล้ว
2. มิฉะนั้นจะสํารวจ URL การดําเนินการ (สร้างจาก Location or x-ms-operation-id) โดยใช้ Retry-After.
3. ปฏิบัติ 200 OK ต่อ status: "Running" / "NotStarted" การที่ยังอยู่ในระหว่างดําเนินการและดําเนินการสํารวจความคิดเห็นต่อไป
4. เมื่อสําเร็จ จะส่งคืนรหัสทรัพยากรจากเนื้อหา หรือย้อนกลับไปแสดงรายการทรัพยากรและการจับคู่ตาม displayName (ด้วยการลองใหม่) เมื่อเนื้อหาเป็นสถานะเท่านั้น
5. ใช้ _pbi_headers() เมื่อ URL การสํารวจอยู่บนคลัสเตอร์ Power BI (*.analysis.windows.net) และส่วนหัว Fabric

ตัวช่วยตัวเดียวนี้แทนที่ความต้องการตัวช่วย "แก้ไขตามชื่อ" ต่อทรัพยากร — ทุกcreate_*ฟังก์ชันในบทช่วยสอนนี้เรียก_handle_lroด้วยตัวlist_urlmatch_display_nameช่วยและ .

เพิ่มสิ่งต่อไปนี้หลัง FabricClient ชั้นเรียน

# =========================================================
# LRO handler 
# =========================================================

def _handle_lro(
    client: httpx.Client,
    initial_response: httpx.Response,
    *,
    list_url: str | None = None,
    match_display_name: str | None = None,
    id_field: str = "id",
    max_attempts: int = 10,
    delay: int = 5,
) -> str:
    """
    Handle a Fabric long-running operation (LRO) and return the resource id.

    Supports the response patterns used by Fabric REST APIs:
    - 200/201 with the resource body inline (synchronous).
    - 202 with a `Location` header or `x-ms-operation-id` (asynchronous).
    - 200 with `status: "Running"` / `"NotStarted"` while polling.
    - 200 with `status: "Succeeded"` but no id (resolve by listing and matching `displayName`).

    Polling uses `Retry-After` and switches to a Power BI audience token when
    the operation URL is on `*.analysis.windows.net`.
    """
    # Sync 200/201 with body: return the id immediately.
    if initial_response.status_code in (200, 201):
        try:
            body = initial_response.json() if initial_response.content else {}
        except ValueError:
            body = {}
        if isinstance(body, dict) and body.get(id_field):
            return body[id_field]

    # Location header, with x-ms-operation-id fallback.
    op_url = initial_response.headers.get("Location")
    if not op_url:
        op_id = initial_response.headers.get("x-ms-operation-id")
        if op_id:
            op_url = f"https://api.fabric.microsoft.com/v1/operations/{op_id}"
        else:
            raise RuntimeError(
                f"Missing LRO Location/x-ms-operation-id. "
                f"status={initial_response.status_code} body={initial_response.text[:500]!r}"
            )

    # Audience-aware polling: Power BI cluster endpoints need a different token.
    poll_headers = _pbi_headers() if "analysis.windows.net" in op_url else _fabric_headers()
    retry_after = int(initial_response.headers.get("Retry-After", "5"))

    while True:
        time.sleep(retry_after)
        poll = client.get(op_url, headers=poll_headers)

        if poll.status_code == 202:
            retry_after = int(poll.headers.get("Retry-After", "5"))
            continue

        poll.raise_for_status()
        body = poll.json() if poll.content else {}
        status = body.get("status") if isinstance(body, dict) else None

        if status in ("Running", "NotStarted"):
            retry_after = int(poll.headers.get("Retry-After", "5"))
            continue
        if status == "Failed":
            raise RuntimeError(f"LRO failed. Body: {body}")

        if isinstance(body, dict) and body.get(id_field):
            return body[id_field]

        # Status-only success: list and match by displayName, with retries.
        if status == "Succeeded" and list_url and match_display_name:
            for attempt in range(max_attempts):
                r = client.get(list_url, headers=_fabric_headers())
                r.raise_for_status()
                match = next(
                    (i for i in r.json().get("value", []) if i.get("displayName") == match_display_name),
                    None,
                )
                if match and match.get(id_field):
                    return match[id_field]
                time.sleep(delay)
            raise RuntimeError(
                f"LRO succeeded but resource not visible after retries. "
                f"match_display_name={match_display_name!r}"
            )

        raise RuntimeError(f"LRO completed but no resource id was returned. Body: {body}")

ตัวช่วยเพย์โหลดคําจํากัดความ

เมื่อคุณสร้างแผนที่ด้วยคําจํากัดความสาธารณะ API สร้างแผนที่ REST API คาดหวังให้แต่ละส่วนมีdefinition.partsเพย์โหลดที่เข้ารหัส base64 ด้วย"payloadType": "InlineBase64" ตัวช่วย _json_to_b64 เข้ารหัส Python dict (map.json) ของคุณในรูปแบบนั้น เพื่อให้ create_map สามารถวางลงในเนื้อหาคําขอได้โดยตรง

เพิ่มสิ่งต่อไปนี้หลัง _handle_lro ฟังก์ชัน:

# =========================================================
# Definition payload helper
#
# Encodes map.json as base64 for inline Create map payloads.
# =========================================================

def _json_to_b64(obj: dict) -> str:
    """
    Convert a Python dict to base64-encoded JSON text.

    Fabric Map "Create Map with definition inline" requires:
    - definition.parts[].payloadType = InlineBase64
    - definition.parts[].payload     = base64(json(map_json))
    """
    return base64.b64encode(json.dumps(obj).encode("utf-8")).decode("utf-8")

สร้างตัวช่วยเพื่อจัดเตรียม eventstream สายอักขระการเชื่อมต่อ

หากต้องการส่งเหตุการณ์ไปยังตําแหน่งข้อมูลที่กําหนดเองของสตรีมเหตุการณ์ สคริปต์ต้องมี สายอักขระการเชื่อมต่อ สําหรับตําแหน่งข้อมูลนั้น

ซึ่งแตกต่างจาก REST API Fabric ที่คุณเรียกใช้จนถึงตอนนี้ (ซึ่งเป็นการดําเนินการระนาบควบคุมสําหรับการสร้างและจัดการทรัพยากร) การนําเข้า eventstream ใช้ ตําแหน่งข้อมูลระนาบข้อมูลที่เข้ากันได้กับฮับเหตุการณ์ และปลายทางนั้นรับรองความถูกต้องด้วย สายอักขระการเชื่อมต่อ ที่ใช้ SAS แทนที่จะเป็นโทเค็น Microsoft Entra สายอักขระการเชื่อมต่อ ถูกสร้างขึ้นเมื่อคุณเพิ่มแหล่งที่มาของปลายทางแบบกําหนดเอง และไม่ถูกเปิดเผยโดย Fabric REST API ดังนั้นจึงต้องคัดลอกจากพอร์ทัล Fabric

get_eventhub_connection_string_interactive ใช้ค่าจาก EVENTHUB_CONNECTION_STRING ตัวแปรสภาพแวดล้อมซ้ํา (สะดวกในการเรียกใช้ซ้ํา) หรือแจ้งให้คุณใช้ค่านั้นในขณะรันไทม์ จากนั้นแคชไว้เพื่อให้ cfg ขั้นตอนต่อมาสามารถนํากลับมาใช้ใหม่ได้โดยไม่ต้องแจ้งอีกครั้ง

เพิ่มสิ่งต่อไปนี้หลัง _json_to_b64 ฟังก์ชัน:

def get_eventhub_connection_string_interactive(cfg: Config) -> str:
    """
    Prompt for (or read) the eventstream custom endpoint connection string.

    The connection string is created when the custom endpoint source is added
    to the eventstream and isn't exposed by the Fabric REST API, so we read it
    from the `EVENTHUB_CONNECTION_STRING` environment variable when set, or
    prompt interactively otherwise. The value is cached on `cfg` for reuse.
    """
    if getattr(cfg, "eventhub_connection_string", None):
        return cfg.eventhub_connection_string

    print("\n=== Eventstream connection string required ===")
    print("In the Fabric portal:")
    print("  1) Open the eventstream you just created")
    print("  2) Select the custom endpoint source")
    print("  3) Select SAS Key Authentication")
    print("  4) Copy Connection string-primary key\n")

    cfg.eventhub_connection_string = input("Paste connection string here: ").strip()

    if not cfg.eventhub_connection_string:
        raise RuntimeError("Connection string cannot be empty.")

    return cfg.eventhub_connection_string

สร้างตัวช่วยเพื่อเพาะเหตุการณ์เริ่มต้นจาก CSV

เพื่อให้แน่ใจว่าแผนที่จะแสดงข้อมูลทันทีหลังจากสร้างแล้ว สคริปต์จะส่ง เหตุการณ์เริ่มต้น ชุดเล็กๆ ไปยังสตรีมเหตุการณ์ก่อนที่จะสร้างแผนที่

หากไม่มีขั้นตอนนี้ ตาราง Eventhouse อาจยังไม่มีข้อมูลใดๆ และแผนที่อาจว่างเปล่าในการโหลดครั้งแรก

ฟังก์ชันตัวช่วยนี้จะอ่านข้อมูลจากไฟล์ CSV ในเครื่องและส่งแต่ละแถวเป็นเหตุการณ์ JSON ไปยังสตรีมเหตุการณ์โดยใช้โปรโตคอล EventHub

เนื่องจากทรัพยากร Eventstream ได้รับการจัดเตรียมแบบอะซิงโครนัส ตําแหน่งข้อมูลแบบกําหนดเองอาจไม่พร้อมที่จะยอมรับเหตุการณ์ทันทีหลังจากสร้าง ในการจัดการกับปัญหานี้ ฟังก์ชันตัวช่วยจะมีตรรกะการลองใหม่ในตัวที่พยายามส่งเหตุการณ์โดยอัตโนมัติจนกว่าตําแหน่งข้อมูลจะพร้อมใช้งาน สิ่งนี้ทําให้มั่นใจได้ว่ากระบวนการเริ่มต้นมีความน่าเชื่อถือและทําซ้ําได้ และไม่จําเป็นต้องปรับเวลาด้วยตนเอง

วิธีการนี้สะท้อนรูปแบบการนําเข้าในโลกแห่งความเป็นจริง:

• ข้อมูลถูกสร้างขึ้นจากภายนอก (เช่น อุปกรณ์หรือแอปพลิเคชัน IoT)
• เหตุการณ์จะถูกสตรีมไปยัง Eventstream
• Eventstream ส่งข้อมูลไปยัง Eventhouse เพื่อการสืบค้นและการแสดงภาพ

การเพาะเหตุการณ์เริ่มต้นจะจําลองขั้นตอนการนําเข้านี้และตรวจสอบให้แน่ใจว่า:

• ตารางปลายทางถูกเติมข้อมูล
• ฟังก์ชัน KQL มีข้อมูลที่จะส่งคืน
• แผนที่จะแสดงผลทันทีหลังจากสร้าง

เพิ่มรหัสต่อไปนี้หลัง get_eventhub_connection_string_interactive() ฟังก์ชัน:

def seed_eventstream_from_csv(cfg: Config, max_attempts: int = 10, delay: int = 3) -> int:
    """
    Send seed events from a CSV with retries to handle eventstream readiness delay.
    """
    conn_str = get_eventhub_connection_string_interactive(cfg)

    last_error = None
    for attempt in range(1, max_attempts + 1):
        print(f"Seeding attempt {attempt}/{max_attempts}...")
        try:
            sent = 0
            producer = EventHubProducerClient.from_connection_string(conn_str=conn_str)
            try:
                with open(cfg.seed_csv_path, newline="", encoding="utf-8") as f:
                    reader = csv.DictReader(f)
                    batch = producer.create_batch()
                    batch_count = 0
                    for row in reader:
                        event = {
                            "VehicleId": row["VehicleId"],
                            "Latitude": float(row["Latitude"]),
                            "Longitude": float(row["Longitude"]),
                            "EventTime": row["EventTime"],
                        }
                        data = EventData(json.dumps(event))
                        try:
                            batch.add(data)
                            batch_count += 1
                        except ValueError:
                            producer.send_batch(batch)
                            sent += batch_count
                            batch = producer.create_batch()
                            batch.add(data)
                            batch_count = 1
                    if batch_count > 0:
                        producer.send_batch(batch)
                        sent += batch_count
                print(f"Seed events sent: {sent}")
                return sent
            finally:
                producer.close()
        except (EventHubError, ConnectionError, TimeoutError) as exc:
            last_error = exc
            print(f"Seeding failed (attempt {attempt}): {exc}. Retrying in {delay}s...")
            time.sleep(delay)

    raise RuntimeError(f"Seeding failed after {max_attempts} attempts. Last error: {last_error}")

รอความพร้อมใช้งานของฐานข้อมูล KQL

เมื่อมี eventhouse ฐานข้อมูล KQL และฟังก์ชัน KQL ฐานข้อมูล KQL อาจยังไม่สามารถแก้ไขได้ทันทีจากปลายทาง Fabric REST อื่น ๆ บริการ Fabric ทํางานบนแบ็กเอนด์แบบกระจาย ดังนั้นทรัพยากรที่สร้างขึ้นใหม่อาจใช้เวลาสั้น ๆ ในการเผยแพร่ไปทั่ว

หากคุณเรียก Create Map ทันทีหลังจากที่ฟังก์ชัน KQL อยู่ในตําแหน่ง Create Map อาจไม่สามารถแก้ไขแหล่งข้อมูลและส่งคืนข้อผิดพลาด เช่น ไม่พบฐานข้อมูล Kusto

wait_for_kql_database_ready สํารวจตําแหน่งข้อมูล Fabric REST สําหรับฐานข้อมูล KQL และส่งคืนทันทีที่ตอบสนอง 200 OK การตอบสนองที่ประสบความสําเร็จบนปลายทางระนาบควบคุมนี้เป็นพร็อกซีที่แข็งแกร่งซึ่ง Maps สามารถแก้ไขฐานข้อมูลได้เช่นกัน และจะเพิ่มขึ้น RuntimeError หลังจากนั้น max_attempts หากฐานข้อมูลไม่ปรากฏให้เห็น

เพิ่มสิ่งต่อไปนี้หลัง seed_eventstream_from_csv ฟังก์ชัน:

# =========================================================
# KQL database readiness helper
#
# Polls the KQL database's Fabric REST endpoint until it
# responds 200, as a best-effort gate before Create Map
# references it as a data source.
# =========================================================

def wait_for_kql_database_ready(
    client: httpx.Client,
    cfg: Config,
    kql_database_item_id: str,
    max_attempts: int = 10,
    delay: int = 3,
) -> None:
    """
    Poll the Fabric REST endpoint for a KQL database until it returns 200.

    Acts as a best-effort readiness gate before calling Create Map with the
    KQL database as a data source. Retries `max_attempts` times with `delay`
    seconds between attempts, then raises `RuntimeError` if the database
    never becomes visible.
    """
    url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    for attempt in range(1, max_attempts + 1):
        resp = client.get(url, headers=_fabric_headers())
        if resp.status_code == 200:
            print("KQL database is available to Fabric Maps")
            return
        print(
            f"Waiting for KQL database availability "
            f"(attempt {attempt}/{max_attempts}, status={resp.status_code})..."
        )
        time.sleep(delay)

    raise RuntimeError(
        f"KQL database {kql_database_item_id!r} did not become available after {max_attempts} attempts."
    )

สร้างฟังก์ชันหลัก

ถัดไป คุณเพิ่มฟังก์ชันหลักที่กําหนดเวิร์กโฟลว์ ทั้งหมดนี้เรียกว่า main().

ฟังก์ชันจะถูกเพิ่มตามลําดับที่กําหนดไว้ในโค้ด main() เรียกตารางในลําดับที่แตกต่างออกไปเล็กน้อย เพื่อให้ตาราง KQL มีอยู่ก่อนที่สตรีมเหตุการณ์จะผูกกับตารางนั้น และข้อมูลที่เริ่มต้นจึงพร้อมใช้งานก่อนการยืนยันจะทํางาน

• สร้างอีเวนต์เฮ้าส์
• สร้างตาราง KQL
• ยืนยันการนําเข้า (เรียกหลังจากการเพาะเมล็ด)
• สร้างเหตุการณ์สตรีม
• สร้างฟังก์ชัน KQL
• สร้างคําจํากัดความแผนที่ (map.json)
• สร้างข้อมูลเมตาของแพลตฟอร์ม (.platform)
• สร้างแผนที่

สร้างอีเวนต์เฮ้าส์

create_eventhouse สร้าง eventhouse ในพื้นที่ทํางานและส่งคืน ID รายการ สร้าง Eventhouse REST API สามารถรับสายเดียวกันได้สามวิธี:

• 201 Created ด้วยรหัสของบ้านเหตุการณ์แบบอินไลน์ (ซิงโครนัส)
• 202 Accepted ด้วย URL การดําเนินการ LRO (อะซิงโครนัส)
• 409 Conflict with x-ms-public-api-error-code ตั้งค่าเป็น ItemDisplayNameNotAvailableYet (ชื่อก่อนหน้านี้ยังคงสงวนไว้ที่แบ็กเอนด์) หรือ ItemDisplayNameAlreadyInUse (Eventhouse ที่มีชื่อนั้นมีอยู่แล้วในพื้นที่ทํางาน)

เพื่อจัดการกับทั้งสามอย่างได้อย่างน่าเชื่อถือ create_eventhouse:

• ผู้รับ 201 มอบสิทธิ์และการ 202 ตอบสนองต่อ _handle_lroซึ่งครอบคลุมการเสร็จสมบูรณ์แบบซิงโครนัสและแบบอะซิงโครนัสอย่างสม่ําเสมออยู่แล้ว
• เกียรติยศRetry-Afterและการลองใหม่ (พยายามสูงสุดห้าครั้ง)ItemDisplayNameNotAvailableYet
• นําบ้านItemDisplayNameAlreadyInUseงานที่มีอยู่กลับมาใช้ใหม่โดยแสดงรายการบ้านจัดงานในพื้นที่ทํางานและจับคู่ตามdisplayName
• ถอยกลับไปใช้ชื่อที่แสดงที่ไม่ได้รับการคัดเลือก (ต่อท้ายด้วย UUID แบบสั้น) หากชื่อไม่พร้อมใช้งานหลังจากงบประมาณการลองใหม่หมดลง ดังนั้นสคริปต์จึงยังคงสามารถดําเนินการไปข้างหน้าได้

เพิ่มสิ่งต่อไปนี้หลัง wait_for_kql_database_ready ฟังก์ชัน:

# =========================================================
# Create an eventhouse
# =========================================================

def create_eventhouse(client: httpx.Client, fabric: FabricClient, cfg: Config) -> str:
    """
    Create an eventhouse in the workspace and return its item ID.

    Handles the three response patterns Create Eventhouse can return:
    - 201/202: delegate to `_handle_lro` (synchronous body or LRO completion).
    - 409 `ItemDisplayNameNotAvailableYet`: honor `Retry-After` and retry.
    - 409 `ItemDisplayNameAlreadyInUse`: list eventhouses and reuse the one
      whose `displayName` matches `cfg.eventhouse_display_name`.

    If the name remains unavailable after the retry budget, falls back to a
    uniquified display name (suffixed with a short UUID) so the script can
    still make forward progress.
    """
    eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses"
    eventhouse_payload = {
        "displayName": cfg.eventhouse_display_name,
        "description": cfg.eventhouse_description
    }

    # Retry loop to handle transient "name not available yet"
    for attempt in range(1, 6):
        eh_resp = fabric.request("POST", eventhouse_url, json_body=eventhouse_payload)

        print("Create eventhouse status:", eh_resp.status_code)
        print("Create eventhouse headers:", dict(eh_resp.headers))

        # 201/202: success or LRO — _handle_lro handles both.
        if eh_resp.status_code in (201, 202):
            eventhouse_id = _handle_lro(
                client, eh_resp,
                list_url=eventhouse_url,
                match_display_name=cfg.eventhouse_display_name,
            )
            print("Eventhouse created. Eventhouse ID:", eventhouse_id)
            return eventhouse_id

        # 409: name issues
        if eh_resp.status_code == 409:
            api_code = eh_resp.headers.get("x-ms-public-api-error-code")

            # Name reserved temporarily: wait and retry
            if api_code == "ItemDisplayNameNotAvailableYet":
                wait_s = int(eh_resp.headers.get("retry-after", "20"))
                print(f"Name not available yet (attempt {attempt}/5). Waiting {wait_s}s then retrying...")
                time.sleep(wait_s)
                continue

            # Name already exists: reuse existing eventhouse by displayName
            if api_code == "ItemDisplayNameAlreadyInUse":
                print(f"Eventhouse {cfg.eventhouse_display_name!r} already exists. Reusing it...")
                r = client.get(eventhouse_url, headers=_fabric_headers())
                r.raise_for_status()
                match = next(
                    (i for i in r.json().get("value", []) if i.get("displayName") == cfg.eventhouse_display_name),
                    None,
                )
                if match and match.get("id"):
                    return match["id"]
                raise RuntimeError(
                    f"Eventhouse {cfg.eventhouse_display_name!r} reported as existing but not found in list."
                )

        # Anything else: fail fast with details
        raise RuntimeError(f"Create eventhouse failed: {eh_resp.status_code} {eh_resp.text}")

    # If the name never becomes available, last-resort: pick a unique name and try once
    cfg.eventhouse_display_name = f"{cfg.eventhouse_display_name}-{uuid.uuid4().hex[:8]}"
    print(f"Name still not available; switching to unique name: {cfg.eventhouse_display_name}")
    eh_resp = fabric.request("POST", eventhouse_url, json_body={
        "displayName": cfg.eventhouse_display_name,
        "description": cfg.eventhouse_description
    })
    eh_resp.raise_for_status()
    return eh_resp.json()["id"]

สร้างตาราง KQL

create_kql_table_if_missing ตรวจสอบให้แน่ใจว่าตารางปลายทางมีอยู่ในฐานข้อมูล KQL ก่อนที่สตรีมเหตุการณ์จะเริ่มเขียนลงไป ปลายทางสตรีมเหตุการณ์ที่คุณสร้างในภายหลังได้รับการกําหนดค่าด้วย ProcessedIngestion และ tableNameคงที่ ดังนั้นตารางจะต้องมีอยู่แล้วเมื่อเหตุการณ์มาถึง มิฉะนั้น การนําเข้าจะล้มเหลว

ฟังก์ชันจะออก.create-merge tableคําสั่งกับตําแหน่งข้อมูลการจัดการ Kusto ของอีเวนต์เฮาส์ (queryServiceUri + /v1/rest/mgmt) .create-merge เป็น idempotent: มันสร้างตารางหากไม่มีอยู่จริง และรวมสคีมาหากมีอยู่ ทําให้ปลอดภัยในการเรียกใช้ทุกครั้ง

ก่อนที่จะออกคําสั่ง ฟังก์ชันจะอ่านคุณสมบัติ eventhouse เพื่อรับ queryServiceUri และ ID รายการฐานข้อมูล KQL จากนั้นแก้ไขฐานข้อมูล displayName เพื่อให้ mgmt เพย์โหลดอ้างอิงตามชื่อแทนที่จะเป็น ID

เพิ่มสิ่งต่อไปนี้หลัง create_eventhouse ฟังก์ชัน:

# =========================================================
# Create the KQL table (idempotent)
# =========================================================

def create_kql_table_if_missing(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str) -> None:
    """
    Create or merge the destination table in the eventhouse's KQL database.

    Reads the eventhouse properties to discover `queryServiceUri` and the KQL
    database item ID, resolves the database's `displayName`, then issues a
    `.create-merge table` command against the Kusto management endpoint.
    `.create-merge` is idempotent: it creates the table if missing and merges
    the schema if it already exists.
    """
    # Get queryServiceUri + KQL database item id
    get_eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}"
    eh = fabric.request("GET", get_eventhouse_url)
    eh.raise_for_status()

    props = eh.json().get("properties") or {}
    query_service_uri = props.get("queryServiceUri")
    databases_item_ids = props.get("databasesItemIds") or []
    if not query_service_uri or not databases_item_ids:
        raise RuntimeError("Eventhouse missing queryServiceUri or databasesItemIds")

    kql_database_item_id = databases_item_ids[0]

    # Resolve actual DB displayName (don't rely on cfg.kql_database_name)
    get_db_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    db_resp = fabric.request("GET", get_db_url)
    db_resp.raise_for_status()
    kql_database_name = db_resp.json().get("displayName")
    if not kql_database_name:
        raise RuntimeError("KQL database response did not include displayName")

    # Create (or merge) the table schema
    # (Schema matches what your CSV sends: VehicleId, Latitude, Longitude, EventTime)
    csl = f""".create-merge table {cfg.eventhouse_table_name} (
        VehicleId: string,
        Latitude: real,
        Longitude: real,
        EventTime: datetime
    )"""

    mgmt_url = f"{query_service_uri}/v1/rest/mgmt"
    mgmt_payload = {"db": kql_database_name, "csl": csl}
    resp = client.post(mgmt_url, headers=_kusto_headers(), json=mgmt_payload)
    if resp.status_code >= 400:
        raise RuntimeError(f"Create table failed: {resp.status_code}\n{resp.text}")

    print(f"Ensured table exists: {cfg.eventhouse_table_name}")

ยืนยันการนําเข้าข้อมูล

verify_eventhouse_data ยืนยันว่าเหตุการณ์ที่ฝังอยู่ในสตรีมเหตุการณ์ได้ลงจอดในตาราง Eventhouse จริงๆ โดยจะสํารวจ <table> | count การสืบค้นกับจุดสิ้นสุด การสืบค้น Kusto ของ eventhouse (queryServiceUri + /v1/rest/query) จนกว่าจํานวนที่ส่งคืนจะมากกว่าศูนย์ หรือล้มเหลวหลังจากหมดเวลา การนําเข้า Eventstream ใช้เวลาสองสามวินาทีในการไหลจากจุดสิ้นสุดแบบกําหนดเองไปยังตาราง ดังนั้นการสํารวจ — แทนที่จะเป็นการสืบค้นเดียว — จึงเป็นสิ่งที่ให้สัญญาณผ่าน/ไม่ผ่านอย่างมั่นใจ

มีการกําหนดไว้ถัดจาก create_kql_table_if_missing เนื่องจากตัวช่วยทั้งสองค้นหาคุณสมบัติ eventhouse เดียวกัน (queryServiceUri, databasesItemIds) และแก้ไข displayName. ระบบจะเรียกmain()จากหลังseed_eventstream_from_csv เพื่อให้เหตุการณ์ที่เพาะเมล็ดมีโอกาสไหลผ่านสตรีมเหตุการณ์และไปถึงตารางก่อนที่การนับจะดําเนินไป

การเรียกใช้การตรวจสอบนี้ล่วงหน้าจะตรวจจับการกําหนดค่าการนําเข้าที่ไม่ถูกต้องตั้งแต่เนิ่นๆ ตัวอย่างเช่น ปลายทาง eventstream ที่ต่อสายไปยังชื่อตารางที่ไม่ถูกต้อง แทนที่จะปล่อยให้ปรากฏในภายหลังเป็นแผนที่ว่างเปล่า

เพิ่มสิ่งต่อไปนี้หลัง create_kql_table_if_missing ฟังก์ชัน:

# =========================================================
# Verify data ingestion (called after seeding)
# =========================================================

def verify_eventhouse_data(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str):
    """
    Poll a count query against the eventhouse table until rows arrive.

    Reads the eventhouse properties to get `queryServiceUri` and the KQL
    database item ID, resolves the database's `displayName`, then polls
    `<table> | count` against the Kusto query endpoint
    (`queryServiceUri` + `/v1/rest/query`) until the count is greater than
    zero or the timeout elapses. Eventstream ingestion is asynchronous, so
    polling avoids a false negative when the query runs before seeded
    events have landed in the table.
    """

    # Reuse your existing pattern to get KQL DB info
    eh = fabric.request("GET", f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}")
    eh.raise_for_status()

    props = eh.json().get("properties") or {}
    db_ids = props.get("databasesItemIds") or []
    query_service_uri = props.get("queryServiceUri")

    if not db_ids or not query_service_uri:
        raise RuntimeError("Missing eventhouse properties for verification")

    db_id = db_ids[0]

    db_resp = fabric.request("GET", f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{db_id}")
    db_resp.raise_for_status()

    db_name = db_resp.json().get("displayName")

    # Simple count query
    csl = f"{cfg.eventhouse_table_name} | count"
    query_url = f"{query_service_uri}/v1/rest/query"

    max_attempts = 12
    delay_seconds = 5

    for attempt in range(1, max_attempts + 1):
        resp = client.post(
            query_url,
            headers=_kusto_headers(),
            json={"db": db_name, "csl": csl}
        )

        if resp.status_code >= 400:
            raise RuntimeError(f"Verification query failed: {resp.text}")

        # Kusto v1 query response: Tables[0].Rows[0][0] holds the count.
        count = resp.json()["Tables"][0]["Rows"][0][0]
        print(f"Data verification attempt {attempt}/{max_attempts}: count = {count}")

        if count > 0:
            print(f"Data ingestion verified: {count} row(s) in {cfg.eventhouse_table_name}")
            return

        if attempt < max_attempts:
            time.sleep(delay_seconds)

    raise RuntimeError(
        f"Data verification failed: no rows in {cfg.eventhouse_table_name} after "
        f"{max_attempts * delay_seconds}s"
    )

สร้างสตรีมเหตุการณ์ด้วยคําจํากัดความ

create_eventstream_with_definition สร้าง eventstream ในพื้นที่ทํางานที่มีโทโพโลยีแบบเต็มรวมอยู่ในคําขอ จากนั้นส่งคืน ID รายการของ eventstream การใช้คําจํากัดความสาธารณะช่วยให้คุณสามารถจัดเตรียมสตรีมเหตุการณ์และเชื่อมต่อแหล่งที่มา สตรีม และปลายทางในการเรียกครั้งเดียว แทนที่จะสร้างสตรีมเหตุการณ์ก่อนแล้วจึงแก้ไขคําจํากัดความ

ก่อนส่งคําขอ ฟังก์ชันจะอ่านคุณสมบัติของ eventhouse เพื่อรับ ID รายการฐานข้อมูล KQL และแก้ไขฐานข้อมูล displayName เพื่อให้ปลายทางอ้างอิงตามชื่อแทนที่จะเป็น ID จากนั้นจะสร้างกราฟ eventstream ที่มี CustomEndpoint แหล่งที่มา a DefaultStreamและ Eventhouse ปลายทางที่กําหนดค่าด้วย ProcessedIngestion และ fixed tableName from cfg, base64-เข้ารหัสกราฟเป็น eventstream.json ส่วน และ POSTs เพื่อสร้าง Eventstream

สร้าง Eventstream REST API สามารถตอบด้วย 201 Created (ซิงโครนัส เนื้อหาแบบอินไลน์) 202 Accepted (LRO แบบอะซิงโครนัสผ่าน Location หรือ x-ms-operation-id) หรือ 200 OK ด้วยเพย์โหลดการเสร็จสมบูรณ์เฉพาะสถานะที่สตรีมเหตุการณ์ยังไม่ปรากฏในการตอบสนอง List Eventstreams เนื่องจากความล่าช้าในการเผยแพร่แบ็กเอนด์ _handle_lro ครอบคลุมกรณีเหล่านี้ทั้งหมด รวมถึงการแสดงรายการและการจับคู่ตาม displayName ดังนั้นฟังก์ชันนี้จึงมอบหมายการจัดการการตอบกลับแบบเต็มในการเรียกครั้งเดียว

เพิ่มสิ่งต่อไปนี้หลัง verify_eventhouse_data ฟังก์ชัน:

# =========================================================
# Create eventstream with definition
# =========================================================

def create_eventstream_with_definition(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str) -> str:
    """
    Create an eventstream with a public definition and return its item ID.

    Reads the eventhouse properties to discover the KQL database item ID and
    resolves the database's `displayName`, then builds an eventstream graph
    with a `CustomEndpoint` source, a `DefaultStream`, and an `Eventhouse`
    destination configured with `ProcessedIngestion` and the table name from
    `cfg`. Base64-encodes the graph as the `eventstream.json` part, POSTs it
    to Create Eventstream, and delegates response handling to `_handle_lro`.
    """
    eventstream_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventstreams"

    source_name = "CustomEndpointSource"
    stream_name = "DefaultStream"
    destination_name = "EventhouseDestination"

    # Resolve the KQL database item ID from the Eventhouse
    get_eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}"
    eh = fabric.request("GET", get_eventhouse_url)
    eh.raise_for_status()

    props = (eh.json().get("properties") or {})
    databases_item_ids = props.get("databasesItemIds") or []
    if not databases_item_ids:
        raise RuntimeError("Eventhouse properties did not include databasesItemIds.")

    kql_database_item_id = databases_item_ids[0]

    # Resolve the actual KQL database *name* (displayName) to avoid name drift
    get_db_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    db_resp = fabric.request("GET", get_db_url)
    db_resp.raise_for_status()
    kql_database_name = db_resp.json().get("displayName")
    if not kql_database_name:
        raise RuntimeError("KQL database response did not include displayName.")

    print("Eventhouse ID:", eventhouse_id)
    print("KQL database item ID:", kql_database_item_id)
    print("KQL database name:", kql_database_name)

    eventstream_json = {
        "sources": [
            {
                "name": source_name,
                "type": "CustomEndpoint",
                "properties": {
                    "inputSerialization": {"type": "Json", "properties": {"encoding": "UTF8"}}
                }
            }
        ],
        "streams": [
            {
                "name": stream_name,
                "type": "DefaultStream",
                "properties": {},
                "inputNodes": [{"name": source_name}]
            }
        ],
        "operators": [],
        "destinations": [
            {
                "name": destination_name,
                "type": "Eventhouse",
                "properties": {
                    "dataIngestionMode": "ProcessedIngestion",
                    "workspaceId": cfg.workspace_id,
                    "itemId": kql_database_item_id,
                    "databaseName": kql_database_name,
                    "tableName": cfg.eventhouse_table_name,
                    "inputSerialization": {"type": "Json", "properties": {"encoding": "UTF8"}}
                },
                "inputNodes": [{"name": stream_name}]
            }
        ],
        "compatibilityLevel": "1.1"
    }

    eventstream_payload = {
        "displayName": cfg.eventstream_display_name,
        "description": cfg.eventstream_description,
        "definition": {
            "parts": [
                {
                    "path": "eventstream.json",
                    "payload": _json_to_b64(eventstream_json),
                    "payloadType": "InlineBase64"
                }
            ]
        }
    }

    es_resp = fabric.request("POST", eventstream_url, json_body=eventstream_payload)

    eventstream_id = _handle_lro(
        client,
        es_resp,
        list_url=eventstream_url,
        match_display_name=cfg.eventstream_display_name,
    )

    print("Eventstream created. Eventstream ID:", eventstream_id)
    return eventstream_id

สร้างฟังก์ชัน KQL

create_kql_function สร้าง (หรืออัปเดต) ฟังก์ชัน Kusto ที่เก็บไว้ในฐานข้อมูล KQL ของบ้านเหตุการณ์ และส่งคืนรหัสรายการของฐานข้อมูล KQL เพื่อให้ผู้เรียกสามารถต่อสายแหล่งข้อมูลของแผนที่ไปยังมันได้ ฟังก์ชัน LatestVehicleLocations โดยค่าเริ่มต้น จะแสดงแถวล่าสุดต่อ VehicleId ผ่าน arg_max(EventTime, *) ฉายภาพ Latitude, Longitude, VehicleId และ EventTime เพื่อให้ Fabric Maps สามารถผูกคอลัมน์ละติจูดและลองจิจูดของเลเยอร์ได้

เช่นเดียวกับ create_kql_table_if_missingตัวช่วยนี้ทํางานกับปลายทางการจัดการ Kusto ของบ้านเหตุการณ์ (queryServiceUri + /v1/rest/mgmt) และเป็น idempotent: .create-or-alter function สร้างฟังก์ชันหากไม่มีอยู่จริง และแทนที่เนื้อหาหากมี ดังนั้นตัวช่วยจึงปลอดภัยในการเรียกใช้ทุกครั้ง

คําสั่งถูกส่งไปด้วย skipvalidation=true เนื่องจากเนื้อหาของฟังก์ชันอ้างอิงตารางปลายทางผ่านแทนที่ table("<name>") จะเป็นตัวระบุเปล่า table()ฟอร์มจะเลื่อนการแก้ไขชื่อเป็นเวลาคิวรี ดังนั้นการตรวจสอบความถูกต้องในเวลาที่สร้างจะล้มเหลวหากตารางยังไม่ได้รับข้อมูลใดๆ และผู้ตรวจสอบความถูกต้องไม่สามารถมองเห็น Schema ได้อย่างสมบูรณ์ การจับคู่ skipvalidation=true กับ table("...") ช่วยให้สามารถสร้างฟังก์ชันได้ก่อนที่จะมีการนําเข้าข้อมูลในตาราง ซึ่งเป็นลําดับที่บทช่วยสอนนี้ทํางาน

เพิ่มสิ่งต่อไปนี้หลัง create_eventstream_with_definition ฟังก์ชัน:

# =========================================================
# Create KQL function
# =========================================================

def create_kql_function(client: httpx.Client, fabric: FabricClient, cfg: Config, eventhouse_id: str) -> str:
    """
    Create or update the stored Kusto function used by the map layer.

    Reads the eventhouse properties to discover `queryServiceUri` and the
    KQL database item ID, resolves the database's `displayName`, then
    issues a `.create-or-alter function` command against the Kusto
    management endpoint with `skipvalidation=true` and a `table("...")`
    reference so the function can be created before the destination table
    has any data. Returns the KQL database item ID so the caller can wire
    the map's data source to it.
    """
    # Get eventhouse properties (queryServiceUri + databasesItemIds)
    get_eventhouse_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/eventhouses/{eventhouse_id}"
    eh = fabric.request("GET", get_eventhouse_url)
    eh.raise_for_status()

    props = (eh.json().get("properties") or {})
    query_service_uri = props.get("queryServiceUri")
    databases_item_ids = props.get("databasesItemIds") or []

    if not query_service_uri:
        raise RuntimeError("Eventhouse properties did not include queryServiceUri.")
    if not databases_item_ids:
        raise RuntimeError("Eventhouse properties did not include databasesItemIds.")

    # We'll return this so the caller can wire the map to the correct KQL database item id.
    kql_database_item_id = databases_item_ids[0]

    # Resolve actual KQL database name (displayName)
    get_db_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/kqlDatabases/{kql_database_item_id}"
    db_resp = fabric.request("GET", get_db_url)
    db_resp.raise_for_status()

    kql_database_name = db_resp.json().get("displayName")
    if not kql_database_name:
        raise RuntimeError("KQL database response did not include displayName.")

    # Create the function that returns the latest location per vehicle.
    # Keep columns explicit so the map config can bind Latitude/Longitude.
    kql = f""".create-or-alter function with (skipvalidation=true) {cfg.kql_function_name}() {{
    table("{cfg.eventhouse_table_name}")
    | summarize arg_max(EventTime, *) by VehicleId
    | project Latitude, Longitude, VehicleId, EventTime
    }}"""

    mgmt_url = f"{query_service_uri}/v1/rest/mgmt"
    mgmt_payload = {"db": kql_database_name, "csl": kql}

    mgmt_resp = client.post(mgmt_url, headers=_kusto_headers(), json=mgmt_payload)

    if mgmt_resp.status_code >= 400:
        # Kusto usually returns a detailed JSON error body on 400s.
        raise RuntimeError(
            "Kusto mgmt call failed.\n"
            f"URL: {mgmt_url}\n"
            f"DB: {mgmt_payload.get('db')}\n"
            f"Status: {mgmt_resp.status_code}\n"
            f"Body: {mgmt_resp.text}"
        )

    print("KQL function created/updated:", cfg.kql_function_name)
    return kql_database_item_id

สร้าง map.json

build_map_json สร้างและส่งคืนเพย์โหลด map.json ที่กําหนดเนื้อหาของ Fabric Map เพย์โหลดเป็นไปตามสคีมาคําจํากัดความของรายการแผนที่และประกอบด้วยสี่ส่วน: dataSources (ที่มาของข้อมูล) iconSources (เครื่องหมายที่กําหนดเองเพิ่มเติม) layerSources (สิ่งที่ถูกสืบค้นและความถี่) และ layerSettings (วิธีแสดงผลลัพธ์บนแผนที่)

สําหรับบทช่วยสอน dataSources นี้ชี้ไปที่ฐานข้อมูล KQL (itemType: "KqlDatabase") ที่สร้างขึ้นก่อนหน้านี้และรายการเดียวใน layerSources คือเลเยอร์ที่ได้รับการสนับสนุนจาก Kusto (type: "kusto", queryType: "function") ซึ่ง query เรียกฟังก์ชัน LatestVehicleLocations()ที่เก็บไว้ refreshIntervalMs อ่านจาก cfg.refresh_interval_ms (5,000 มิลลิวินาทีโดยค่าเริ่มต้น) ดังนั้นเลเยอร์จะเรียกใช้ฟังก์ชันอีกครั้งบนตัวจับเวลา และแผนที่จะสะท้อนการนําเข้าใหม่ในแบบเรียลไทม์

รายการที่ layerSettings ตรงกันจะผูกคอลัมน์ผลลัพธ์ของเลเยอร์กับแผนที่ผ่าน latitudeColumnName: "Latitude" ทาง และ longitudeColumnName: "Longitude"แสดงแต่ละแถวเป็น bubble จุด และพื้นผิว VehicleId และ EventTime ในคําแนะนําเครื่องมือ ฟังก์ชันจะพิมพ์เพย์โหลดที่ประกอบเข้าด้วยกัน เพื่อให้คุณสามารถตรวจสอบ JSON ที่แน่นอนที่การเรียก Create Map ส่ง

สําหรับข้อมูลเพิ่มเติมเกี่ยวกับข้อกําหนดแผนที่ REST API โปรดดู คําจํากัดความของรายการแผนที่

เพิ่มสิ่งต่อไปนี้หลัง create_kql_function ฟังก์ชัน:

# =========================================================
# Build map.json
# =========================================================

def build_map_json(cfg: Config, kql_database_item_id: str) -> dict:
    """
    Build and return the map.json payload for the Fabric Map.

    Wires `dataSources` to the KQL database created earlier, defines a
    single Kusto-backed layer in `layerSources` that calls the stored
    function `cfg.kql_function_name` and re-runs it every
    `cfg.refresh_interval_ms` milliseconds, and configures `layerSettings`
    to bind the `Latitude` / `Longitude` columns and render each row as a
    bubble point. Prints the assembled payload for inspection.
    """
    layer_source_id = str(uuid.uuid4())
    layer_setting_id = str(uuid.uuid4())
    data_source_name = "kqlConnection"

    map_json = {
        "$schema": "https://developer.microsoft.com/json-schemas/fabric/item/map/definition/2.0.0/schema.json",
        "basemap": {},

        "dataSources": [
            {
                "name": data_source_name,
                "itemType": "KqlDatabase",
                "workspaceId": cfg.workspace_id,
                "itemId": kql_database_item_id
            }
        ],

        "iconSources": [],

        "layerSources": [
            {
                "id": layer_source_id,
                "name": cfg.kql_function_name,
                "type": "kusto",
                "dataSourceName": data_source_name,
                "workspaceId": cfg.workspace_id,
                "itemId": kql_database_item_id,
                "refreshIntervalMs": cfg.refresh_interval_ms,
                "queryType": "function",
                "query": f"{cfg.kql_function_name}()"
            }
        ],
        "layerSettings": [
            {
                "id": layer_setting_id,
                "name": "Live Locations",
                "sourceId": layer_source_id,
                "options": {
                    "type": "vector",
                    "visible": True,
                    "pointLayerType": "bubble",
                    "tooltipKeys": ["VehicleId", "EventTime"],
                    "bubbleOptions": {
                        "color": "#0078D4"
                    }
                },
                "latitudeColumnName": "Latitude",
                "longitudeColumnName": "Longitude"

            }
        ]
    }

    
    print("Map definition (map.json):", json.dumps(map_json, indent=2))
    return map_json

สร้าง .platform (ข้อมูลเมตาของแพลตฟอร์ม)

build_platform_json สร้างและส่งคืนส่วนที่ไม่บังคับ .platform ที่การเรียกสร้างแผนที่สามารถรวมไว้ควบคู่ map.json ไปกับเมื่อคุณต้องการตั้งค่าข้อมูลเมตาของรายการที่ไม่ใช่ค่าเริ่มต้นบนแผนที่ ไม่จําเป็นต้องรวมส่วน .platform — Fabric ใช้ข้อมูลเมตาเริ่มต้นเมื่อละเว้นส่วนนั้น — แต่บทช่วยสอนนี้แสดงวิธีการเขียน เพื่อให้คุณสามารถใช้รูปแบบซ้ําได้เมื่อคุณต้องการการควบคุมอย่างชัดเจนเกี่ยวกับประเภทรายการ ชื่อที่แสดง คําอธิบาย หรือตัวระบุเชิงตรรกะที่เสถียร

เพย์โหลดเป็นไปตามสคีมาคุณสมบัติของแพลตฟอร์มและมีสองส่วน: (, , ) metadataและ type: "Map" (displayName, description) configversionlogicalId สร้างขึ้น logicalId เป็น UUID ใหม่ที่นี่ ซึ่งใช้ได้สําหรับการสร้างครั้งเดียวหากคุณวางแผนที่จะปรับใช้แผนที่เดิมใหม่ผ่านการรวม Git หรือการเรียกใช้ซ้ํา ให้ปักหมุด logicalId ไว้ที่ค่าที่เสถียรเพื่อให้การอัปเดตกําหนดเป้าหมายรายการเดียวกัน

สําหรับข้อมูลเพิ่มเติม ให้ดูที่ ข้อกําหนดของรายการแผนที่ และ ภาพรวมของข้อกําหนดรายการ

เพิ่มสิ่งต่อไปนี้หลัง build_map_json ฟังก์ชัน:

# =========================================================
# Build .platform (platform metadata)
# =========================================================

def build_platform_json(cfg: Config) -> dict:
    """
    Build and return the optional .platform payload for a Fabric Map item.

    The map definition supports an optional .platform part alongside
    map.json that carries non-default item metadata: the item type,
    display name and description, and a `logicalId` used for
    deterministic updates. Fabric applies defaults when the part is
    omitted, so this payload is only needed when you want explicit
    control over those fields. A fresh UUID is used for `logicalId`
    here; pin it to a stable value if repeat runs should target the
    same item.
    """
    return {
        "$schema": "https://developer.microsoft.com/json-schemas/fabric/gitIntegration/platformProperties/2.0.0/schema.json",
        "metadata": {
            "type": "Map",
            "displayName": cfg.map_display_name,
            "description": cfg.map_description
        },
        "config": {
            "version": "2.0",
            # Use a stable logicalId if you want deterministic updates; UUID is fine for create.
            "logicalId": str(uuid.uuid4())
        }
    }

สร้างแผนที่ด้วยคําจํากัดความแบบอินไลน์

create_map สร้างแผนที่โดยการลงรายการบัญชีข้อกําหนดแบบอินไลน์ที่คุณได้ประกอบไว้ และส่งคืนรหัสไอเท็มของแผนที่ใหม่ คําขอมีส่วนที่เข้ารหัส base64 สามส่วนภายใต้ payloadType: "InlineBase64": map.json (คําจํากัดความหลักที่จําเป็น) ข้อมูลเมตาเสริม .platform ที่คุณสร้างขึ้นในขั้นตอนก่อนหน้า และไฟล์คิวรี Kusto ที่มีชื่อ queries/layerSource-<layerSourceId>.kql ที่มีการเรียกฟังก์ชัน KQL ที่เก็บไว้ การรวมทั้งสามส่วนในการเรียกครั้งเดียวจะจัดเตรียมแผนที่และต่อเลเยอร์ข้อมูลไปยังฟังก์ชัน KQL แบบอะตอม ดังนั้นจึงไม่จําเป็นต้องติดตามผล getDefinition / updateDefinition ไปกลับ

ชื่อไฟล์คิวรีมีความสําคัญ: Fabric แก้ไขคิวรีของเลเยอร์โดยการจับคู่ queries/layerSource-<layerSourceId>.kql กับ id ของรายการที่สอดคล้องกันใน layerSources ดังนั้นฟังก์ชันจะดึงรหัสแหล่งที่มาของเลเยอร์ออกจาก map_json["layerSources"][0]["id"] เพื่อสร้างเส้นทาง map.json และ .platform เข้ารหัส base64 ผ่าน _json_to_b64;ข้อความแบบสอบถามถูกเข้ารหัส base64 โดยตรง เนื่องจากเป็นสตริงแทนที่จะเป็น dict.

สร้างแผนที่ REST API สามารถตอบด้วย 201 Created (ซิงโครนัส, ID แบบอินไลน์), 202 Accepted (LRO แบบอะซิงโครนัสผ่าน Location หรือ x-ms-operation-id) หรือ 200 OK ด้วยเพย์โหลดการเสร็จสมบูรณ์แบบสถานะเท่านั้นที่แผนที่ยังไม่ปรากฏใน List Maps เนื่องจากความล่าช้าในการเผยแพร่แบ็กเอนด์ _handle_lro ครอบคลุมกรณีเหล่านี้ทั้งหมด รวมถึงการแสดงรายการและการจับคู่ตาม displayName ดังนั้นฟังก์ชันนี้จึงมอบหมายการจัดการการตอบกลับแบบเต็มในการเรียกครั้งเดียว

สําหรับข้อมูลเพิ่มเติม โปรดดู ข้อกําหนดของรายการแผนที่

เพิ่มสิ่งต่อไปนี้หลัง build_platform_json ฟังก์ชัน:

# =========================================================
# Create a map with inline definition
# =========================================================


def create_map(client: httpx.Client, fabric: FabricClient, cfg: Config, map_json: dict, platform_json: dict) -> str:
    """
    Create the Fabric Map with its definition inline and return its item ID.

    Sends a single Create Map request whose `parts` array carries three
    base64-encoded payloads: `map.json` (the required core definition),
    the optional `.platform` metadata, and a Kusto query file named
    `queries/layerSource-<layerSourceId>.kql` whose `<layerSourceId>`
    matches `map_json["layerSources"][0]["id"]` so Fabric can bind the
    query to the layer. Delegates response handling to `_handle_lro`,
    which covers synchronous, asynchronous, and status-only completions.
    """

    create_map_url = f"https://api.fabric.microsoft.com/v1/workspaces/{cfg.workspace_id}/maps"

    # Extract the layer source id so we can name the query file correctly
    layer_source_id = map_json["layerSources"][0]["id"]

    # Kusto query content (bind to the stored function)
    query_text = f"{cfg.kql_function_name}()"
    query_b64 = base64.b64encode(query_text.encode("utf-8")).decode("utf-8")

    create_map_payload = {
        "displayName": cfg.map_display_name,
        "description": cfg.map_description,
        "definition": {
            "parts": [
                {
                    "path": "map.json",
                    "payload": _json_to_b64(map_json),
                    "payloadType": "InlineBase64"
                },
                {
                    "path": ".platform",
                    "payload": _json_to_b64(platform_json),
                    "payloadType": "InlineBase64"
                },
                {
                    # Kusto layer query file naming convention
                    "path": f"queries/layerSource-{layer_source_id}.kql",
                    "payload": query_b64,
                    "payloadType": "InlineBase64"
                }
            ]
        }
    }

    map_resp = fabric.request("POST", create_map_url, json_body=create_map_payload)

    return _handle_lro(
        client, map_resp,
        list_url=create_map_url,
        match_display_name=cfg.map_display_name,
    )

ประสานเวิร์กโฟลว์

main เป็นจุดเริ่มต้นเดียวที่เรียกใช้บทช่วยสอนแบบ end-to-end มันสร้างConfigอินสแตนซ์ , เปิดหนึ่งhttpx.Clientตัวที่ใช้ซ้ําในทุกตัวช่วย ห่อไว้ใน FabricClientจากนั้นเรียกใช้ฟังก์ชันแต่ละขั้นตอนตามลําดับการขึ้นต่อกัน: create_eventhouse → create_kql_table_if_missing (ต้องมีอยู่ก่อนที่กระแสเหตุการณ์จะผูกกับมัน) → create_eventstream_with_definition → seed_eventstream_from_csv → verify_eventhouse_data (จับการกําหนดค่าการนําเข้าผิดพลาดก่อนที่แผนที่จะทํางาน) → → create_kql_functionwait_for_kql_database_ready (ประตูที่พยายามอย่างเต็มที่เพื่อให้ Create Map สามารถแก้ไขฐานข้อมูล KQL ได้) → build_map_json → → build_platform_jsoncreate_map

การเรียงลําดับมีความสําคัญเนื่องจากขั้นตอนส่วนใหญ่ใช้บางสิ่งที่สร้างขึ้นโดยขั้นตอนก่อนหน้านี้ — create_eventstream_with_definition ต้องการ ของอีเวนต์เฮาส์ databasesItemIdsและ create_map ต้องการชื่อฟังก์ชัน KQL และรหัสรายการฐานข้อมูล KQL บล็อก print สุดท้ายจะแสดงรหัสของทรัพยากรทั้งหมดที่สร้างขึ้น เพื่อให้คุณสามารถค้นหาได้ในพอร์ทัล Fabric

เพิ่มสิ่งต่อไปนี้หลัง create_map ฟังก์ชัน:

# =========================================================
# main(): orchestrates the full workflow
# =========================================================

def main():
    """
    Orchestrate the tutorial workflow.

    1) Create eventhouse
    2) Create KQL table (required for ingestion)
    3) Create Eventstream (definition-based)
    4) Seed initial data so the map is not empty on first open
    5) Validate ingestion BEFORE moving on
    6) Create KQL function (required for Maps layer)
    7) Ensure KQL database is available to Maps
    8) Build map.json
    9) Build .platform metadata
    10) Create map with inline definition
    """
    cfg = Config()

    print("Initializing clients...")
    with httpx.Client(timeout=60) as client:
        fabric = FabricClient(client)

        # Step 1: Create eventhouse
        eventhouse_id = create_eventhouse(client, fabric, cfg)

        # Step 2: Ensure table exists BEFORE Eventstream binds to it
        create_kql_table_if_missing(client, fabric, cfg, eventhouse_id)

        # Step 3: Create Eventstream (definition-based)
        eventstream_id = create_eventstream_with_definition(client, fabric, cfg, eventhouse_id)

        # Step 4: Seed initial data so the map is not empty on first open
        seed_count = seed_eventstream_from_csv(cfg)

        # Step 5: Validate ingestion BEFORE moving on
        verify_eventhouse_data(client, fabric, cfg, eventhouse_id)

        # Step 6: Create KQL function (required for Maps layer)
        kql_database_item_id = create_kql_function(client, fabric, cfg, eventhouse_id)

        # Step 7: Ensure KQL database is available to Maps
        wait_for_kql_database_ready(client, cfg, kql_database_item_id)

        # Step 8: Build map.json (Kusto function layer)
        map_json = build_map_json(cfg, kql_database_item_id)

        # Step 9: Build .platform metadata
        platform_json = build_platform_json(cfg)

        # Step 10: Create map with inline definition
        map_id = create_map(client, fabric, cfg, map_json, platform_json)

        print("\nDONE")
        print("Eventhouse ID:", eventhouse_id)
        print("Eventstream ID:", eventstream_id)
        print(f"Seed events sent: {seed_count}")
        print("KQL database item ID:", kql_database_item_id)
        print("KQL function:", cfg.kql_function_name)
        print("Map ID:", map_id)

if __name__ == "__main__":
    main()

เรียกใช้แอปพลิเคชัน

ในระหว่างการเรียกใช้สคริปต์ คุณจะได้รับพร้อมท์ให้วางสตริง สายอักขระการเชื่อมต่อ eventstream

เมื่อต้องการดึงค่านี้:

1. เปิดพื้นที่ทํางาน Fabric ของคุณ
2. เปิดสตรีมเหตุการณ์ที่สร้างโดยสคริปต์
3. เลือก แหล่งที่มาปลายทางแบบกําหนดเอง
4. เปิด การรับรองความถูกต้องด้วยคีย์ SAS
5. คัดลอก คีย์หลักของสตริงการเชื่อมต่อ

วางค่าลงในคอนโซลเมื่อได้รับพร้อมท์

เรียกใช้สคริปต์:

python create_realtime_map.py

ตรวจสอบว่ารายการทั้งหมดถูกสร้างขึ้น:

ณ จุดนี้ ทรัพยากรทั้งหมดจะถูกสร้างและกําหนดค่า

หากต้องการจําลองการสตรีมต่อเนื่องและดูการอัปเดตแผนที่แบบเกือบเรียลไทม์ ให้ดําเนินการติดตามผล บทช่วยสอน: จําลองการนําเข้าข้อมูลแบบเรียลไทม์สําหรับแผนที่โดยใช้ REST API และ Python สร้างขึ้นโดยตรงจากบทช่วยสอนนี้และนํา eventhouse, eventstream, ฟังก์ชัน KQL และแผนที่ที่คุณสร้างขึ้นมาใช้ซ้ํา

Summary

ในบทช่วยสอนนี้ คุณได้เตรียมใช้งานทรัพยากรที่จําเป็นสําหรับโซลูชันเชิงพื้นที่แบบเรียลไทม์ใน Microsoft Fabric โดยใช้ Fabric REST API และ Python

คุณทําสิ่งต่อไปนี้สําเร็จ:

• สร้างฐานข้อมูล eventhouse และ KQL โดยใช้ Fabric REST API
• สร้าง สตรีมเหตุการณ์ที่มีตําแหน่งข้อมูลแบบกําหนดเอง สําหรับการนําเข้าเหตุการณ์การสตรีม
• กําหนด ฟังก์ชัน KQL เพื่อสืบค้นและจัดรูปร่างข้อมูลแบบเรียลไทม์สําหรับการแสดงภาพแผนที่
• สร้างและปรับใช้แผนที่ Fabric พร้อมคําจํากัดความแบบอินไลน์ ที่อ้างอิงข้อมูลบ้านเหตุการณ์
• เพาะสตรีมเหตุการณ์ด้วยเหตุการณ์เริ่มต้นเพื่อให้แผนที่แสดงข้อมูลทันที

สถาปัตยกรรมนี้แสดงให้เห็นถึงรูปแบบการวิเคราะห์แบบเรียลไทม์ทั่วไปใน Fabric:

• ผู้ผลิตภายนอกส่งเหตุการณ์ไปยัง Eventstream
• Eventstream กําหนดเส้นทางและนําเข้าข้อมูลไปยัง Eventhouse
• ฟังก์ชัน KQL แปลงข้อมูล
• แผนที่สืบค้น Eventhouse และรีเฟรชโดยอัตโนมัติเพื่อแสดงเหตุการณ์ใหม่

ด้วยการสร้างทรัพยากรโดยอัตโนมัติโดยใช้ Python และ REST API ตอนนี้คุณมีวิธีการที่ทําซ้ําได้สําหรับการสร้างแอปพลิเคชันเชิงพื้นที่แบบเรียลไทม์โดยไม่ต้องกําหนดค่าด้วยตนเอง หากต้องการขับเคลื่อนข้อมูลอย่างต่อเนื่องลงในแผนที่ ให้ดําเนินการต่อด้วยบทช่วยสอนการจําลองการติดตามผล

ขั้นตอนถัดไป

เมื่อคุณเข้าใจโฟลว์แบบ end-to-end แล้ว คุณสามารถขยายโซลูชันนี้เพื่อรวมตัวจําลองแบบเรียลไทม์ได้

สําหรับบทช่วยสอนที่สาธิตการสร้างเครื่องจําลองแบบเรียลไทม์สําหรับแผนที่ที่คุณเพิ่งสร้างขึ้นโดยใช้ REST API โปรดดู: