Tutorial: Eine Echtzeitkarte mit REST-APIs und Python erstellen

Fabric Karten können realzeit-Geospatialdaten visualisieren, indem sie eine Verbindung mit eventhouse-Datasets herstellen, die kontinuierlich über eventstream-Erfassung aktualisiert werden.

Im Gegensatz zu statischen Szenarien, in denen Dateien verwendet werden, die in einem Lakehouse gespeichert sind, veranschaulicht dieses Lernprogramm eine Streaming-, ereignisgesteuerte Architektur , in der:

• Ereignisse werden in ein Eventhouse eingespeist.
• Daten werden mithilfe der Kusto Query Language (KQL) abgefragt.
• Eine Karte wird dynamisch aktualisiert, wenn neue Daten eintreffen

Dieses Tutorial konzentriert sich auf die Automatisierung des End-to-End-Workflows mit Fabric-REST-APIs und Python, damit Sie Ressourcen programmgesteuert bereitstellen und eine Echtzeit-Kartenerfahrung konfigurieren können. Statische Datenszenarien mit Lakehouse-Dateien finden Sie unter Create a static map using REST APIs and Python.

In diesem Lernprogramm erfahren Sie, wie Sie eine Echtzeit-Geospatiallösung in Microsoft Fabric mithilfe von Eventstream, Eventhouse und KQL erstellen und automatisieren.

Um kontinuierliches Streaming zu simulieren und die Aktualisierung der Karte nahezu in Echtzeit zu beobachten, schließen Sie dieses Tutorial zunächst ab und fahren Sie dann mit dem anschließenden Tutorial: Simulieren Sie die Datenerfassung in Echtzeit für eine Karte mithilfe von REST-APIs und Python fort, das direkt auf dem Eventhouse, dem Eventstream, der KQL-Funktion und der Karte aufbaut, die Sie hier erstellen.

Szenarioübersicht: Nachverfolgung von Ressourcen in Echtzeit

Dieses Tutorial basiert auf einem Szenario zur Echtzeit-Anlagenverfolgung, ähnlich dem Flottenverfolgungsszenario, das im ursprünglichen Fabric Maps Tutorial: Erstellen von Echtzeit-Workorderrouting mit Fabric Maps verwendet wird.

Szenario:

• Fahrzeuge geben regelmäßig Standortupdates aus.
• Ortsereignisse werden in ein Eventhouse aufgenommen.
• Eine Karte zeigt die neuesten Fahrzeugpositionen und Updates automatisch an, wenn neue Ereignisse eintreffen

Dieses Muster ist repräsentativ für allgemeine Anwendungsfälle in Echtzeit, z. B.:

• Flottenverfolgung
• Disponieren von Arbeitsaufträgen
• Überwachung von Anlagen und Ausrüstung

Microsoft Fabric verwendet Eventstream und Eventhouse zum Aufnehmen, Verarbeiten und Analysieren von Streamingdaten in nahezu Echtzeit, sodass Live-Betriebsdaten direkt auf einer Karte dargestellt werden können.

Dieses Tutorial folgt einem gängigen Automatisierungsmuster in Fabric: Infrastruktur erstellen → Datenstrom erfassen → Erfassung überprüfen → Karte darstellen.

Voraussetzungen

• Python 3.10 oder höher
• Azure CLI
• Fabric-Arbeitsbereichs-ID
• Berechtigungen zum Aufrufen Fabric REST-APIs, z. B.:
  • Item.ReadWrite.All

Authentication

In diesem Tutorial wird DefaultAzureCredential verwendet, das sich mithilfe mehrerer lokaler bzw. Entwicklungsquellen für Anmeldeinformationen authentifizieren kann. Für Erstleser ist die einfachste Methode die Azure CLI-Anmeldung.

Lokal authentifizieren (empfohlen für die erste Ausführung)

1. Öffnen Sie ein Terminal.
2. Laufen:

az login

DefaultAzureCredential kann Ihre angemeldete Identität verwenden, um Zugriffstoken für Folgendes abzurufen:

• Fabric-REST-APIs (Ressource: https://api.fabric.microsoft.com/.default)
• Kusto (KQL)-Datenebenenabfragen für das Eventhouse (Ressource: https://api.kusto.windows.net/.default)
• Power BI / Fabric REST-Endpunkte zur Statusabfrage lang andauernder Vorgänge (Ressource: https://analysis.windows.net/powerbi/api/.default)

Anmelden bei Microsoft Fabric (empfohlen)

Bevor Sie dieses Lernprogramm ausführen, empfehlen wir, sich mindestens einmal bei Microsoft Fabric anzumelden:

https://app.fabric.microsoft.com

Durch die Anmeldung wird sichergestellt, dass Ihre Fabric-Identität, Arbeitsbereichsrollenmitgliedschaft und Kapazitätszuweisungen vollständig bereitgestellt werden, bevor Sie ein Microsoft Entra-Zugriffstoken programmgesteuert abrufen.

Dieser Schritt ist besonders hilfreich, wenn:

• Sie sind neu bei Microsoft Fabric
• Der Arbeitsbereich wurde kürzlich erstellt.
• Ihre Rollenzuweisung wurde kürzlich hinzugefügt.

Erstellen Sie die Seed-Datendatei (initiale Kartendaten)

Um sicherzustellen, dass die Karte unmittelbar nach der Bereitstellung Daten anzeigt, sendet das Skript eine kleine Menge an Initialereignissen an den Ereignisstrom.

1. Erstellen Sie eine neue Datei im selben Verzeichnis wie Ihr Python Skript: vehicle_locations_seed.csv
2. Fügen Sie den folgenden Inhalt ein:

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

Schritt 1 – Erstellen einer neuen Python-Projektdatei

In diesem Schritt erstellen Sie eine leere Python-Datei, die Sie Abschnitt für Abschnitt aufbauen.

Erstellen Sie eine neue Datei mit dem Namen:

create_realtime_map.py

Öffnen Sie die Datei in Ihrem Editor.

Schritt 2– Installieren erforderlicher Bibliotheken und Hinzufügen erforderlicher Importanweisungen

In diesem Schritt installieren Sie die Abhängigkeiten, und fügen Sie die Importe hinzu, die Ihr Skript verwendet.

Installieren erforderlicher Bibliotheken

Führen Sie im soeben geöffneten Terminalfenster den folgenden Befehl aus:

pip install httpx azure-identity azure-eventhub

Was für jede Bibliothek vorgesehen ist

• httpx: sendet HTTP-Anforderungen an die Fabric-REST-APIs.
• azure-identity: stellt DefaultAzureCredential für Microsoft Entra Authentifizierung bereit.
• azure-eventhub: sendet Seedereignisse an den Event Hub-kompatiblen Endpunkt des Eventstreams, um das Eventhouse aufzufüllen.

Hinzufügen von Importanweisungen zu Ihrer .py-Datei

Fügen Sie oben in create_realtime_map.py Folgendes hinzu:

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

Schritt 3 – Hinzufügen eines Konfigurationsabschnitts

In diesem Schritt definieren Sie die Variablen, die Ihre Anwendung verwendet, einschließlich der Arbeitsbereichs-ID und der Ressourcennamen.

Die Zentrale Konfiguration in einer einzelnen Config Klasse – anstatt hartcodierte Werte über Funktionen hinweg zu verteilen – bietet Ihnen drei konkrete Vorteile:

• Portabilität der Umgebung: Arbeitsbereichs-IDs, Ressourcennamen und andere Einstellungen befinden sich an einem Ort, sodass Sie das Skript für einen anderen Arbeitsbereich oder einen anderen Computer erneut ausführen können, indem Sie einige Zeilen (oder eine Umgebungsvariable) ändern, anstatt den Code zu durchsuchen.
• Übersichtlichere Funktionssignaturen: Schrittfunktionen akzeptieren ein einzelnes cfg Objekt anstelle langer Parameterlisten, wodurch die Orchestrierung main() leicht lesbar bleibt.
• Sicherere Handhabung geheimer Schlüssel: Vertrauliche Werte wie die Arbeitsbereichs-ID werden aus Umgebungsvariablen geladen, sodass sie niemals zusammen mit dem Skript zugesichert werden.

Fügen Sie folgendes unter den Importanweisungen hinzu:

# =========================================================
# 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", "")

Festlegen der Arbeitsbereichs-ID mithilfe einer Umgebungsvariable

Anstatt die Arbeitsbereichs-ID direkt im Skript zu codieren, liest dieses Lernprogramm sie aus einer Umgebungsvariable. Dadurch bleiben umgebungsspezifische Werte außerhalb des Quellcodes erhalten, und Sie können das Skript für Arbeitsbereiche oder Computer wiederverwenden, ohne es zu bearbeiten.

Erstellen Sie vor dem Ausführen des Skripts eine Umgebungsvariable mit dem Namen FABRIC_WORKSPACE_ID.

Festlegen der Umgebungsvariable für Windows

Unter Windows können Sie die Variable in jedem Terminal festlegen, das Umgebungsvariablen unterstützt – PowerShell, Windows PowerShell, die in Visual Studio und Visual Studio Code integrierten PowerShell- oder Eingabeaufforderungsfenster, Windows-Terminal und die meisten anderen Shells.

Führen Sie folgendes in PowerShell oder im integrierten VS Code-Terminal aus:

$env:FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

So bestätigen Sie, dass die Variable festgelegt ist:

echo $env:FABRIC_WORKSPACE_ID

Dadurch wird die Variable nur für die aktuelle Terminalsitzung festgelegt.

Eine persistente Umgebungsvariable festlegen (Windows)

Verwenden Sie eine der folgenden Optionen, um die Variable in zukünftigen Sitzungen verfügbar zu machen:

• PowerShell (Einzeiler): Ausführen: setx FABRIC_WORKSPACE_ID "<WORKSPACE_ID>". Der setx Befehl nimmt Änderungen an der Benutzerumgebung vor, aktualisiert jedoch nicht das aktuelle Terminal – schließen Sie das Terminal und öffnen Sie es erneut (oder öffnen Sie ein neues), bevor Sie das Skript ausführen.
• GUI:
  1. Öffnen Sie die Systemeigenschaften.
  2. Wählen Sie erweiterte Systemeinstellungen aus.
  3. Wählen Sie "Umgebungsvariablen" aus.
  4. Wählen Sie unter Benutzervariablen"Neu" aus.
  5. Eintreten:
     • Name: FABRIC_WORKSPACE_ID
     • Wert: Ihre Arbeitsbereichs-ID
  6. Wählen Sie "OK" aus, um sie zu speichern.
  7. Schließen Sie das Terminal, und öffnen Sie es erneut, bevor Sie das Skript erneut ausführen.

Festlegen der Umgebungsvariable unter macOS oder Linux

Unter macOS und Linux können Sie die Variable aus jeder Shell festlegen, die export unterstützt– Bash, Zsh (standard auf modernen macOS), Fish (mit einer etwas anderen Syntax) und die integrierten Terminals in Visual Studio Code und anderen Editoren.

Laufen:

export FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

So bestätigen Sie, dass die Variable festgelegt ist:

echo $FABRIC_WORKSPACE_ID

Dadurch wird die Variable nur für die aktuelle Shellsitzung festgelegt.

Eine persistente Umgebungsvariable festlegen (macOS oder Linux)

Um die Variable in zukünftigen Sitzungen verfügbar zu machen, fügen Sie die export Zeile zu Ihrem Shellprofil hinzu:

• Zsh (Standard unter macOS): ~/.zshrc
• Bash: ~/.bashrc (Linux) oder ~/.bash_profile (macOS)
• Fish: Ausführen set -Ux FABRIC_WORKSPACE_ID "<WORKSPACE_ID>" statt Bearbeiten einer Datei

Öffnen Sie nach dem Aktualisieren des Profils entweder ein neues Terminal oder führen Sie source ~/.zshrc aus (oder die entsprechende Datei), damit die Änderung wirksam wird.

Schritt 4 – Hinzufügen von Hilfsfunktionen

In diesem Schritt lagern Sie querschnittliche Belange – Authentifizierung, das Erstellen von Headern, das Polling lang andauernder Vorgänge und die Wiederholungslogik – in eine kleine Gruppe wiederverwendbarer Hilfsfunktionen aus, die von jeder Schrittfunktion aufgerufen werden können.

Diese Aspekte in Hilfsfunktionen zu zentralisieren – anstatt sie direkt an jeder Aufrufstelle einzubetten – bietet Ihnen drei konkrete Vorteile:

• Eine einzige Quelle der Wahrheit für grenzüberschreitende Bedenken: Authentifizierung, Header und LRO-Abrufe werden von fast jedem API-Aufruf benötigt. Ihre Zentralisierung sorgt dafür, dass sich jede einzelne Step Function auf ihre eigene Ressource konzentrieren kann, anstatt die Tokenbeschaffung und die Logik für Wiederholungsversuche erneut zu implementieren.
• Resilienz ohne Ballast: Hilfsfunktionen fangen vorübergehende Zustände ab – asynchrone Bereitstellung, Verzögerungen bei der Backend-Propagation, Sendefehler, die erneut versucht werden können – sodass Step Functions kurz bleiben und sich wie eine Checkliste lesen.
• Einfacher zu vermitteln und anzupassen: Jeder Helfer wird einmal definiert und wiederverwendet. Wenn Fabric ein LRO-Muster oder einen Autorisierungsbereich ändert, passen Sie es an einer einzigen Stelle an.

Die Helfer, die Sie in diesem Schritt hinzufügen, sind:

• Auth-Hilfsprogramme: Erstellen von Headern für Fabric REST-APIs (und Power BI Cluster-LRO-Endpunkte)
• FabricClient: einfacher Wrapper für konsistente API-Aufrufe
• LRO-Handler: zum Abfragen lang andauernder Vorgänge mit Location / x-ms-operation-id / Retry-After, einschließlich Antworten mit 200 und Running, Power BI-Clusterendpunkten und Abschlussnutzlasten, die nur aus einem Status bestehen (aufgelöst durch displayName)
• Hilfsfunktion für Definitions-Payload: map.json für Inline-Definitionen Base64-kodieren
• Eventstream-Verbindungsassistent: fordert zur Eingabe der Verbindungszeichenfolge für den benutzerdefinierten Endpunkt auf
• Seed-Hilfsprogramm: sendet erste Ereignisse mit Wiederholungslogik, um sicherzustellen, dass die Aufnahme erfolgreich ist.
• KQL-Datenbankbereitschaftshilfsprogramm: wartet, bis die KQL-Datenbank für Fabric Karten verfügbar ist.

Erstellen von Authentifizierungshilfsfunktionen

Jeder Fabric-REST-Aufruf in diesem Lernprogramm enthält im Header Authorization ein Microsoft Entra-Zugriffstoken (Bearer-Token). Anstatt Tokens ad hoc abzurufen, kapselt dieser Schritt DefaultAzureCredential in ein kleines TokenProvider und stellt für jede Endpunktfamilie, die das Skript aufruft, einen audienzspezifischen Header-Generator bereit.

Die Zentralisierung des Tokenabrufs und der Header-Erstellung in Hilfsfunktionen — anstatt Token an jeder Aufrufstelle abzurufen — bietet Ihnen drei konkrete Vorteile:

• Centralized credential: Ein einzelner DefaultAzureCredential wird in TokenProvider eingeschlossen und für jeden API-Aufruf wiederverwendet, sodass die Identitätsermittlung (Azure CLI, VS Code, verwaltete Identität usw.) einmal erfolgt.
• Zielgruppenbezogene Token: Fabric-, Kusto- und Power BI-Cluster-Endpunkte lehnen Token ab, die für die falsche Zielgruppe ausgestellt wurden. Ein separater Header-Generator pro Zielgruppe hält den richtigen Gültigkeitsbereich direkt neben der Aufrufstelle, sodass sofort ersichtlich ist, welchen Endpunkt jede Funktion anspricht.
• Neu bei jeder Anforderung: Header-Generatoren erstellen den Authorization Header bei Bedarf, anstatt das Token selbst zwischenspeichern zu müssen. Die zugrunde liegende Anmeldeinformation wird im Hintergrund automatisch erneuert, sodass sich die Aufrufstellen nie Gedanken über den Ablauf machen müssen.

In diesem Lernprogramm werden Fabric REST-APIs mithilfe delegierter Bereiche wie Item.ReadWrite.All aufgerufen.

Fügen Sie folgendes nach der Config Klasse hinzu:

# =========================================================
# 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"
    }

Erstellen Sie einen Fabric-Clientwrapper

Die meisten Fabric REST-Aufrufe in diesem Lernprogramm senden dieselben Authorization und Content-Type Header. Anstatt sie an jeder Aufrufstelle zu wiederholen, umschließt dieses Lernprogramm httpx.Client in einem kleinen FabricClient Wrapper, der die Header automatisch anfügt und gleichzeitig das rohe httpx.Response zurückgibt, sodass jeder Aufrufer Statuscodes prüfen kann (z. B. um 201 von 202 zu unterscheiden).

httpx.Client auf diese Weise zu umschließen – anstatt headers=_fabric_headers() bei jedem Aufruf zu übergeben – bringt Ihnen zwei konkrete Vorteile:

• Kopfzeilen an einer zentralen Stelle: Jede Anrufwebsite nimmt die neueste _fabric_headers() automatisch auf, sodass eine neue Anforderung nicht versehentlich ohne den Authorization Header gesendet werden kann.
• Statuscodes bleiben sichtbar: request() gibt das rohe httpx.Response statt decodiertem JSON zurück, sodass Aufrufstellen weiterhin anhand des Status (201 vs. 202) verzweigen und Header wie Location oder Retry-After für die LRO-Handhabung prüfen können.

Fügen Sie Folgendes nach den Authentifizierungshilfsfunktionen hinzu:

# =========================================================
# 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)

Erstellen einer LRO-Hilfsfunktion

Mehrere in diesem Tutorial verwendete Fabric-REST-APIs – z. B. Create Eventhouse, Create Eventstream und Create Map – unterstützen lang andauernde Vorgänge (LROs).

Diese APIs können Antworten in verschiedenen Mustern zurückgeben:

• 201 Created mit dem Ressourcentext inline (synchron)
• 202 Accepted mit einer Location Kopfzeile, die auf eine Vorgangsstatus-URL zeigt (asynchron)
• 202 Accepted mit einem x-ms-operation-id Header anstelle von Location (asynchrone, alternative Form)
• 200 OK mit status: "Running" oder status: "NotStarted" während der Abfrage (noch in Bearbeitung)
• 200 OK mit status: "Succeeded", aber keine Ressourcen-ID im Textkörper (erfolgreich; auflösen durch Auflisten und Abgleichen von displayName)

Um all diese konsistent zu verarbeiten, erstellen Sie eine einzige Hilfsfunktion, die:

1. Gibt die Ressourcen-ID sofort zurück, wenn die ursprüngliche Antwort sie bereits enthält.
2. Andernfalls wird die Operations-URL, die aus Location oder x-ms-operation-id erstellt wird, unter Verwendung von Retry-After abgefragt.
3. Behandelt 200 OK mit status: "Running" / "NotStarted" weiterhin als noch in Bearbeitung und setzt die Abfrage fort.
4. Bei Erfolg wird die Ressourcen-ID aus dem Antworttextkörper zurückgegeben, oder es wird auf das Auflisten von Ressourcen und den Abgleich anhand von displayName (mit Wiederholungsversuchen) zurückgegriffen, wenn der Antworttextkörper nur den Status enthält.
5. Verwendet _pbi_headers(), wenn sich die Abfrage-URL auf einem Power BI-Cluster (*.analysis.windows.net) befindet, andernfalls Fabric-Header.

Diese einzelne Hilfsfunktion ersetzt die Notwendigkeit von Hilfsfunktionen zur „Auflösung nach Namen“ für jede Ressource – jede Funktion create_* in diesem Tutorial ruft _handle_lro mit den jeweils passenden list_url und match_display_name auf.

Fügen Sie folgendes nach der FabricClient Klasse hinzu:

# =========================================================
# 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}")

Helfer für Definitions-Payload

Wenn Sie eine Karte mit einer öffentlichen Definition erstellen, erwartet die REST-API zum Erstellen von Karten, dass jeder Teil in definition.parts Base64-codierte Nutzdaten mit "payloadType": "InlineBase64" enthält. Das Hilfsprogramm _json_to_b64 kodiert ein Python-dict (Ihr map.json) in dieses Format, damit create_map es direkt in den Request-Body einfügen kann.

Fügen Sie folgendes nach der _handle_lro Funktion hinzu:

# =========================================================
# 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")

Hilfsprogramm zum Bereitstellen der Eventstream-Verbindungszeichenfolge erstellen

Zum Senden von Ereignissen an den benutzerdefinierten Endpunkt des Ereignisstreams benötigt das Skript ein Verbindungszeichenfolge für diesen Endpunkt.

Im Gegensatz zu den bisher aufgerufenen Fabric REST-APIs (die Steuerungsebenenvorgänge zum Erstellen und Verwalten von Ressourcen sind), verwendet die Ereignisstreamaufnahme einen Event Hubs-kompatiblen Datenebenenendpunkt und dieser Endpunkt authentifiziert sich mit einem SAS-basierten Verbindungszeichenfolge anstelle eines Microsoft Entra-Tokens. Die Verbindungszeichenfolge wird erstellt, wenn Sie die benutzerdefinierte Endpunktquelle hinzufügen, und ist nicht über die Fabric-REST-API verfügbar, sodass sie aus dem Fabric-Portal kopiert werden muss.

get_eventhub_connection_string_interactive verwendet entweder einen Wert aus der EVENTHUB_CONNECTION_STRING Umgebungsvariable (praktisch bei Wiederholungsläufen) oder fordert Sie zur Laufzeit zur Eingabe auf, und speichert ihn cfg dann zwischen, sodass spätere Schritte sie wiederverwenden können, ohne erneut aufzufordern.

Fügen Sie folgendes nach der _json_to_b64 Funktion hinzu:

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

Hilfsprogramm zum Initialisieren von Ereignissen aus einer CSV-Datei erstellen

Um sicherzustellen, dass die Karte Daten unmittelbar nach der Erstellung anzeigt, sendet das Skript vor dem Erstellen der Karte einen kleinen Satz von Seedereignissen in den Eventstream.

Ohne diesen Schritt enthält die Eventhouse-Tabelle möglicherweise noch keine Daten, und die Karte könnte beim ersten Laden leer angezeigt werden.

Diese Hilfsfunktion liest Daten aus einer lokalen CSV-Datei und sendet jede Zeile als JSON-Ereignis mithilfe des EventHub-Protokolls an den Eventstream.

Da Eventstream-Ressourcen asynchron bereitgestellt werden, ist der benutzerdefinierte Endpunkt nach der Erstellung möglicherweise nicht sofort bereit, Ereignisse zu akzeptieren. Um dies zu behandeln, enthält die Hilfsfunktion integrierte Wiederholungslogik, die automatisch versucht, Ereignisse zu senden, bis der Endpunkt verfügbar ist. Dadurch wird sichergestellt, dass der Initialisierungsprozess zuverlässig und wiederholbar ist und keine manuellen Zeitlichen Anpassungen erfordert.

Dieser Ansatz spiegelt reale Erfassungsmuster wider:

• Daten werden extern erstellt (z. B. IoT-Geräte oder -Anwendungen)
• Ereignisse werden in Eventstream gestreamt
• Eventstream liefert Daten für Abfragen und Visualisierungen an Eventhouse

Durch das Seeding von Anfangsereignissen simulieren Sie diesen Aufnahmefluss und stellen Folgendes sicher:

• Die Zieltabelle wird mit Daten befüllt.
• Die KQL-Funktion enthält Daten, die zurückgegeben werden sollen.
• Die Karte wird unmittelbar nach der Erstellung gerendert.

Fügen Sie nach der get_eventhub_connection_string_interactive() Funktion den folgenden Code hinzu:

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}")

Warten auf die Verfügbarkeit der KQL-Datenbank

Sobald das Eventhouse, seine KQL-Datenbank und die KQL-Funktion vorhanden sind, kann die KQL-Datenbank möglicherweise nicht sofort aus anderen Fabric REST-Endpunkten aufgelöst werden. Fabric-Dienste werden auf verteilten Backends ausgeführt, sodass eine neu erstellte Ressource kurze Zeit benötigen kann, bis sie auf allen verfügbar ist.

Wenn Sie „Create Map“ unmittelbar nachdem die KQL-Funktion eingerichtet wurde aufrufen, kann „Create Map“ die Datenquelle möglicherweise nicht auflösen und einen Fehler wie Kusto-Datenbank nicht gefunden zurückgeben.

wait_for_kql_database_ready ruft den Fabric REST-Endpunkt für die KQL-Datenbank ab und gibt zurück, sobald er 200 OK antwortet. Es ist eine Best-Effort-Prüfung – eine erfolgreiche Antwort an diesem Control-Plane-Endpunkt ist ein starkes Indiz dafür, dass Maps die Datenbank ebenfalls erreichen kann –, und sie löst nach RuntimeErrormax_attempts aus, wenn die Datenbank nie sichtbar wird.

Fügen Sie folgendes nach der seed_eventstream_from_csv Funktion hinzu:

# =========================================================
# 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."
    )

Erstellen primärer Funktionen

Als Nächstes fügen Sie die primären Funktionen hinzu, die den Workflow definieren. Diese werden alle von main() aufgerufen.

Die Funktionen werden in der Reihenfolge hinzugefügt, in der sie im Code definiert sind. main() ruft sie in einer etwas anderen Reihenfolge auf, sodass die KQL-Tabelle vorhanden ist, bevor der Ereignisstream an ihn gebunden wird, und damit Seeded-Daten vor dem Ausführen der Überprüfung verfügbar sind.

• Ein Eventhouse erstellen
• Erstellen der KQL-Tabelle
• Erfassung überprüfen (wird nach dem Seeding aufgerufen)
• Erstellen eines Eventstreams
• Erstellen einer KQL-Funktion
• Erstellen der Kartendefinition (map.json)
• Erstellen der Plattformmetadaten (.platform)
• Erstellen der Karte

Ein Eventhouse erstellen

create_eventhouse erstellt ein Eventhouse im Arbeitsbereich und gibt dessen Element-ID zurück. Die Create Eventhouse-REST-API kann denselben Aufruf auf drei verschiedene Arten annehmen:

• 201 Created mit der Eventhouse-ID inline (synchron).
• 202 Accepted mit einer URL für den LRO-Vorgang (asynchron).
• 409 Conflict mit x-ms-public-api-error-code, das entweder auf ItemDisplayNameNotAvailableYet gesetzt ist (der vorherige Name ist im Backend weiterhin reserviert) oder auf ItemDisplayNameAlreadyInUse (ein Eventhouse mit diesem Namen ist im Arbeitsbereich bereits vorhanden).

Um alle drei zuverlässig zu verarbeiten, : create_eventhouse

• Delegiert 201- und 202-Antworten an _handle_lro, das bereits synchronen und asynchronen Abschluss einheitlich abdeckt.
• Beachtet Retry-After und Wiederholungsversuche (bis zu fünf Versuche) bei ItemDisplayNameNotAvailableYet.
• Verwendet das vorhandene Eventhouse auf ItemDisplayNameAlreadyInUse erneut, indem die Eventhouses im Arbeitsbereich aufgelistet und anhand von displayName abgeglichen werden.
• Greift auf einen eindeutig gemachten Anzeigenamen (mit einer kurzen UUID als Suffix) zurück, wenn der Name auch nach Ausschöpfen des Wiederholungsbudgets nie verfügbar wird, sodass das Skript weiterhin Fortschritte machen kann.

Fügen Sie folgendes nach der wait_for_kql_database_ready Funktion hinzu:

# =========================================================
# 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"]

Erstellen einer KQL-Tabelle

create_kql_table_if_missing stellt sicher, dass die Zieltabelle in der KQL-Datenbank vorhanden ist, bevor der Ereignisstream beginnt, in sie zu schreiben. Das Eventstream-Ziel, das Sie später erstellen, ist mit ProcessedIngestion und einer festen tableName konfiguriert, sodass die Tabelle bereits vorhanden sein muss, wenn Ereignisse ankommen — andernfalls schlägt die Erfassung fehl.

Die Funktion gibt einen .create-merge table Befehl für den Kusto-Verwaltungsendpunkt des Ereignishauses aus (queryServiceUri + /v1/rest/mgmt). .create-merge ist idempotent: Sie erstellt die Tabelle, wenn sie nicht vorhanden ist, und führt das Schema zusammen, wenn dies der Fall ist. Dadurch kann es bei jeder Ausführung sicher aufgerufen werden.

Bevor Sie den Befehl ausgeben, liest die Funktion die Eventhouse-Eigenschaften, um queryServiceUri und die Element-ID der KQL-Datenbank abzurufen, und löst dann das displayName der Datenbank auf, sodass die mgmt Payload sie anhand ihres Namens statt anhand ihrer ID referenziert.

Fügen Sie folgendes nach der create_eventhouse Funktion hinzu:

# =========================================================
# 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}")

Überprüfen der Datenerfassung

verify_eventhouse_data bestätigt, dass Ereignisse, die in den Eventstream eingespeist wurden, tatsächlich in der Eventhouse-Tabelle gelandet sind. Es abruft eine <table> | count Abfrage für den Kusto-Abfrageendpunkt des Ereignishauses (queryServiceUri + /v1/rest/query), bis die zurückgegebene Anzahl größer als 0 ist oder nach einem Timeout fehlschlägt. Es dauert einige Sekunden, bis die Eventstream-Erfassung vom benutzerdefinierten Endpunkt in der Tabelle ankommt. Daher liefert Polling – statt einer einzelnen Abfrage – ein zuverlässiges Bestanden-/Nicht-bestanden-Signal.

Es wird neben create_kql_table_if_missing definiert, da beide Hilfsfunktionen dieselben Eventhouse-Eigenschaften (queryServiceUri, databasesItemIds) ermitteln und das displayName der KQL-Datenbank auflösen. Es wird von main()afterseed_eventstream_from_csv aus aufgerufen, damit die initialisierten Ereignisse den Eventstream durchlaufen und die Tabelle erreichen können, bevor die Zählung erfolgt.

Wenn Sie diese Überprüfung vorab ausführen, werden Fehlkonfigurationen bei der Datenaufnahme frühzeitig erkannt – zum Beispiel ein Eventstream-Ziel, das mit dem falschen Tabellennamen verknüpft ist – statt dass sich der Fehler erst später als leere Zuordnung zeigt.

Fügen Sie folgendes nach der create_kql_table_if_missing Funktion hinzu:

# =========================================================
# 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"
    )

Erstellen eines Eventstreams mit Definition

create_eventstream_with_definition erstellt einen Ereignisstream im Arbeitsbereich mit seiner vollständigen Topologie, die in die Anforderung integriert ist, und gibt dann die Element-ID des Ereignisstreams zurück. Mithilfe einer öffentlichen Definition können Sie den Ereignisstream bereitstellen und seine Quellen, Datenströme und Ziele in einem einzigen Aufruf verbinden, anstatt zuerst den Ereignisstream zu erstellen und dann die Definition zu patchen.

Bevor die Anfrage gesendet wird, liest die Funktion die Eventhouse-Eigenschaften aus, um die Element-ID der KQL-Datenbank abzurufen, und löst displayName der Datenbank auf, sodass das Ziel auf sie anhand ihres Namens statt ihrer ID verweist. Anschließend wird ein Eventstream-Graph mit einer CustomEndpoint-Quelle, einem DefaultStream und einem Eventhouse-Ziel erstellt, das mit ProcessedIngestion und dem festen tableName aus cfg konfiguriert ist, der Graph wird als eventstream.json-Teil base64-kodiert und per POST an Create Eventstream gesendet.

Die Create Eventstream-REST-API kann mit 201 Created antworten (synchron, mit Inline-Textkörper), mit 202 Accepted (asynchron, als LRO über Location oder x-ms-operation-id) oder mit 200 OK mit einer Abschlussnutzlast, die nur den Status enthält, wobei der Eventstream aufgrund einer Verzögerung bei der Backend-Propagation in der Antwort von „List Eventstreams“ noch nicht sichtbar ist. _handle_lro deckt all diese Fälle ab – einschließlich Auflistung und Abgleich nach displayName –, sodass diese Funktion die vollständige Antwortverarbeitung in einem einzigen Aufruf delegiert.

Fügen Sie folgendes nach der verify_eventhouse_data Funktion hinzu:

# =========================================================
# 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-Funktion erstellen

create_kql_function erstellt (oder aktualisiert) eine gespeicherte Kusto-Funktion in der KQL-Datenbank des Ereignishauses und gibt die Element-ID der KQL-Datenbank zurück, damit der Aufrufer die Datenquelle der Karte mit ihr verknüpfen kann. Die Funktion – standardmäßig LatestVehicleLocations – gibt über VehicleId die neueste Zeile pro arg_max(EventTime, *) zurück, wobei Latitude, Longitude, VehicleId und EventTime projiziert werden, sodass Fabric Maps die Latitude- und Longitude-Spalten der Ebene binden kann.

Wie create_kql_table_if_missing arbeitet dieses Hilfsprogramm mit dem Kusto-Verwaltungsendpunkt des Eventhouse (queryServiceUri + /v1/rest/mgmt) und ist idempotent: .create-or-alter function erstellt die Funktion, wenn sie nicht vorhanden ist, und ersetzt ihren Funktionskörper, falls sie bereits vorhanden ist, sodass das Hilfsprogramm bei jeder Ausführung sicher aufgerufen werden kann.

Der Befehl wird mit skipvalidation=true gesendet, da der Funktionskörper die Zieltabelle über table("<name>") referenziert und nicht als bloßen Bezeichner. Die table() Form verschiebt die Namensauflösung auf den Zeitpunkt der Abfrage, sodass die Validierung zum Erstellungszeitpunkt andernfalls fehlschlagen würde, wenn die Tabelle noch keine Daten enthält und ihr Schema für den Validator noch nicht vollständig sichtbar ist. Das Koppeln von skipvalidation=true mit table("...") ermöglicht es, die Funktion zu erstellen, bevor die Erfassung die Tabelle mit Daten gefüllt hat; in dieser Reihenfolge wird in diesem Tutorial vorgegangen.

Fügen Sie folgendes nach der create_eventstream_with_definition Funktion hinzu:

# =========================================================
# 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 erstellen

build_map_json erstellt und gibt die map.json Payload zurück, die den Inhalt der Fabric Map definiert. Die Nutzlast folgt dem Schema der Kartenelementdefinition und besteht aus vier Abschnitten: dataSources (wo Daten stammen), iconSources (optionale benutzerdefinierte Markierungen), layerSources (was abgefragt wird und wie oft) und layerSettings (wie das Ergebnis auf der Karte gerendert wird).

Für dieses Tutorial verweist dataSources auf die zuvor erstellte KQL-Datenbank (itemType: "KqlDatabase"), und der einzelne Eintrag in layerSources ist eine Kusto-gestützte Ebene (type: "kusto", queryType: "function"), deren query die gespeicherte Funktion LatestVehicleLocations() aufruft. refreshIntervalMs wird aus cfg.refresh_interval_ms gelesen (standardmäßig alle 5000 ms), sodass der Layer die Funktion in einem Timer-Intervall erneut ausführt und die Karte neu eingehende Daten nahezu in Echtzeit abbildet.

Der entsprechende layerSettings-Eintrag bindet die Ergebnisspalten der Ebene über latitudeColumnName: "Latitude" und longitudeColumnName: "Longitude" an die Karte, rendert jede Zeile als einen bubble-Punkt und zeigt VehicleId und EventTime in Tooltips an. Die Funktion druckt die zusammengesetzte Nutzlast, sodass Sie den genauen JSON-Code untersuchen können, den der Create Map-Aufruf sendet.

Weitere Informationen zur REST-API der Kartendefinition finden Sie unter Kartenelementdefinition.

Fügen Sie folgendes nach der create_kql_function Funktion hinzu:

# =========================================================
# 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

Build.platform (Plattformmetadaten)

build_platform_json erstellt und gibt einen optionalen .platform-Teil zurück, den der Aufruf „Create Map“ zusammen mit map.json einschließen kann, wenn Sie vom Standard abweichende Elementmetadaten für die Map festlegen möchten. Das Einschließen eines .platform-Webparts ist nicht erforderlich – Fabric wendet Standardmetadaten an, wenn der Teil weggelassen wird. In diesem Lernprogramm wird jedoch gezeigt, wie Sie eins erstellen, damit Sie das Muster wiederverwenden können, wenn Sie explizite Kontrolle über den Elementtyp, den Anzeigenamen, die Beschreibung oder einen stabilen logischen Bezeichner benötigen.

Die Nutzlast folgt dem Schema der Plattformeigenschaften und weist zwei Abschnitte auf: metadata (type: "Map", , displayNamedescription) und config (version, logicalId). Die logicalId wird hier als neue UUID generiert, was für eine Einmalige Erstellung in Ordnung ist. Wenn Sie beabsichtigen, die gleiche Karte über die Git-Integration oder wiederholte Ausführung erneut bereitzustellen, heften logicalId Sie an einen stabilen Wert an, sodass updates auf dasselbe Element abzielen.

Weitere Informationen finden Sie unter Zuordnungselementdefinition und Übersicht über Elementdefinitionen.

Fügen Sie folgendes nach der build_map_json Funktion hinzu:

# =========================================================
# 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())
        }
    }

Eine Karte mit Inline-Definition erstellen

create_map erstellt die Karte durch POSTing der von Ihnen zusammengestellten Inlinedefinition und gibt die Element-ID der neuen Karte zurück. Die Anforderung enthält drei base64-codierte Teile unter payloadType: "InlineBase64": map.json (die erforderliche Kerndefinition), die optionalen .platform Metadaten, die Sie im vorherigen Schritt erstellt haben, und eine Kusto-Abfragedatei mit dem Namen queries/layerSource-<layerSourceId>.kql , die den Aufruf der gespeicherten KQL-Funktion enthält. Durch das Bündeln aller drei Teile in einem einzigen Aufruf wird die Karte provisioniert und ihre Datenebene atomar mit der KQL-Funktion verbunden, sodass kein zusätzlicher getDefinition / updateDefinition Roundtrip erforderlich ist.

Der Name der Abfragedatei ist wichtig: Fabric löst die Abfrage einer Ebene durch Abgleichen von queries/layerSource-<layerSourceId>.kql mit dem id des entsprechenden Eintrags in layerSources auf, sodass die Funktion die Layerquell-ID aus map_json["layerSources"][0]["id"] abruft, um den Pfad zu erstellen. map.json und .platform werden über _json_to_b64 Base64-kodiert; der Abfragetext wird direkt Base64-kodiert, da er eine Zeichenfolge und kein dict ist.

Die Create Map-REST-API kann mit 201 Created (synchron, mit Inline-ID), 202 Accepted (asynchroner LRO über Location oder x-ms-operation-id) oder 200 OK mit einer nur Status enthaltenden Abschluss-Payload antworten, bei der die Karte wegen einer backendseitigen Propagationsverzögerung in „List Maps“ noch nicht sichtbar ist. _handle_lro deckt all diese Fälle ab – einschließlich Auflistung und Abgleich nach displayName –, sodass diese Funktion die vollständige Antwortverarbeitung in einem einzigen Aufruf delegiert.

Weitere Informationen finden Sie unter Definition des Kartenelements.

Fügen Sie folgendes nach der build_platform_json Funktion hinzu:

# =========================================================
# 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,
    )

Koordinieren des Workflows

main ist der zentrale Einstiegspunkt, der das Tutorial vollständig ausführt. Es instanziiert Config, öffnet ein httpx.Client, das für alle Hilfsfunktionen wiederverwendet wird, kapselt es in FabricClient und ruft dann jede Schrittfunktion in der Reihenfolge ihrer Abhängigkeiten auf: create_eventhouse → create_kql_table_if_missing (muss vorhanden sein, bevor der Eventstream daran gebunden wird) → create_eventstream_with_definition → seed_eventstream_from_csv → verify_eventhouse_data (fängt eine Fehlkonfiguration der Datenerfassung ab, bevor irgendeine Map-Arbeit erfolgt) → create_kql_function → wait_for_kql_database_ready (eine Best-Effort-Prüfung, damit Create Map die KQL-Datenbank auflösen kann) → build_map_json → build_platform_json → create_map.

Die Reihenfolge ist wichtig, da die meisten Schritte etwas benötigen, das in einem vorherigen Schritt erstellt wurde – create_eventstream_with_definition benötigt das databasesItemIds des Eventhouse, und create_map benötigt den Namen der KQL-Funktion und die Element-ID der KQL-Datenbank. Mit dem letzten print-Block werden die IDs jeder erstellten Ressource angezeigt, sodass Sie sie im Fabric-Portal finden können.

Fügen Sie folgendes nach der create_map Funktion hinzu:

# =========================================================
# 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()

Ausführen der Anwendung

Während der Skriptausführung werden Sie aufgefordert, die Ereignisstream-Verbindungszeichenfolge einzufügen.

So rufen Sie diesen Wert ab:

1. Öffnen des Fabric-Arbeitsbereichs
2. Öffnen des vom Skript erstellten Eventstreams
3. Auswählen der benutzerdefinierten Endpunktquelle
4. Öffnen der SAS-Schlüsselauthentifizierung
5. Verbindungszeichenfolgen-Primärschlüssel kopieren

Fügen Sie den Wert in die Konsole ein, wenn Sie dazu aufgefordert werden.

Führen Sie das Skript aus:

python create_realtime_map.py

Überprüfen Sie, ob alle Elemente erstellt wurden:

Zu diesem Zeitpunkt werden alle Ressourcen erstellt und konfiguriert.

Um kontinuierliches Streaming zu simulieren und das Kartenupdate in Echtzeit zu überwachen, fahren Sie mit der Nachverfolgung Tutorial fort: Simulieren der Echtzeitdatenaufnahme für eine Karte mithilfe von REST-APIs und Python. Es baut direkt auf diesem Tutorial auf und verwendet das Eventhouse, den Eventstream, die KQL-Funktion und die Zuordnung wieder, die Sie erstellt haben.

Zusammenfassung

In diesem Lernprogramm haben Sie die Ressourcen bereitgestellt, die für eine Echtzeit-Geospatiallösung in Microsoft Fabric erforderlich sind, indem Sie Fabric REST-APIs und Python verwenden.

Sie haben Folgendes erreicht:

• Erstellt eine Eventhouse- und KQL-Datenbank mithilfe der Fabric REST-API
• Erstellt einen Eventstream mit einem benutzerdefinierten Endpunkt zum Aufnehmen von Streamingereignissen
• Definiert eine KQL-Funktion zum Abfragen und Aufbereiten von Echtzeitdaten für die Kartenvisualisierung
• Erstellt und bereitgestellt: eine Fabric-Zuordnung mit Inline-Definition, die auf Eventhouse-Daten verweist
• Den Event-Stream mit initialen Ereignissen befüllt, sodass die Karte sofort Daten anzeigte.

Diese Architektur veranschaulicht ein gängiges Echtzeitanalysemuster in Fabric:

• Externe Produzenten senden Ereignisse in Eventstream
• Eventstream leitet Daten an Eventhouse weiter und erfasst sie dort
• KQL-Funktionen transformieren die Daten
• Karten rufen Eventhouse ab und aktualisieren sich automatisch, um neue Ereignisse anzuzeigen.

Durch die Automatisierung der Ressourcenerstellung mithilfe von Python- und REST-APIs haben Sie jetzt einen wiederholbaren Ansatz zum Erstellen räumlicher Echtzeitanwendungen ohne manuelle Konfiguration. Um kontinuierlich Daten in die Karte einzuspeisen, fahren Sie mit dem anschließenden Simulator-Tutorial fort.

Nächste Schritte

Nachdem Sie nun den End-to-End-Fluss verstanden haben, können Sie diese Lösung erweitern, um einen Echtzeitsimulator zu integrieren.

Ein Lernprogramm, das das Erstellen eines Echtzeitsimulators für die soeben erstellte Karte mithilfe von REST-APIs veranschaulicht, finden Sie unter: