Tutorial: Creación de un mapa en tiempo real mediante api REST y Python

Fabric Maps puede visualizar datos geoespaciales en tiempo real conectándose a conjuntos de datos eventhouse que se actualizan continuamente a través de la ingesta de Eventstream.

A diferencia de los escenarios estáticos en los que se usan archivos almacenados en un Lakehouse, este tutorial muestra una arquitectura de streaming basada en eventos donde:

• Los eventos se ingieren en un centro de eventos
• Los datos se consultan mediante el lenguaje de consulta kusto (KQL)
• Un mapa se actualiza dinámicamente a medida que llegan nuevos datos

Este tutorial se centra en automatizar el flujo de trabajo de extremo a extremo mediante las API REST de Fabric y Python, para que puedas aprovisionar recursos y configurar una experiencia de mapa en tiempo real de forma programática. Para ver escenarios de datos estáticos mediante archivos de Lakehouse, consulte Crear un mapa estático mediante las API REST y Python.

En este tutorial, aprenderá a crear y automatizar una solución geoespacial en tiempo real en Microsoft Fabric mediante Eventstream, Eventhouse y KQL.

Para simular el streaming continuo y ver la actualización del mapa casi en tiempo real, complete primero este tutorial y continúe con el seguimiento Tutorial: Simular la ingesta de datos en tiempo real para un mapa mediante las API REST y Python, que se basa directamente en el centro de eventos, la secuencia de eventos, la función KQL y la asignación que cree aquí.

Introducción al escenario: Seguimiento de recursos en tiempo real

Este tutorial se basa en un escenario de seguimiento de activos en tiempo real, similar al escenario de seguimiento de flotas utilizado en el tutorial original de Fabric Maps Tutorial: Crear el enrutamiento de órdenes de trabajo en tiempo real con Fabric Maps.

En este escenario:

• Los vehículos emiten periódicamente actualizaciones de ubicación
• Los eventos de ubicación se incorporan a un eventhouse.
• Un mapa muestra las últimas posiciones y actualizaciones del vehículo automáticamente a medida que llegan nuevos eventos

Este patrón es representativo de casos de uso operativos en tiempo real comunes, como:

• Seguimiento de flotas
• Distribución de órdenes de trabajo
• Supervisión de activos y equipos

Microsoft Fabric usa Eventstream y Eventhouse para ingerir, procesar y analizar datos de streaming casi en tiempo real, lo que permite visualizar datos operativos en directo directamente en un mapa.

En este tutorial se sigue un patrón de automatización común en Fabric: crear la infraestructura → ingerir el flujo → validar la ingesta → representar el mapa.

Prerequisites

• Python 3.10 o posterior
• CLI de Azure
• Identificador del área de trabajo de Fabric
• Permisos para llamar a Fabric API REST, como:
  • Item.ReadWrite.All

Autenticación

En este tutorial se usa DefaultAzureCredential, que puede autenticarse mediante varios orígenes de credenciales locales o de desarrollo. Para los lectores de primera vez, el enfoque más sencillo es el inicio de sesión de la CLI de Azure.

Autenticación local (recomendada para la primera ejecución)

1. Abra un terminal.
2. ¡Corre!

az login

DefaultAzureCredential puede usar la identidad con sesión iniciada para adquirir tokens de acceso para:

• API REST de Fabric (recurso: https://api.fabric.microsoft.com/.default)
• Consultas del plano de datos de Kusto (KQL) en el eventhouse (recurso: https://api.kusto.windows.net/.default)
• Power BI/Fabric puntos de conexión REST usados para el sondeo de operaciones de ejecución prolongada (recurso: https://analysis.windows.net/powerbi/api/.default)

Iniciar sesión en Microsoft Fabric (recomendado)

Antes de ejecutar este tutorial, se recomienda iniciar sesión en Microsoft Fabric al menos una vez:

https://app.fabric.microsoft.com

Iniciar sesión garantiza que tu identidad de Fabric, la pertenencia al rol del área de trabajo y las asignaciones de capacidad estén completamente aprovisionadas antes de adquirir un token de acceso de Microsoft Entra de manera programática.

Este paso es especialmente útil si:

• Es nuevo en Microsoft Fabric
• El área de trabajo se creó recientemente.
• Se ha añadido recientemente la asignación de tu rol

Crear el archivo de datos de inicialización (datos de mapa inicial)

Para asegurarse de que el mapa muestra los datos inmediatamente después del aprovisionamiento, el script envía un pequeño conjunto de eventos de inicialización a la secuencia de eventos.

1. Cree un nuevo archivo en el mismo directorio que el script de Python: vehicle_locations_seed.csv
2. Pegue el siguiente contenido:

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

Paso 1: Creación de un nuevo archivo de proyecto de Python

En este paso, crea un archivo de Python en blanco que irá completando sección por sección.

Cree un nuevo archivo denominado:

create_realtime_map.py

Abra el archivo en el editor.

Paso 2: Instalar las bibliotecas necesarias y agregar instrucciones de importación necesarias

En este paso, instalará las dependencias y agregará las importaciones que usa el script.

Instalación de bibliotecas necesarias

En la ventana de terminal que acaba de abrir, ejecute el siguiente comando:

pip install httpx azure-identity azure-eventhub

Para qué es cada biblioteca

• httpx: realiza solicitudes HTTP a las API REST de Fabric.
• azure-identity: proporciona DefaultAzureCredential para la autenticación de Microsoft Entra.
• azure-eventhub: envía eventos semilla al punto de conexión compatible con Event Hub de Eventstream para rellenar el eventhouse.

Adición de instrucciones de importación al archivo de .py

En la parte superior de create_realtime_map.py, agregue:

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

Paso 3: Agregar una sección de configuración

En este paso, definirá las variables que usa la aplicación, incluidos el identificador del área de trabajo y los nombres de recursos.

La centralización de la configuración en una sola Config clase , en lugar de la dispersión de valores codificados de forma rígida entre funciones, ofrece tres ventajas concretas:

• Portabilidad del entorno: los identificadores de área de trabajo, los nombres de recursos y otras configuraciones se encuentran en un solo lugar, por lo que puede volver a ejecutar el script en un área de trabajo o máquina diferente cambiando algunas líneas (o una variable de entorno) en lugar de buscar a través del código.
• Firmas de función más limpias: las funciones de paso aceptan un solo objeto cfg en lugar de largas listas de parámetros, lo que hace que la orquestación en main() sea fácil de leer.
• Control de secretos más seguros: los valores confidenciales, como el identificador del área de trabajo, se cargan desde variables de entorno, por lo que nunca se confirman junto con el script.

Agregue lo siguiente debajo de las instrucciones 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", "")

Establecimiento del identificador del área de trabajo mediante una variable de entorno

En lugar de codificar el identificador del área de trabajo directamente en el script, este tutorial lo lee de una variable de entorno. Esto mantiene los valores específicos del entorno fuera del código fuente y le permite reutilizar el script en áreas de trabajo o máquinas sin editarlo.

Antes de ejecutar el script, cree una variable de entorno denominada FABRIC_WORKSPACE_ID.

Establecer la variable de entorno en Windows

En Windows, puedes establecer la variable desde cualquier terminal que admita variables de entorno: PowerShell, Windows PowerShell, las ventanas de PowerShell o del Símbolo del sistema integradas en Visual Studio y Visual Studio Code, Terminal Windows y la mayoría de los demás shells.

Ejecute lo siguiente en PowerShell o en el terminal integrado de VS Code:

$env:FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Para confirmar que la variable está establecida:

echo $env:FABRIC_WORKSPACE_ID

Esto establece la variable solo para la sesión de terminal actual.

Establecimiento de una variable de entorno persistente (Windows)

Para que la variable esté disponible en sesiones futuras, use cualquiera de las siguientes opciones:

• PowerShell (comando de una sola línea): Ejecute setx FABRIC_WORKSPACE_ID "<WORKSPACE_ID>". El comando setx escribe en el entorno del usuario, pero no actualiza el terminal actual; cierra y vuelve a abrir el terminal (o abre uno nuevo) antes de ejecutar el script.
• GUI:
  1. Abra Propiedades del sistema.
  2. Seleccione Configuración avanzada del sistema.
  3. Elija Variables de entorno.
  4. En Variables de usuario, seleccione Nuevo.
  5. Entrar:
     • Nombre: FABRIC_WORKSPACE_ID
     • Valor: el ID de tu espacio de trabajo
  6. Seleccione Aceptar para guardar.
  7. Cierre y vuelva a abrir el terminal antes de volver a ejecutar el script.

Establecimiento de la variable de entorno en macOS o Linux

En macOS y Linux, puede establecer la variable desde cualquier shell que admita export: Bash, Zsh (el valor predeterminado en macOS moderno), Fish (con una sintaxis ligeramente diferente) y los terminales integrados en Visual Studio Code y otros editores.

¡Corre!

export FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Para confirmar que la variable está establecida:

echo $FABRIC_WORKSPACE_ID

Esto establece la variable solo para la sesión de shell actual.

Establecimiento de una variable de entorno persistente (macOS o Linux)

Para que la variable esté disponible en sesiones futuras, agregue la export línea al perfil de shell:

• Zsh (valor predeterminado en macOS): ~/.zshrc
• Bash: ~/.bashrc (Linux) o ~/.bash_profile (macOS)
• Fish: ejecutar set -Ux FABRIC_WORKSPACE_ID "<WORKSPACE_ID>" en lugar de editar un archivo

Después de actualizar el perfil, abra un nuevo terminal o ejecute source ~/.zshrc (o el archivo adecuado) para que el cambio surta efecto.

Paso 4: Agregar funciones auxiliares

En este paso, se separan los aspectos transversales —la autenticación, la construcción de encabezados, el sondeo de operaciones de larga duración y la lógica de reintento— en un pequeño conjunto de funciones auxiliares reutilizables a las que pueden llamar todas las funciones de paso.

La centralización de estas preocupaciones en los asistentes , en lugar de insertarlas en cada sitio de llamada, le ofrece tres ventajas concretas:

• Un único origen de verdad para los problemas transversales: casi todas las llamadas API necesitan autenticación, encabezados y sondeoS LRO. Centralizarlos permite que cada función escalonada se centre en su propio recurso, en lugar de tener que reimplementar la adquisición de tokens y la lógica de reintento.
• Resiliencia sin complejidad: los asistentes absorben condiciones transitorias —aprovisionamiento asíncrono, retraso en la propagación del back-end y errores de envío que admiten reintento—, por lo que las funciones de pasos siguen siendo breves y se leen como una lista de comprobación.
• Más fácil de enseñar y modificar: cada asistente se introduce una vez y se reutiliza. Si Fabric cambia un patrón LRO o un alcance de autenticación, lo corriges en un solo lugar.

Los asistentes que agregue en este paso son:

• Utilidades de autenticación: crear encabezados para las API REST de Fabric (y los puntos de conexión de LRO del clúster de Power BI)
• FabricClient: contenedor ligero para llamadas API coherentes
• Controlador de LRO: sondeo de operaciones de larga duración usando Location / x-ms-operation-id / Retry-After, incluidas las respuestas 200-con-Running, los puntos de conexión de clúster de Power BI y las cargas de finalización solo con estado (se resuelve con displayName)
• Asistente de carga útil de definiciones: codificar en base64 map.json para definiciones en línea
• Asistente de conexión de Eventstream: solicita la cadena de conexión del punto de conexión personalizado
• Asistente de inicialización: envía eventos iniciales con lógica de reintento para asegurarse de que la ingesta se realiza correctamente.
• KQL database readiness helper: espera a que la base de datos KQL esté disponible para Fabric Maps

Creación de funciones auxiliares de autenticación

Cada llamada a la API REST de Fabric que se realiza en este tutorial incluye un token de acceso de Microsoft Entra (token de portador) en el encabezado Authorization. En lugar de adquirir tokens ad hoc, este paso encapsula DefaultAzureCredential en un pequeño TokenProvider y expone un generador de cabeceras específico para la audiencia de cada familia de endpoints a la que llama el script.

Centralizar la obtención de tokens y la creación de encabezados en funciones auxiliares, en lugar de obtener tokens en cada punto de llamada, ofrece tres ventajas concretas:

• Credencial centralizada: una única DefaultAzureCredential se encapsula en TokenProvider y se reutiliza para cada llamada a la API, por lo que la detección de identidad (CLI de Azure, VS Code, identidad administrada, etc.) se realiza una sola vez.
• Tokens orientados a la audiencia: los puntos de conexión del clúster de Fabric, Kusto y Power BI rechazan los tokens emitidos para una audiencia incorrecta. Un generador de encabezados independiente por audiencia mantiene el ámbito correcto justo junto al sitio de llamada, por lo que es obvio qué punto de conexión tiene como destino cada función.
• Actualizado en cada solicitud: los generadores de cabeceras generan la cabecera Authorization bajo demanda, en lugar de almacenar ellos mismos el token en caché. La credencial subyacente se actualiza de forma transparente, por lo que los sitios de llamada nunca tienen que pensar en la expiración.

En este tutorial se usan las API REST de Fabric mediante permisos delegados, como Item.ReadWrite.All.

Agregue lo siguiente después de la Config clase :

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

Cree un envoltorio de cliente de Fabric

La mayoría de las llamadas a la API REST de Fabric de este tutorial envían los mismos encabezados Authorization y Content-Type. En lugar de repetirlos en cada punto de llamada, este tutorial encapsula httpx.Client en un pequeño FabricClient que añade las cabeceras automáticamente sin dejar de devolver el httpx.Response sin procesar, para que el código que llama pueda inspeccionar los códigos de estado (por ejemplo, para distinguir 201 de 202).

Encapsular httpx.Client de esta manera —en lugar de pasar headers=_fabric_headers() en cada punto de llamada— te da dos ventajas concretas:

• Cabeceras en un solo lugar: cada punto de llamada obtiene automáticamente la versión más reciente de _fabric_headers(), por lo que una nueva petición no puede enviarse accidentalmente sin la cabecera Authorization.
• Los códigos de estado siguen siendo visibles: request() devuelve el httpx.Response sin procesar en lugar de JSON descodificado, por lo que los puntos de llamada pueden seguir ramificándose según el estado (201 frente a 202) e inspeccionar cabeceras como Location o Retry-After para gestionar las LRO.

Agregue lo siguiente después de las funciones auxiliares de autenticación:

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

Creación de una función auxiliar LRO

Varias API rest de Fabric usadas en este tutorial, como Create Eventhouse, Create Eventstream y Create Map , admiten long-running operations (LROs).

Estas API pueden devolver respuestas en varios patrones:

• 201 Created con el cuerpo del recurso en línea (de forma síncrona)
• 202 Accepted con un Location encabezado que apunta a una dirección URL de estado de la operación (asincrónica)
• 202 Accepted con un x-ms-operation-id encabezado en lugar de Location (asincrónico, formulario alternativo)
• 200 OK con status: "Running" o status: "NotStarted" durante el sondeo (todavía en curso)
• 200 OK con status: "Succeeded" pero sin identificador de recurso en el cuerpo (correcta; se resuelve enumerando y haciendo coincidir displayName)

Para controlar todos estos elementos de forma coherente, cree una única función auxiliar que:

1. Devuelve el identificador de recurso inmediatamente si la respuesta inicial ya la contiene.
2. De lo contrario, sondea la dirección URL de la operación (compilada a partir de Location o x-ms-operation-id) mediante Retry-After.
3. Considera 200 OK con status: "Running" / "NotStarted" como aún en curso y continúa sondeando.
4. Si se ejecuta correctamente, devuelve el identificador de recurso del cuerpo o vuelve a enumerar los recursos y buscar coincidencias con displayName (con reintentos) cuando el cuerpo es de solo estado.
5. Usa _pbi_headers() cuando la dirección URL de sondeo está en un clúster de Power BI (*.analysis.windows.net) y los encabezados de Fabric en caso contrario.

Esta única función auxiliar reemplaza la necesidad de contar con funciones auxiliares de «resolver por nombre» para cada recurso: todas las funciones create_* de este tutorial llaman a _handle_lro con los list_url y match_display_name adecuados.

Agregue lo siguiente después de la FabricClient clase :

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

Asistente de carga de definición

Cuando se crea un mapa con una definición pública, la API REST Create map espera que cada parte definition.parts de lleve una carga codificada en base64 con "payloadType": "InlineBase64". El auxiliar _json_to_b64 codifica un dict de Python (tu map.json) en ese formato para que create_map pueda insertarlo directamente en el cuerpo de la solicitud.

Agregue lo siguiente después de la _handle_lro función :

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

Crear una función auxiliar para proporcionar la cadena de conexión de eventstream

Para enviar eventos al punto de conexión personalizado de eventstream, el script necesita un cadena de conexión para ese punto de conexión.

A diferencia de las API REST de Fabric que ha estado usando hasta ahora (que son operaciones del plano de control para crear y administrar recursos), la ingesta de flujos de eventos usa un punto de conexión del plano de datos compatible con Event Hubs, y ese punto de conexión se autentica con una cadena de conexión basada en SAS en lugar de un token de Microsoft Entra. La cadena de conexión se crea al agregar el origen de punto de conexión personalizado y la API REST de Fabric no la expone, por lo que debe copiarse desde el portal de Fabric.

get_eventhub_connection_string_interactive reutiliza un valor de la variable de entorno EVENTHUB_CONNECTION_STRING (útil en ejecuciones repetidas) o te lo solicita en tiempo de ejecución y, a continuación, lo almacena en caché en cfg para que los pasos posteriores puedan reutilizarlo sin tener que volver a solicitarlo.

Agregue lo siguiente después de la _json_to_b64 función :

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

Creación de un asistente para inicializar eventos iniciales desde CSV

Para asegurarse de que el mapa muestra los datos inmediatamente después de crearlos, el script envía un pequeño conjunto de eventos de inicialización a la secuencia de eventos antes de crear el mapa.

Sin este paso, es posible que la tabla Eventhouse aún no contenga datos y el mapa podría aparecer vacío en la primera carga.

Esta función auxiliar lee datos de un archivo CSV local y envía cada fila como un evento JSON a la secuencia de eventos mediante el protocolo EventHub.

Dado que los recursos de Eventstream se aprovisionan de forma asincrónica, es posible que el punto de conexión personalizado no esté listo inmediatamente para aceptar eventos después de la creación. Para controlar esto, la función auxiliar incluye lógica de reintento integrada que intenta enviar automáticamente eventos hasta que el punto de conexión esté disponible. Esto garantiza que el proceso de inicialización sea confiable y repetible y no requiera ajustes manuales de control de tiempo.

Este enfoque refleja patrones de ingesta reales:

• Los datos se generan externamente (por ejemplo, dispositivos o aplicaciones de IoT).
• Los eventos se transmiten a Eventstream
• Eventstream entrega datos a Eventhouse para consultar y visualizar

Al introducir eventos iniciales, se simula este flujo de ingesta y se garantiza que:

• La tabla de destino se rellena
• La función KQL tiene datos para devolver
• El mapa se representa inmediatamente después de la creación.

Agregue el código siguiente después de la get_eventhub_connection_string_interactive() función :

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

Esperar la disponibilidad de la base de datos KQL

Una vez que el centro de eventos, su base de datos KQL y la función KQL existen, es posible que la base de datos KQL todavía no se pueda resolver inmediatamente desde otros puntos de conexión REST de Fabric. Los servicios de Fabric se ejecutan en servidores back-end distribuidos, por lo que un recurso recién creado puede tardar un poco en propagarse por todos ellos.

Si llama a Create Map inmediatamente después de que la función KQL esté implementada, es posible que Create Map no pueda resolver el origen de datos y devuelva un error como Base de datos de Kusto no encontrada.

wait_for_kql_database_ready sondea el punto de conexión rest de Fabric para la base de datos KQL y devuelve tan pronto como responda 200 OK. Se trata de una comprobación de máximo esfuerzo: una respuesta satisfactoria en este punto de conexión del plano de control es un indicio sólido de que Maps también puede resolver la base de datos, y genera RuntimeError después de max_attempts si la base de datos nunca llega a ser visible.

Agregue lo siguiente después de la seed_eventstream_from_csv función :

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

Creación de funciones principales

A continuación, agregue las funciones principales que definen el flujo de trabajo. Todos estos se invocan desde main().

Las funciones se agregan en el orden en que se definen en el código. main() las llama en un orden ligeramente diferente, por lo que la tabla KQL existe antes de que el flujo de eventos se enlace a él y, por tanto, los datos iniciales están disponibles antes de que se ejecute la comprobación.

• Crear una casa de eventos
• Creación de la tabla KQL
• Verificar la ingesta (se ejecuta después de la inicialización)
• Creación de una secuencia de eventos
• Creación de una función KQL
• Crear la definición del mapa (map.json)
• Compilación de los metadatos de la plataforma (.platform)
• Creación del mapa

Crear una casa de eventos

create_eventhouse crea un centro de eventos en el área de trabajo y devuelve su identificador de elemento. La API REST Create Eventhouse puede responder a la misma llamada de tres maneras diferentes:

• 201 Created con el identificador del centro de eventos insertado (sincrónico).
• 202 Accepted con una dirección URL de operación LRO (asincrónica).
• 409 Conflict con x-ms-public-api-error-code configurado como ItemDisplayNameNotAvailableYet (el nombre anterior sigue reservado en el backend) o ItemDisplayNameAlreadyInUse (ya existe un eventhouse con ese nombre en el espacio de trabajo).

Para gestionar las tres de manera fiable, create_eventhouse:

• Delega las respuestas 201 y 202 a _handle_lro, que ya cubre de manera uniforme la finalización sincrónica y asincrónica.
• Respeta Retry-After y vuelve a intentarlo (hasta cinco intentos) en ItemDisplayNameNotAvailableYet.
• Reutiliza el centro de eventos existente en ItemDisplayNameAlreadyInUse enumerando los centros de eventos del área de trabajo y haciéndolo coincidir con displayName.
• Recurre a un nombre para mostrar único (con un UUID corto como sufijo) si el nombre nunca llega a estar disponible tras agotar el número de reintentos permitidos, para que el script pueda seguir avanzando.

Agregue lo siguiente después de la wait_for_kql_database_ready función :

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

Creación de una tabla KQL

create_kql_table_if_missing garantiza que la tabla de destino existe en la base de datos KQL antes de que el flujo de eventos empiece a escribir en ella. El destino de Eventstream que cree más adelante se configura con ProcessedIngestion y un tableName fijo, por lo que la tabla ya debe existir cuando lleguen los eventos; de lo contrario, la ingesta falla.

La función emite un comando .create-merge table contra el punto de conexión de administración de Kusto del Eventhouse (queryServiceUri + /v1/rest/mgmt). .create-merge es idempotente: crea la tabla si no existe y combina el esquema si lo hace. Eso hace que sea seguro invocarlo en cada ejecución.

Antes de ejecutar el comando, la función lee las propiedades del eventhouse para obtener queryServiceUri y el identificador del elemento de la base de datos KQL y, a continuación, resuelve el displayName de la base de datos para que la carga de mgmt haga referencia a ella por nombre en lugar de por identificador.

Agregue lo siguiente después de la create_eventhouse función :

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

Comprobación de la ingesta de datos

verify_eventhouse_data confirma que los eventos insertados en el flujo de eventos realmente llegaron a la tabla de Eventhouse. Sondea una <table> | count consulta en el punto de conexión de consulta de Kusto del eventhouse (queryServiceUri + /v1/rest/query) hasta que el recuento devuelto sea mayor que cero o produzca un error al agotarse el tiempo de espera. La ingesta del flujo de eventos tarda unos segundos en llegar desde el punto de conexión personalizado hasta la tabla, por lo que realizar consultas periódicas, en lugar de una sola consulta, es lo que proporciona una indicación fiable de éxito o fallo.

Está definido junto a create_kql_table_if_missing porque ambas funciones auxiliares buscan las mismas propiedades de eventhouse (queryServiceUri, databasesItemIds) y resuelven el displayName de la base de datos KQL. Se llama desde main()despuésseed_eventstream_from_csv para que los eventos de inicialización tengan la oportunidad de fluir a través de la secuencia de eventos y llegar a la tabla antes de que se ejecute el recuento.

Ejecutar esta comprobación de antemano permite detectar pronto una mala configuración de la ingesta —por ejemplo, un destino del flujo de eventos configurado con un nombre de tabla incorrecto— en lugar de dejar que se manifieste más adelante como un mapa vacío.

Agregue lo siguiente después de la create_kql_table_if_missing función :

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

Creación de una secuencia de eventos con definición

create_eventstream_with_definition crea una secuencia de eventos en el área de trabajo con toda su topología integrada en la solicitud y, a continuación, devuelve el identificador del elemento de la secuencia de eventos. El uso de una definición pública permite aprovisionar la secuencia de eventos y conectar sus orígenes, secuencias y destinos en una sola llamada, en lugar de crear primero el flujo de eventos y, a continuación, aplicar revisiones a su definición.

Antes de enviar la solicitud, la función lee las propiedades del centro de eventos para obtener el identificador de elemento de la base de datos KQL y resuelve la base de datos displayName para que el destino haga referencia a ella por nombre en lugar de por identificador. A continuación, construye un grafo de Eventstream con una fuente CustomEndpoint, un DefaultStream y un destino Eventhouse configurado con ProcessedIngestion y el valor fijo tableName de cfg, codifica el grafo en base64 como la parte eventstream.json y lo envía mediante POST a Create Eventstream.

La API REST Create Eventstream puede responder con 201 Created (sincrónica, con el cuerpo insertado en línea), 202 Accepted (LRO asincrónica mediante Location o x-ms-operation-id) o 200 OK con una carga de finalización solo de estado, en la que el Eventstream aún no está visible en la respuesta de List Eventstreams debido a un retraso en la propagación del backend. _handle_lro cubre todos estos casos —incluida la enumeración y la coincidencia por displayName—, por lo que esta función le delega la gestión completa de la respuesta en una sola llamada.

Agregue lo siguiente después de la verify_eventhouse_data función :

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

Creación de una función KQL

create_kql_function crea (o actualiza) una función de Kusto almacenada en la base de datos KQL del eventhouse y devuelve el identificador del elemento de la base de datos KQL para que quien realiza la llamada pueda conectar a ella el origen de datos del mapa. La función — LatestVehicleLocations de manera predeterminada — devuelve la fila más reciente por VehicleId mediante arg_max(EventTime, *), proyectando Latitude, Longitude, VehicleId y EventTime para que Fabric Maps pueda enlazar las columnas de latitud y longitud de la capa.

Al igual que create_kql_table_if_missing, esta utilidad auxiliar se ejecuta contra el punto de conexión de administración de Kusto de Eventhouse (queryServiceUri + /v1/rest/mgmt) y es idempotente: .create-or-alter function crea la función si no existe y sustituye su cuerpo si ya existe, por lo que se puede invocar con seguridad en cada ejecución.

El comando se envía con skipvalidation=true porque el cuerpo de la función hace referencia a la tabla de destino a través de table("<name>") en lugar de hacerlo como un identificador simple. El table() formulario aplaza la resolución de nombres al tiempo de consulta, por lo que la validación en el momento de la creación produciría un error si la tabla aún no ha recibido ningún dato y su esquema no es totalmente visible para el validador. El emparejamiento skipvalidation=true con table("...") permite crear la función antes de que la ingesta haya rellenado la tabla, que es el orden en el que se ejecuta este tutorial.

Agregue lo siguiente después de la create_eventstream_with_definition función :

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

Generar map.json

build_map_json genera y devuelve la carga útil de map.json que define el contenido de Fabric Map. La carga sigue el esquema de definición de elemento de mapa y se compone de cuatro secciones: dataSources (de dónde proceden los datos), iconSources (marcadores personalizados opcionales), layerSources (qué se consulta y con qué frecuencia) y layerSettings (cómo se representa el resultado en el mapa).

En este tutorial, dataSources apunta a la base de datos KQL (itemType: "KqlDatabase") creada anteriormente, y la única entrada de layerSources es una capa respaldada por Kusto (type: "kusto", queryType: "function") cuya query llama a la función almacenada LatestVehicleLocations(). refreshIntervalMs se lee de cfg.refresh_interval_ms (5000 ms de forma predeterminada), por lo que la capa vuelve a ejecutar la función en un temporizador y el mapa refleja la nueva ingesta casi en tiempo real.

La entrada coincidente layerSettings vincula las columnas de resultados de la capa al mapa a través de latitudeColumnName: "Latitude" y longitudeColumnName: "Longitude", representa cada fila como un punto bubble y muestra VehicleId y EventTime en la información de herramientas. La función imprime la carga ensamblada para que pueda inspeccionar el JSON exacto que envía la llamada crear mapa.

Para obtener más información sobre la API REST de definición de mapa, consulte Definición de elemento de mapa.

Agregue lo siguiente después de la create_kql_function función :

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

Generación de .platform (metadatos de la plataforma)

build_platform_json genera y devuelve una parte opcional .platform que la llamada Create Map puede incluir junto con map.json cuando desea establecer metadatos no predeterminados de los elementos en el mapa. No es necesario incluir un elemento .platform , Fabric aplica metadatos predeterminados cuando se omite el elemento, pero en este tutorial se muestra cómo crear uno para poder reutilizar el patrón cuando necesite un control explícito sobre el tipo de elemento, el nombre para mostrar, la descripción o un identificador lógico estable.

La carga sigue el esquema de propiedades de la plataforma y tiene dos secciones: metadata (type: "Map", displayName, description) y config (version, logicalId). logicalId se genera como un UUID nuevo aquí, que es adecuado para una creación única; si planea volver a implementar el mismo mapa a través de la integración de Git o ejecuciones repetidas, ancle logicalId a un valor estable para que las actualizaciones tengan como destino el mismo elemento.

Para obtener más información, vea Definición de elemento de mapa e Introducción a la definición de elemento.

Agregue lo siguiente después de la build_map_json función :

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

Crear un mapa con definición en línea

create_map crea el mapa mediante una solicitud POST de la definición insertada que has ensamblado y devuelve el ID del elemento del nuevo mapa. La solicitud incluye tres partes codificadas en base64 en payloadType: "InlineBase64": map.json (la definición básica necesaria), los metadatos opcionales .platform que creó en el paso anterior y un archivo de consulta kusto denominado queries/layerSource-<layerSourceId>.kql que contiene la llamada a la función KQL almacenada. Agrupar las tres partes en una sola llamada aprovisiona el mapa y conecta su capa de datos a la función KQL de forma atómica, por lo que no se necesita ningún intercambio posterior getDefinition / updateDefinition.

El nombre del archivo de consulta es importante: Fabric resuelve la consulta de una capa mediante la coincidencia de queries/layerSource-<layerSourceId>.kql con la id de la entrada correspondiente en layerSources, por lo que la función extrae el identificador de origen de capa de map_json["layerSources"][0]["id"] para construir la ruta de acceso. map.json y .platform se codifican en base64 mediante _json_to_b64; el texto de consulta se codifica directamente en base64 porque es una cadena y no un dict.

La API REST Create Map puede responder con 201 Created (sincrónica, ID en línea), 202 Accepted (LRO asíncrona mediante Location o x-ms-operation-id) o 200 OK con una carga de finalización con solo el estado, en la que el mapa aún no es visible en List Maps debido a un retraso en la propagación del backend. _handle_lro cubre todos estos casos —incluida la enumeración y la coincidencia por displayName—, por lo que esta función le delega la gestión completa de la respuesta en una sola llamada.

Para obtener más información, vea Definición de elemento de mapa.

Agregue lo siguiente después de la build_platform_json función :

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

Orquestación del flujo de trabajo

main es el único punto de entrada que ejecuta el tutorial de un extremo a otro. Instancia Config, abre un httpx.Client que reutiliza en cada función auxiliar, lo encapsula en FabricClient y, a continuación, llama a cada función de paso en orden de dependencia: create_eventhouse → create_kql_table_if_missing (debe existir antes de que el flujo de eventos se vincule a él) → create_eventstream_with_definition → seed_eventstream_from_csv → verify_eventhouse_data (detecta una configuración incorrecta de la ingesta antes de cualquier operación del mapa) → create_kql_function → wait_for_kql_database_ready (control de mejor esfuerzo para que Create Map pueda resolver la base de datos KQL) → build_map_json → build_platform_json → create_map.

El orden es importante porque la mayoría de los pasos consumen algo creado por un paso anterior: create_eventstream_with_definition necesita el databasesItemIds del eventhouse, y create_map necesita el nombre de la función KQL y el identificador del elemento de la base de datos KQL. El bloque final print muestra los identificadores de todos los recursos creados para que puedas encontrarlos en el portal de Fabric.

Agregue lo siguiente después de la create_map función :

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

Ejecutar la aplicación

Durante la ejecución del script, se le solicitará que pegue la cadena de conexión de Eventstream.

Para recuperar este valor:

1. Abre tu área de trabajo de Fabric
2. Abra la secuencia de eventos creada por el script.
3. Selección del origen del punto de conexión personalizado
4. Abrir autenticación con clave SAS
5. Copiar cadena de conexión: clave principal

Pegue el valor en la consola cuando se le solicite.

Ejecute el script:

python create_realtime_map.py

Compruebe que se crearon todos los elementos:

En este momento, todos los recursos se crean y configuran.

Para simular el streaming continuo y ver la actualización del mapa casi en tiempo real, continúe con la Tutorial: Simular la ingesta de datos en tiempo real para un mapa mediante las API REST y Python. Se basa directamente en este tutorial y reutiliza el centro de eventos, la secuencia de eventos, la función KQL y el mapa que ha creado.

Resumen

En este tutorial, aprovisionó los recursos necesarios para una solución geoespacial en tiempo real en Microsoft Fabric, mediante Fabric API REST y Python.

Ha logrado lo siguiente:

• Se creó una base de datos eventhouse y KQL mediante la API rest de Fabric
• Creación de una secuencia de eventos con un punto de conexión personalizado para la ingesta de eventos de streaming
• Se definió una función KQL para consultar y dar forma a los datos en tiempo real para la visualización del mapa.
• Se ha creado e implementado un mapa de Fabric con una definición en línea que hace referencia a datos de Eventhouse
• Inicializa la secuencia de eventos con eventos iniciales para que el mapa muestre los datos inmediatamente

Esta arquitectura muestra un patrón de análisis en tiempo real común en Fabric:

• Los productores externos envían eventos a Eventstream
• Eventstream enruta e ingiere datos en Eventhouse
• Las funciones KQL transforman los datos
• Maps consulta Eventhouse y se actualiza automáticamente para mostrar los nuevos eventos

Al automatizar la creación de recursos mediante Python y API REST, ahora tiene un enfoque repetible para compilar aplicaciones espaciales en tiempo real sin configuración manual. Para enviar datos continuamente al mapa, continúa con el siguiente tutorial del simulador.

Pasos siguientes

Ahora que comprende el flujo de un extremo a otro, puede ampliar esta solución para incorporar un simulador en tiempo real.

Para ver un tutorial que muestra cómo crear un simulador en tiempo real para el mapa que acaba de crear mediante las API REST, consulte: