Esercitazione: Creare una mappa in tempo reale usando le API REST e Python

Fabric Maps può visualizzare dati geospaziali in tempo reale collegandosi a dataset di Eventhouse che vengono continuamente aggiornati tramite l'ingestione con Eventstream.

A differenza degli scenari statici che usano file archiviati in un Lakehouse, questa esercitazione illustra un'architettura basata su eventi in streaming in cui:

• Gli eventi vengono inseriti in una eventhouse
• I dati vengono sottoposti a query usando il linguaggio di query Kusto (KQL)
• Una mappa viene aggiornata dinamicamente man mano che arrivano nuovi dati

Questa esercitazione è incentrata su l'automazione del flusso di lavoro end-to-end utilizzando le API REST di Fabric e Python, in modo da poter effettuare il provisioning delle risorse e configurare tramite codice un'esperienza di mappa in tempo reale. Per scenari di dati statici che usano file Lakehouse, vedere Creare una mappa statica usando le API REST e Python.

Questa esercitazione illustra come creare e automatizzare una soluzione geospaziale in tempo reale in Microsoft Fabric usando Eventstream, Eventhouse e KQL.

Per simulare lo streaming continuo e osservare l'aggiornamento della mappa quasi in tempo reale, completare prima di tutto questa esercitazione, quindi continuare con il completamento Tutorial: Simulare l'inserimento di dati in tempo reale per una mappa usando le API REST e Python, che si basa direttamente sulla eventhouse, sulla funzione eventstream, sulla funzione KQL e sulla mappa creata qui.

Panoramica dello scenario: Rilevamento degli asset in tempo reale

Questa esercitazione si basa su uno scenario di tracciamento delle risorse in tempo reale, simile allo scenario di tracciamento della flotta utilizzato nell'esercitazione originale di Fabric Maps Esercitazione: creare un instradamento degli ordini di lavoro in tempo reale con Fabric Maps.

In questo scenario:

• I veicoli generano periodicamente aggiornamenti della posizione
• Gli eventi di localizzazione vengono acquisiti in un eventhouse
• Una mappa visualizza automaticamente le posizioni e gli aggiornamenti più recenti del veicolo man mano che arrivano nuovi eventi

Questo modello è rappresentativo dei casi d'uso comuni operativi in tempo reale, ad esempio:

• Tracciamento della flotta
• Invio dell'ordine di lavoro
• Monitoraggio delle risorse e delle apparecchiature

Microsoft Fabric usa Eventstream e Eventhouse per inserire, elaborare e analizzare i dati di streaming quasi in tempo reale, rendendo possibile visualizzare i dati operativi live direttamente su una mappa.

Questa esercitazione segue un modello di automazione comune in Fabric: creare l'infrastruttura → acquisire il flusso → convalidare l'acquisizione → visualizzare la mappa.

Prerequisiti

• Python 3.10 o versione successiva
• interfaccia della riga di comando di Azure
• ID spazio di lavoro tessuto
• Autorizzazioni per chiamare Fabric API REST, ad esempio:
  • Item.ReadWrite.All

Autenticazione

Questa esercitazione usa DefaultAzureCredential, che può autenticarsi usando diverse fonti di credenziali locali o di sviluppo. Per i lettori alla prima esperienza, l'approccio più semplice è l'accesso tramite interfaccia della riga di comando di Azure.

Eseguire l'autenticazione in locale (scelta consigliata per la prima esecuzione)

1. Aprire un terminale.
2. Corri!

az login

DefaultAzureCredential può utilizzare l'identità autenticata per acquisire token di accesso per:

• API REST Fabric (risorsa: https://api.fabric.microsoft.com/.default)
• Query del piano dati Kusto (KQL) sulla sede eventi (risorsa: https://api.kusto.windows.net/.default)
• Power BI/Fabric endpoint REST usati per il polling delle operazioni a esecuzione prolungata (risorsa: https://analysis.windows.net/powerbi/api/.default)

Accedere a Microsoft Fabric (scelta consigliata)

Prima di eseguire questa esercitazione, è consigliabile accedere almeno una volta a Microsoft Fabric:

https://app.fabric.microsoft.com

L'accesso garantisce che l'identità di Fabric, l'appartenenza ai ruoli dell'area di lavoro e le assegnazioni di capacità vengano sottoposte a provisioning completo prima di acquisire programmaticamente un token di accesso Microsoft Entra.

Questo passaggio è particolarmente utile se:

• Sei nuovo di Microsoft Fabric
• L'area di lavoro è stata creata di recente
• L'assegnazione di ruolo è stata aggiunta di recente

Creare il file di dati di inizializzazione (dati della mappa iniziale)

Per garantire che la mappa visualizzi i dati immediatamente dopo il provisioning, lo script invia un piccolo set di eventi di inizializzazione al flusso di eventi.

1. Creare un nuovo file nella stessa directory dello script di Python: vehicle_locations_seed.csv
2. Incollare il contenuto seguente:

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

Passaggio 1: Creare un nuovo file di progetto Python

In questo passaggio, si crea un file Python vuoto che viene completato sezione per sezione.

Creare un nuovo file denominato:

create_realtime_map.py

Aprire il file nell'editor.

Passaggio 2: Installare le librerie necessarie e aggiungere le istruzioni di importazione necessarie

In questo passaggio si installano le dipendenze e si aggiungono le importazioni usate dallo script.

Installare le librerie necessarie

Nella finestra del terminale appena aperta eseguire il comando seguente:

pip install httpx azure-identity azure-eventhub

A cosa serve ogni libreria

• httpx: effettua richieste HTTP alle API REST di Fabric.
• azure-identity: fornisce DefaultAzureCredential per l'autenticazione Microsoft Entra.
• azure-eventhub: invia gli eventi di inizializzazione all'endpoint compatibile con Hub eventi di Eventstream per popolare la eventhouse.

Aggiungere istruzioni import al file .py

Nella parte superiore di create_realtime_map.py aggiungere:

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

Passaggio 3: Aggiungere una sezione di configurazione

In questo passaggio si definiscono le variabili usate dall'applicazione, inclusi l'ID dell'area di lavoro e i nomi delle risorse.

La centralizzazione della configurazione in una singola Config classe, anziché la dispersione dei valori hardcoded tra le funzioni, offre tre vantaggi concreti:

• Portabilità dell'ambiente: ID dell'area di lavoro, nomi di risorse e altre impostazioni si trovano in un'unica posizione, in modo da poter rieseguire lo script in un'area di lavoro o in un computer diverso modificando alcune righe (o una variabile di ambiente) invece di cercare il codice.
• Signature di funzione più pulite: le funzioni di passaggio accettano un unico oggetto cfg invece di lunghi elenchi di parametri, il che rende l'orchestrazione in main() facile da leggere.
• Gestione dei segreti più sicura: i valori sensibili, ad esempio l'ID dell'area di lavoro, vengono caricati dalle variabili di ambiente, quindi non viene mai eseguito il commit insieme allo script.

Aggiungere quanto segue sotto le istruzioni import:

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

Impostare l'ID dell'area di lavoro usando una variabile di ambiente

Invece di codificare direttamente l'ID dell'area di lavoro nello script, questo tutorial lo legge da una variabile di ambiente. In questo modo i valori specifici dell'ambiente non sono disponibili nel codice sorgente e consentono di riutilizzare lo script tra aree di lavoro o computer senza modificarli.

Prima di eseguire lo script, creare una variabile di ambiente denominata FABRIC_WORKSPACE_ID.

Impostare la variabile di ambiente in Windows

In Windows, è possibile impostare la variabile in qualsiasi terminale che supporti le variabili d’ambiente: PowerShell, Windows PowerShell, le finestre PowerShell o del Prompt dei comandi integrate in Visual Studio e Visual Studio Code, Terminale Windows e la maggior parte delle altre shell.

Eseguire quanto segue in PowerShell o nel terminale integrato di VS Code:

$env:FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Per verificare che la variabile sia impostata:

echo $env:FABRIC_WORKSPACE_ID

In questo modo viene impostata solo la variabile per la sessione del terminale corrente.

Impostare una variabile di ambiente permanente (Windows)

Per rendere disponibile la variabile nelle sessioni future, usare una delle opzioni seguenti:

• PowerShell (one-liner): Esegui setx FABRIC_WORKSPACE_ID "<WORKSPACE_ID>". Il setx comando scrive nell'ambiente utente ma non aggiorna il terminale corrente, chiudere e riaprire il terminale (o aprirne uno nuovo) prima di eseguire lo script.
• GUI:
  1. Aprire Proprietà di sistema.
  2. Selezionare Impostazioni di sistema avanzate.
  3. Scegliere Variabili di ambiente.
  4. In Variabili utente selezionare Nuovo.
  5. Inserisci:
     • Nome: FABRIC_WORKSPACE_ID
     • Valore: ID della tua area di lavoro
  6. Selezionare OK per salvare.
  7. Chiudere e riaprire il terminale prima di eseguire di nuovo lo script.

Impostare la variabile di ambiente in macOS o Linux

In macOS e Linux è possibile impostare la variabile da qualsiasi shell che supporta export- Bash, Zsh (impostazione predefinita in macOS moderno), Fish (con una sintassi leggermente diversa) e i terminali integrati in Visual Studio Code e altri editor.

Corri!

export FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Per verificare che la variabile sia impostata:

echo $FABRIC_WORKSPACE_ID

In questo modo viene impostata la variabile solo per la sessione della shell corrente.

Impostare una variabile di ambiente permanente (macOS o Linux)

Per rendere disponibile la variabile nelle sessioni future, aggiungere la export riga al profilo della shell:

• Zsh (impostazione predefinita in macOS): ~/.zshrc
• Bash: ~/.bashrc (Linux) o ~/.bash_profile (macOS)
• Pesce: eseguire set -Ux FABRIC_WORKSPACE_ID "<WORKSPACE_ID>" invece di modificare un file

Dopo l'aggiornamento del profilo, aprire un nuovo terminale o eseguire source ~/.zshrc (o il file appropriato) in modo che la modifica venga applicata.

Passaggio 4: Aggiungere funzioni di supporto

In questo passaggio vengono evidenziate le problematiche trasversali, ovvero l'autenticazione, la costruzione dell'intestazione, il polling delle operazioni a esecuzione prolungata e la logica di ripetizione dei tentativi, in un piccolo set di helper riutilizzabili che ogni funzione passo può chiamare.

Centralizzare questi aspetti negli helper, anziché incorporarli direttamente in ogni punto di chiamata, offre tre vantaggi concreti:

• Unica fonte di riferimento per gli aspetti trasversali: l'autenticazione, le intestazioni e il polling LRO sono necessari per quasi ogni chiamata API. Centralizzarli consente a ogni funzione di step di concentrarsi sulle risorse di propria competenza invece di reimplementare l'acquisizione del token e la logica di ritentativo.
• Resilienza senza disordine: gli helper assorbono condizioni transitorie — provisioning asincrono, ritardi di propagazione del backend, errori di invio per cui è possibile riprovare — così le funzioni step restano brevi e si leggono come una checklist.
• Più facile da insegnare e modificare: ogni helper viene introdotto una volta e riutilizzato. Se Fabric modifica un modello LRO o un ambito di autenticazione, è possibile correggerlo in un'unica posizione.

Gli helper aggiunti in questo passaggio sono:

• Helper di autenticazione: creano intestazioni per le API REST di Fabric (e gli endpoint LRO del cluster di Power BI)
• FabricClient: wrapper leggero per chiamate API coerenti
• Gestore LRO: eseguire il polling delle operazioni a esecuzione prolungata usando Location / x-ms-operation-id / Retry-After, incluse le risposte 200-with-Running, gli endpoint del cluster Power BI e i payload di completamento con solo stato (risolto da displayName)
• Helper del payload di definizione: codifica in base64 map.json per definizioni inline
• Helper di connessione Eventstream: richiede la stringa di connessione dell'endpoint personalizzato
• Assistente di inizializzazione: invia eventi iniziali con un meccanismo di ritentativo per garantire che l'acquisizione vada a buon fine
• Assistente per la preparazione del database KQL: attende che il database KQL sia disponibile per Fabric Maps

Creare funzioni di supporto per l'autenticazione

Ogni chiamata REST di Fabric effettuata in questa esercitazione include un token di accesso Microsoft Entra (token bearer) nell'intestazione Authorization. Anziché ottenere i token in modo ad hoc, questo passaggio incapsula DefaultAzureCredential in un piccolo TokenProvider e rende disponibile un generatore di header specifico per l’audience per ogni famiglia di endpoint che lo script richiama.

Centralizzare l'acquisizione dei token e la costruzione delle intestazioni nelle funzioni helper, piuttosto che acquisire i token in ogni punto di chiamata, offre tre vantaggi concreti:

• Credenziale centralizzata: un singolo DefaultAzureCredential viene incapsulato in TokenProvider e riutilizzato per ogni chiamata API, quindi il rilevamento dell'identità (interfaccia della riga di comando di Azure, VS Code, identità gestita e così via) avviene una sola volta.
• Token sensibili al destinatario: gli endpoint del cluster di Fabric, Kusto e Power BI rifiutano i token emessi per il destinatario errato. Un generatore di header separato per ogni target mantiene l'ambito corretto accanto al punto di chiamata, così è chiaro a quale endpoint è destinata ciascuna funzione.
• Aggiornato a ogni richiesta: i generatori di intestazioni costruiscono l'intestazione Authorization su richiesta anziché memorizzare nella cache il token stesso. Le credenziali sottostanti vengono aggiornate in modo trasparente, quindi i siti di chiamata non devono mai pensare alla scadenza.

Questa esercitazione chiama Fabric API REST usando ambiti delegati, ad esempio Item.ReadWrite.All.

Aggiungere quanto segue dopo la Config classe :

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

Creare un wrapper del client Fabric

La maggior parte delle chiamate REST di Fabric in questa esercitazione include le stesse intestazioni Authorization e Content-Type. Anziché ripeterli in ogni punto di chiamata, questa esercitazione racchiude httpx.Client in un piccolo FabricClient che aggiunge automaticamente le intestazioni, restituendo comunque il httpx.Response non elaborato così che ogni chiamante possa controllare i codici di stato (ad esempio, per distinguere 201 da 202).

Racchiudere httpx.Client in questo modo, piuttosto che passare headers=_fabric_headers() in ogni punto di chiamata, offre due vantaggi concreti:

• Header in un unico punto: ogni punto di chiamata utilizza automaticamente l'ultimo _fabric_headers(), quindi una nuova richiesta non può essere inviata per errore senza l'header Authorization.
• I codici di stato rimangono visibili: request() restituisce il codice non elaborato httpx.Response anziché JSON decodificato, in modo che i siti di chiamata possano comunque creare rami sullo stato (201 vs 202) e controllare le intestazioni come Location o Retry-After per la gestione dell'archiviazione con ridondanza locale.

Aggiungere quanto segue dopo le funzioni helper di autenticazione:

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

Creare una funzione di supporto LRO

Diverse API REST Fabric usate in questa esercitazione, ad esempio Create Eventhouse, Create Eventstream e Create Map, supportano long-running operations (LROs).

Queste API possono restituire risposte in diversi modelli:

• 201 Created con il corpo della risorsa in linea (sincrona)
• 202 Accepted con un'intestazione Location che punta a un URL di stato dell'operazione (asincrono)
• 202 Accepted con un'intestazione x-ms-operation-id anziché Location (modulo asincrono, alternativo)
• 200 OK con status: "Running" o status: "NotStarted" durante il polling (ancora in corso)
• 200 OK con status: "Succeeded" ma nessun ID della risorsa nel corpo (operazione riuscita; risolvere elencando e abbinando displayName)

Per gestire tutti questi elementi in modo coerente, creare una singola funzione helper che:

1. Restituisce immediatamente l'ID risorsa se la risposta iniziale lo contiene già.
2. In caso contrario, esegue il polling dell'URL dell'operazione (costruito a partire da Location o x-ms-operation-id) utilizzando Retry-After.
3. Considera 200 OK con status: "Running" / "NotStarted" come ancora in corso e continua il polling.
4. In caso di esito positivo, restituisce l'ID della risorsa dal corpo oppure, se il corpo contiene solo lo stato, ripiega sull'elenco delle risorse e sulla ricerca della corrispondenza in base a displayName (con nuovi tentativi).
5. Usa _pbi_headers() quando l'URL di polling è su un cluster di Power BI (*.analysis.windows.net) e le intestazioni di Fabric in caso contrario.

Questa singola funzione di supporto sostituisce la necessità di funzioni di supporto "risolvi per nome" per ciascuna risorsa: ogni funzione create_* in questa esercitazione chiama _handle_lro con i list_url e match_display_name appropriati.

Aggiungere quanto segue dopo la FabricClient classe :

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

Assistente per il carico utile della definizione

Quando si crea una mappa con una definizione pubblica, l'API REST Crea mappa prevede che ogni parte di definition.parts possa contenere un payload con codifica Base64 con "payloadType": "InlineBase64". L'helper _json_to_b64 codifica un dict Python (il tuo map.json) in quel formato, in modo che create_map possa inserirlo direttamente nel corpo della richiesta.

Aggiungere quanto segue dopo la _handle_lro funzione :

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

Crea un helper per fornire la stringa di connessione di Eventstream

Per inviare eventi all'endpoint personalizzato di eventstream, lo script richiede un stringa di connessione per tale endpoint.

A differenza delle API REST di Fabric che hai chiamato finora (ovvero operazioni del piano di controllo per creare e gestire le risorse), l'acquisizione di eventi tramite Eventstream usa un endpoint del piano dati compatibile con Event Hubs e tale endpoint si autentica con una stringa di connessione basata su SAS anziché con un token Microsoft Entra. Il stringa di connessione viene creato quando si aggiunge l'origine endpoint personalizzata e non viene esposto dall'API REST Fabric, quindi deve essere copiato dal portale di Fabric.

get_eventhub_connection_string_interactive riutilizza un valore dalla EVENTHUB_CONNECTION_STRING variabile di ambiente (utile durante le esecuzioni ripetute) o richiede di specificarlo in fase di esecuzione, quindi lo memorizza nella cache in cfg modo che i passaggi successivi possano riutilizzarlo senza richiedere di nuovo.

Aggiungere quanto segue dopo la _json_to_b64 funzione :

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

Creare un helper per inizializzare gli eventi iniziali da CSV

Per assicurarsi che la mappa visualizzi i dati immediatamente dopo la creazione, lo script invia un piccolo set di eventi di inizializzazione nel flusso di eventi prima di creare la mappa.

Senza questo passaggio, la tabella Eventhouse potrebbe non contenere ancora dati e la mappa potrebbe apparire vuota al primo caricamento.

Questa funzione helper legge i dati da un file CSV locale e invia ogni riga come evento JSON al flusso di eventi usando il protocollo EventHub.

Poiché il provisioning delle risorse Eventstream viene eseguito in modo asincrono, l'endpoint personalizzato potrebbe non essere immediatamente pronto per accettare eventi dopo la creazione. Per gestire questo problema, la funzione di supporto include una logica di ripetizione dei tentativi integrata che tenta automaticamente di inviare eventi finché l'endpoint non diventa disponibile. In questo modo il processo di inizializzazione è affidabile e ripetibile e non richiede regolazioni manuali della tempistica.

Questo approccio rispecchia i modelli di inserimento reali:

• I dati vengono generati esternamente (ad esempio, dispositivi O applicazioni IoT)
• Gli eventi vengono trasmessi a Eventstream
• Eventstream distribuisce dati a Eventhouse per l'esecuzione di query e la visualizzazione

Eseguendo il seeding degli eventi iniziali, si simula questo flusso di inserimento e si garantisce che:

• La tabella di destinazione viene popolata
• La funzione KQL contiene dati da restituire
• Il rendering della mappa viene eseguito immediatamente dopo la creazione

Aggiungere il codice seguente dopo la get_eventhub_connection_string_interactive() funzione :

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

Attendere la disponibilità del database KQL

Una volta che l'eventhouse, il database KQL e la funzione KQL esistono, il database KQL potrebbe comunque non essere immediatamente risolvibile da altri endpoint REST Fabric. I servizi di Fabric vengono eseguiti su backend distribuiti, quindi una risorsa appena creata può impiegare un po’ di tempo per propagarsi su tutti i backend.

Se si chiama Crea mappa immediatamente dopo che è stata eseguita la funzione KQL, la funzione Create Map potrebbe non riuscire a risolvere l'origine dati e restituire un errore, ad esempio il database Kusto non trovato.

wait_for_kql_database_ready esegue il polling dell'endpoint REST Fabric per il database KQL e restituisce non appena risponde 200 OK. Si tratta di un controllo best effort: una risposta positiva su questo endpoint del control plane è un forte indicatore del fatto che anche Maps riesca a risolvere il database; inoltre, solleva RuntimeError dopo max_attempts se il database non diventa mai visibile.

Aggiungere quanto segue dopo la seed_eventstream_from_csv funzione :

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

Creare funzioni primarie

Aggiungere quindi le funzioni primarie che definiscono il flusso di lavoro. Questi vengono tutti chiamati da main().

Le funzioni vengono aggiunte nell'ordine in cui sono definite nel codice. main() li invoca in un ordine leggermente diverso, in modo che la tabella KQL esista prima che il flusso di eventi vi si associ e che i dati inizializzati siano disponibili prima che venga eseguita la verifica.

• Crea una sede per eventi
• Creare la tabella KQL
• Verificare l'inserimento (chiamato dopo il seeding)
• Creare un flusso di eventi
• Creare una funzione KQL
• Compilare la definizione della mappa (map.json)
• Compilare i metadati della piattaforma (.platform)
• Creare la mappa

Crea una sede per eventi

create_eventhouse crea una eventhouse nell'area di lavoro e restituisce il relativo ID elemento. L'API REST Create Eventhouse può rispondere alla stessa chiamata in tre modi diversi:

• 201 Created con l'ID di Eventhouse in linea (sincrona).
• 202 Accepted con un URL di operazione LRO (asincrono).
• 409 Conflict con x-ms-public-api-error-code impostato su ItemDisplayNameNotAvailableYet (il nome precedente è ancora riservato nel backend) o ItemDisplayNameAlreadyInUse (un eventhouse con quel nome esiste già nell'area di lavoro).

Per gestire tutte e tre in modo affidabile, create_eventhouse:

• Delega 201 e le risposte di 202 a _handle_lro, che già gestisce in modo uniforme il completamento sincrono e asincrono.
• Retry-After Rispetta e ritenta (fino a cinque tentativi) su ItemDisplayNameNotAvailableYet.
• Riutilizza l'eventhouse esistente in ItemDisplayNameAlreadyInUse elencando le eventhouse nell'area di lavoro e facendo corrispondere in base a displayName.
• Ripiega su un nome di visualizzazione reso univoco (con suffisso UUID breve) se il nome non diventa mai disponibile dopo l’esaurimento del numero massimo di tentativi, in modo che lo script possa continuare l’esecuzione.

Aggiungere quanto segue dopo la wait_for_kql_database_ready funzione :

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

Creare una tabella KQL

create_kql_table_if_missing assicura che la tabella di destinazione esista nel database KQL prima che il flusso di eventi inizi a scrivervi. La destinazione eventstream creata in un secondo momento è configurata con ProcessedIngestion e una tabella fissa tableName, pertanto la tabella deve esistere già quando arrivano gli eventi. In caso contrario, l'inserimento non riesce.

La funzione invia un comando .create-merge table all'endpoint di gestione Kusto dell'eventhouse (queryServiceUri + /v1/rest/mgmt). .create-merge è idempotente: crea la tabella se non esiste e unisce lo schema in caso affermativo. In questo modo è possibile chiamare in modo sicuro su ogni esecuzione.

Prima di eseguire il comando, la funzione legge le proprietà della raccolta eventi per ottenere queryServiceUri e l'ID elemento del database KQL, quindi risolve il database in modo che il displayName payload vi faccia riferimento in base al nome anziché all'IDmgmt.

Aggiungere quanto segue dopo la create_eventhouse funzione :

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

Verificare l'inserimento dati

verify_eventhouse_data conferma che gli eventi immessi nello stream di eventi sono effettivamente arrivati nella tabella Eventhouse. Interroga una <table> | count query sull'endpoint di query Kusto dell'eventhouse (queryServiceUri + /v1/rest/query) finché il conteggio restituito non è maggiore di zero, oppure termina con un errore allo scadere del timeout. L'acquisizione del flusso di eventi dall'endpoint personalizzato alla tabella richiede alcuni secondi, quindi è il polling, anziché una singola query, a fornire un esito positivo/negativo affidabile.

È definito accanto a create_kql_table_if_missing perché entrambi gli helper recuperano le stesse proprietà dell'eventhouse (queryServiceUri, databasesItemIds) e risolvono il displayName del database KQL. Viene chiamato main()doposeed_eventstream_from_csv in modo che gli eventi inizializzati abbiano il tempo di attraversare il flusso degli eventi e raggiungere la tabella prima che venga eseguito il conteggio.

L'esecuzione di questo controllo iniziale rileva in anticipo la configurazione errata dell'inserimento, ad esempio una destinazione eventstream cablata al nome di tabella errato, invece di lasciarla visualizzare in un secondo momento come mappa vuota.

Aggiungere quanto segue dopo la create_kql_table_if_missing funzione :

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

Creare un flusso di eventi con definizione

create_eventstream_with_definition crea un flusso di eventi nell'area di lavoro con la topologia completa inserita nella richiesta, quindi restituisce l'ID elemento di eventstream. L'uso di una definizione pubblica consente di effettuare il provisioning del flusso di eventi e collegare le relative origini, flussi e destinazioni in una singola chiamata, anziché creare prima il flusso di eventi e quindi applicare patch alla relativa definizione.

Prima di inviare la richiesta, la funzione legge le proprietà dell'eventhouse per ottenere l'ID dell'elemento del database KQL e risolve il displayName del database in modo che la destinazione vi faccia riferimento per nome anziché tramite ID. Crea quindi un grafo di flusso di eventi con un'origine CustomEndpoint, un DefaultStream e una destinazione Eventhouse configurata con ProcessedIngestion e il valore fisso tableName da cfg, codifica il grafo in base64 come parte eventstream.json e lo invia tramite POST a Create Eventstream.

L'API REST Create Eventstream può rispondere con 201 Created (sincrona, corpo nel testo), 202 Accepted (LRO asincrona tramite Location o x-ms-operation-id) oppure 200 OK con un payload di completamento contenente solo lo stato, in cui l'eventstream non è ancora visibile nella risposta di List Eventstreams a causa di un ritardo di propagazione nel back-end. _handle_lro copre tutti questi casi, inclusi l'elenco e la corrispondenza tramite displayName, quindi questa funzione gli delega la gestione dell'intera risposta con un'unica chiamata.

Aggiungere quanto segue dopo la verify_eventhouse_data funzione :

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

Creare una funzione KQL

create_kql_function crea (o aggiorna) una funzione Kusto salvata nel database KQL dell'eventhouse e restituisce l'ID dell'elemento del database KQL, in modo che il chiamante possa associare l'origine dati della mappa ad esso. La funzione , LatestVehicleLocations per impostazione predefinita, restituisce la riga più recente per VehicleId tramite arg_max(EventTime, *), proiettando Latitude, Longitude, VehicleId e EventTime in modo che Fabric Mappe possano associare le colonne di latitudine e longitudine del livello.

Come create_kql_table_if_missing, questo helper viene eseguito sull'endpoint di gestione Kusto dell'eventhouse (queryServiceUri + /v1/rest/mgmt) ed è idempotente: .create-or-alter function crea la funzione se non esiste e ne sostituisce il corpo se esiste già, quindi l'helper può essere chiamato in sicurezza a ogni esecuzione.

Il comando viene inviato con skipvalidation=true perché il corpo della funzione fa riferimento alla tabella di destinazione tramite table("<name>") anziché come identificatore bare. Il table() modulo rinvia la risoluzione dei nomi all'ora di query, pertanto la convalida in fase di creazione avrebbe altrimenti esito negativo se la tabella non ha ancora ricevuto dati e il relativo schema non è completamente visibile al validator. Abbinando skipvalidation=true a table("...") è possibile creare la funzione prima che l'acquisizione dei dati abbia popolato la tabella, che è l'ordine seguito in questa esercitazione.

Aggiungere quanto segue dopo la create_eventstream_with_definition funzione :

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

Genera map.json

compila e restituisce il payload /> dataSources (da cui provengono i dati), iconSources (marcatori personalizzati facoltativi), layerSources (su quali query e con quale frequenza) e layerSettings (come viene eseguito il rendering del risultato sulla mappa).

Per questa esercitazione, dataSources punta al database KQL (itemType: "KqlDatabase") creato in precedenza, e l'unica voce in layerSources è un livello basato su Kusto (type: "kusto", queryType: "function"), il cui query chiama la funzione archiviata LatestVehicleLocations(). refreshIntervalMs viene letto da cfg.refresh_interval_ms (5000 ms per impostazione predefinita), quindi il livello esegue nuovamente la funzione su un timer e la mappa riflette il nuovo inserimento quasi in tempo reale.

La voce layerSettings corrispondente associa le colonne dei risultati del livello alla mappa tramite latitudeColumnName: "Latitude" e longitudeColumnName: "Longitude", visualizza ogni riga come un punto bubble e mostra VehicleId e EventTime nei tooltip. La funzione stampa il payload assemblato in modo da poter esaminare l'esatto JSON inviato dalla chiamata Create Map.

Per altre informazioni sull'API REST per la definizione della mappa, vedere Definizione dell'elemento mappa.

Aggiungere quanto segue dopo la create_kql_function funzione :

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

Compilare .platform (metadati della piattaforma)

build_platform_json compila e restituisce una parte facoltativa .platform che la chiamata Crea mappa può includere insieme map.json quando si desidera impostare i metadati degli elementi non predefiniti nella mappa. L'inclusione di una parte .platform non è necessaria: Fabric applica i metadati predefiniti quando la parte viene omessa, ma questa esercitazione illustra come crearne una in modo da poter riutilizzare il modello quando è necessario un controllo esplicito sul tipo di elemento, sul nome visualizzato, sulla descrizione o su un identificatore logico stabile.

Il payload segue lo schema delle proprietà della piattaforma e include due sezioni: metadata (type: "Map", displayName, description) e config (version, logicalId). logicalId viene generato qui come un nuovo UUID, il che va bene per una creazione una tantum; se prevedi di ridistribuire la stessa mappa tramite l'integrazione Git o esecuzioni ripetute, imposta logicalId su un valore stabile in modo che gli aggiornamenti vengano applicati allo stesso elemento.

Per ulteriori informazioni, vedere Definizione dell'elemento mappa e Panoramica della definizione dell'elemento.

Aggiungere quanto segue dopo la build_map_json funzione :

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

Crea una mappa con definizione in linea

create_map crea la mappa inviando tramite POST la definizione inline che hai creato e restituisce l'ID elemento della nuova mappa. La richiesta include tre parti con codifica Base64 in payloadType: "InlineBase64": map.json (la definizione di base necessaria), i metadati facoltativi .platform compilati nel passaggio precedente e un file di query Kusto denominato queries/layerSource-<layerSourceId>.kql che contiene la chiamata alla funzione KQL archiviata. Raggruppare tutte e tre le parti in una singola chiamata esegue il provisioning della mappa e connette il relativo livello dati alla funzione KQL in modo atomico, quindi non è necessario alcun ulteriore round trip getDefinition / updateDefinition.

Il nome del file di query è importante: Fabric risolve la query di un livello associando queries/layerSource-<layerSourceId>.kql rispetto al id della voce corrispondente in layerSources, quindi la funzione estrae l'ID origine del livello da map_json["layerSources"][0]["id"] per costruire il percorso. map.json e .platform sono codificati in base64 tramite _json_to_b64. Il testo della query è codificato direttamente in base64 perché si tratta di una stringa anziché di un dictoggetto .

L'API REST Create Map può rispondere con 201 Created (sincrona, con ID inline), 202 Accepted (LRO asincrona tramite Location o x-ms-operation-id) oppure 200 OK con un payload di completamento contenente solo lo stato, in cui la mappa non è ancora visibile in List Maps a causa di un ritardo di propagazione del backend. _handle_lro copre tutti questi casi, inclusi l'elenco e la corrispondenza tramite displayName, quindi questa funzione gli delega la gestione dell'intera risposta con un'unica chiamata.

Per altre informazioni, vedere Definizione dell'elemento mappa.

Aggiungere quanto segue dopo la build_platform_json funzione :

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

Orchestrare il flusso di lavoro

main è l'unico punto di ingresso che esegue il tutorial dall'inizio alla fine. Istanzia Config, apre un httpx.Client riutilizzato in tutti gli helper, lo racchiude in un FabricClient, quindi chiama ogni funzione di passaggio nell'ordine delle dipendenze: create_eventhouse → create_kql_table_if_missing (deve esistere prima che il flusso di eventi vi si colleghi) → create_eventstream_with_definition → seed_eventstream_from_csv → verify_eventhouse_data (intercetta errori di configurazione dell'acquisizione prima di qualsiasi operazione sulla mappa) → create_kql_function → wait_for_kql_database_ready (controllo best-effort affinché Create Map possa risolvere il database KQL) → build_map_json → build_platform_json → create_map.

L'ordine è importante perché la maggior parte dei passaggi dipende da qualcosa creato da un passaggio precedente: create_eventstream_with_definition richiede l'databasesItemIds dell'eventhouse e create_map richiede il nome della funzione KQL e l'ID dell'elemento del database KQL. Il blocco finale print espone gli ID di ogni risorsa creata in modo da poterli trovare nel portale di Fabric.

Aggiungere quanto segue dopo la create_map funzione :

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

Eseguire l'applicazione

Durante l'esecuzione dello script, viene richiesto di incollare il stringa di connessione eventstream.

Per recuperare questo valore:

1. Apri l'area di lavoro Fabric
2. Aprire il flusso di eventi creato dallo script
3. Selezionare l'origine dell'endpoint personalizzato
4. Apri Autenticazione con chiave SAS
5. Copiare la chiave primaria della stringa di connessione

Incollare il valore nella console quando richiesto.

Eseguire lo script:

python create_realtime_map.py

Verificare che tutti gli elementi siano stati creati:

A questo punto, tutte le risorse vengono create e configurate.

Per simulare lo streaming continuo e osservare l'aggiornamento della mappa quasi in tempo reale, continuare con il completamento Tutorial: Simulare l'inserimento di dati in tempo reale per una mappa usando le API REST e Python. Si basa direttamente su questa esercitazione e riutilizza Eventhouse, il flusso di eventi, la funzione KQL e il mapping che hai creato.

Sommario

In questa esercitazione è stato effettuato il provisioning delle risorse necessarie per una soluzione geospaziale in tempo reale in Microsoft Fabric, usando Fabric API REST e Python.

Sono state eseguite le operazioni seguenti:

• È stato creato un database eventhouse e KQL usando l'API REST Fabric
• Creazione di un flusso di eventi con un endpoint personalizzato per l'inserimento di eventi di streaming
• Definita una funzione KQL per interrogare e strutturare i dati in tempo reale per la visualizzazione su mappa
• Creata e distribuita una mappa di Fabric con una definizione inline che fa riferimento ai dati di Eventhouse
• Inizializzato lo stream di eventi con eventi iniziali, così che la mappa mostrasse immediatamente i dati

Questa architettura illustra un modello comune di analisi in tempo reale in Fabric:

• I producer esterni inviano eventi a Eventstream
• Eventstream instrada e acquisisce i dati in Eventhouse
• Le funzioni KQL trasformano i dati
• Esegue il mapping automatico della query Eventhouse e aggiorna per riflettere i nuovi eventi

Automatizzando la creazione di risorse usando Python e le API REST, è ora disponibile un approccio ripetibile per la creazione di applicazioni spaziali in tempo reale senza configurazione manuale. Per far confluire dati continui nella mappa, prosegui con il tutorial successivo sul simulatore.

Passaggi successivi

Ora che si comprende il flusso end-to-end, è possibile estendere questa soluzione per incorporare un simulatore in tempo reale.

Per un'esercitazione che illustra la creazione di un simulatore in tempo reale per la mappa appena creata usando le API REST, vedere: