Εκμάθηση: Δημιουργήστε έναν χάρτη σε πραγματικό χρόνο χρησιμοποιώντας REST API και Python

Οι Χάρτες Fabric μπορούν να απεικονίσουν γεωχωρικά δεδομένα σε πραγματικό χρόνο συνδέοντας σε σύνολα δεδομένων eventhouse που ενημερώνονται συνεχώς μέσω της πρόσληψης ροής συμβάντων.

Σε αντίθεση με τα στατικά σενάρια που χρησιμοποιούν αρχεία που είναι αποθηκευμένα σε ένα Lakehouse, αυτό το εκπαιδευτικό βοήθημα παρουσιάζει μια αρχιτεκτονική ροής, βάσει συμβάντων, όπου:

• Τα συμβάντα απορροφώνται σε μια αποθήκη εκδηλώσεων
• Η αναζήτηση δεδομένων γίνεται με χρήση της γλώσσας ερωτημάτων Kusto (KQL)
• Ένας χάρτης ανανεώνεται δυναμικά καθώς φτάνουν νέα δεδομένα

Αυτό το εκπαιδευτικό βοήθημα εστιάζει στο αυτοματοποίηση της ροής εργασιών από άκρο σε άκρο χρησιμοποιώντας Fabric REST API και Python, ώστε να μπορείτε να παρέχετε πόρους και να διαμορφώσετε μια εμπειρία χάρτη σε πραγματικό χρόνο μέσω προγραμματισμού. Για σενάρια στατικών δεδομένων που χρησιμοποιούν αρχεία Lakehouse, ανατρέξτε στο θέμα Δημιουργία στατικού χάρτη χρησιμοποιώντας REST API και Python.

Σε αυτό το εκπαιδευτικό βοήθημα, θα μάθετε πώς μπορείτε να δημιουργήσετε και να αυτοματοποιήσετε μια γεωχωρική λύση σε πραγματικό χρόνο στο Microsoft Fabric χρησιμοποιώντας το Eventstream, το Eventhouse και το KQL.

Για να προσομοιώσετε τη συνεχή ροή και να παρακολουθήσετε την ενημέρωση του χάρτη σε σχεδόν πραγματικό χρόνο, ολοκληρώστε πρώτα αυτό το πρόγραμμα εκμάθησης και, στη συνέχεια, συνεχίστε με το Πρόγραμμα εκμάθησης: Προσομοίωση πρόσληψης δεδομένων σε πραγματικό χρόνο για έναν χάρτη χρησιμοποιώντας REST API και Python, το οποίο δημιουργείται απευθείας στην αποθήκη συμβάντων, τη ροή συμβάντων, τη συνάρτηση KQL και τον χάρτη που δημιουργείτε εδώ.

Επισκόπηση σεναρίου: Παρακολούθηση περιουσιακών στοιχείων σε πραγματικό χρόνο

Αυτό το σεμινάριο βασίζεται σε ένα σενάριο παρακολούθησης περιουσιακών στοιχείων σε πραγματικό χρόνο, παρόμοιο με το σενάριο παρακολούθησης στόλου που χρησιμοποιήθηκε στο αρχικό Fabric Maps Tutorial: Δημιουργήστε δρομολόγηση παραγγελιών εργασίας σε πραγματικό χρόνο με Fabric Maps.

Σε αυτό το σενάριο:

• Τα οχήματα εκπέμπουν περιοδικά ενημερώσεις τοποθεσίας
• Τα συμβάντα τοποθεσίας προσλαμβάνονται σε μια αποθήκη συμβάντων
• Ένας χάρτης εμφανίζει τις πιο πρόσφατες θέσεις οχημάτων και ενημερώνεται αυτόματα καθώς φτάνουν νέα γεγονότα

Αυτό το μοτίβο είναι αντιπροσωπευτικό των κοινών περιπτώσεων λειτουργικής χρήσης σε πραγματικό χρόνο, όπως:

• Παρακολούθηση στόλου
• Αποστολή παραγγελίας εργασίας
• Παρακολούθηση περιουσιακών στοιχείων και εξοπλισμού

Microsoft Fabric χρησιμοποιεί Eventstream και Eventhouse για την πρόσληψη, επεξεργασία και ανάλυση δεδομένων ροής σε σχεδόν πραγματικό χρόνο, καθιστώντας δυνατή την οπτικοποίηση ζωντανών λειτουργικών δεδομένων απευθείας σε έναν χάρτη.

Αυτό το εκπαιδευτικό βοήθημα ακολουθεί ένα κοινό μοτίβο αυτοματοποίησης στο Fabric: δημιουργία υποδομής → πρόσληψη ροής → επικύρωση πρόσληψης → χάρτη απόδοσης.

Προϋποθέσεις

• Python 3.10 ή νεότερη έκδοση
• Azure CLI
• Αναγνωριστικό χώρου εργασίας Fabric
• Δικαιώματα για την κλήση Fabric REST API, όπως:
  • Item.ReadWrite.All

Έλεγχος ταυτότητας

Αυτό το σεμινάριο χρησιμοποιεί DefaultAzureCredential, το οποίο μπορεί να πραγματοποιήσει έλεγχο ταυτότητας χρησιμοποιώντας πολλές πηγές τοπικών/διαπιστευτηρίων προγραμματιστή. Για τους πρώτους αναγνώστες, η απλούστερη προσέγγιση είναι η είσοδος στο Azure CLI.

Τοπικός έλεγχος ταυτότητας (συνιστάται για πρώτη εκτέλεση)

1. Ανοίξτε ένα τερματικό.
2. Τρέξιμο:

az login

DefaultAzureCredential Μπορείτε να χρησιμοποιήσετε την ταυτότητα που έχετε συνδέσει για να αποκτήσετε διακριτικά πρόσβασης για:

• API Fabric REST (πόρος: https://api.fabric.microsoft.com/.default)
• Ερωτήματα επιπέδου δεδομένων Kusto (KQL) σε σχέση με την αποθήκη συμβάντων (πόρος: https://api.kusto.windows.net/.default)
• Τελικά σημεία REST Power BI / Fabric που χρησιμοποιούνται για polling μεγάλης διάρκειας (πόρος: https://analysis.windows.net/powerbi/api/.default)

Είσοδος στο Microsoft Fabric (συνιστάται)

Πριν εκτελέσετε αυτό το πρόγραμμα εκμάθησης, συνιστούμε να εισέλθετε στο Microsoft Fabric τουλάχιστον μία φορά:

https://app.fabric.microsoft.com

Η είσοδος διασφαλίζει ότι η ταυτότητα Fabric, η ιδιότητα μέλους ρόλου χώρου εργασίας και οι εκχωρήσεις εκχωρημένων πόρων παρέχονται πλήρως πριν από την απόκτηση ενός διακριτικού πρόσβασης Microsoft Entra μέσω προγραμματισμού.

Αυτό το βήμα είναι ιδιαίτερα χρήσιμο εάν:

• Είστε νέος χρήστης του Microsoft Fabric
• Ο χώρος εργασίας δημιουργήθηκε πρόσφατα
• Η ανάθεση ρόλου προστέθηκε πρόσφατα

Δημιουργία του αρχείου δεδομένων σπόρου (αρχικά δεδομένα χάρτη)

Για να διασφαλιστεί ότι η αντιστοίχιση εμφανίζει δεδομένα αμέσως μετά την παροχή, η δέσμη ενεργειών στέλνει ένα μικρό σύνολο αρχικών συμβάντων στη ροή συμβάντων.

1. Δημιουργήστε ένα νέο αρχείο στον ίδιο κατάλογο με το σενάριο Python: vehicle_locations_seed.csv
2. Επικολλήστε το ακόλουθο περιεχόμενο:

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

Βήμα 1—Δημιουργήστε ένα νέο αρχείο έργου Python

Σε αυτό το βήμα, δημιουργείτε ένα κενό αρχείο Python το οποίο δημιουργείτε ανά ενότητα.

Δημιουργήστε ένα νέο αρχείο με το όνομα:

create_realtime_map.py

Ανοίξτε το αρχείο στο πρόγραμμα επεξεργασίας.

Βήμα 2—Εγκαταστήστε τις απαιτούμενες βιβλιοθήκες και προσθέστε τις απαιτούμενες προτάσεις εισαγωγής

Σε αυτό το βήμα, εγκαθιστάτε τις εξαρτήσεις και προσθέτετε τις εισαγωγές που χρησιμοποιεί η δέσμη ενεργειών σας.

Εγκατάσταση απαιτούμενων βιβλιοθηκών

Στο παράθυρο τερματικού που μόλις ανοίξατε, εκτελέστε την ακόλουθη εντολή:

pip install httpx azure-identity azure-eventhub

Σε τι χρησιμεύει κάθε βιβλιοθήκη

• httpx: κάνει αιτήσεις HTTP στα API REST του Fabric.
• azure-identity: παρέχει DefaultAzureCredential για έλεγχο ταυτότητας Microsoft Entra.
• azure-Eventhub: Αποστέλλει αρχικά συμβάντα στο τελικό σημείο συμβατό με το Κέντρο συμβάντων του Eventstream για να συμπληρώσει την αποθήκη συμβάντων.

Προσθήκη δηλώσεων εισαγωγής στο αρχείο .py

Στο επάνω μέρος του create_realtime_map.py, προσθέστε:

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

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

Βήμα 3—Προσθήκη ενότητας διαμόρφωσης

Σε αυτό το βήμα, ορίζετε τις μεταβλητές που χρησιμοποιεί η εφαρμογή σας, συμπεριλαμβανομένου του αναγνωριστικού χώρου εργασίας και των ονομάτων πόρων.

Η συγκέντρωση της ρύθμισης παραμέτρων σε μία μόνο Config κλάση — αντί για τη διασπορά τιμών με ενσωματωμένο κώδικα σε συναρτήσεις — σας δίνει τρία συγκεκριμένα πλεονεκτήματα:

• Φορητότητα περιβάλλοντος: Τα αναγνωριστικά χώρου εργασίας, τα ονόματα πόρων και άλλες ρυθμίσεις βρίσκονται σε ένα σημείο, ώστε να μπορείτε να εκτελέσετε ξανά τη δέσμη ενεργειών σε διαφορετικό χώρο εργασίας ή υπολογιστή αλλάζοντας μερικές γραμμές (ή μια μεταβλητή περιβάλλοντος) αντί να αναζητήσετε τον κώδικα.
• Καθαρότερες υπογραφές συναρτήσεων: Οι συναρτήσεις βημάτων δέχονται ένα μεμονωμένο cfg αντικείμενο αντί για μεγάλες λίστες παραμέτρων, γεγονός που διατηρεί την ενορχήστρωση ευανάγνωστη main() .
• Ασφαλέστερος χειρισμός μυστικών: Οι ευαίσθητες τιμές, όπως το αναγνωριστικό χώρου εργασίας, φορτώνονται από μεταβλητές περιβάλλοντος, επομένως δεν δεσμεύονται ποτέ μαζί με τη δέσμη ενεργειών.

Προσθέστε τα ακόλουθα κάτω από τις δηλώσεις εισαγωγής :

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

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

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

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

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

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

        # Map refresh
        self.refresh_interval_ms = 5000

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

Ορισμός του αναγνωριστικού χώρου εργασίας με χρήση μεταβλητής περιβάλλοντος

Αντί να κωδικοποιεί το αναγνωριστικό χώρου εργασίας απευθείας στο σενάριο, αυτό το σεμινάριο το διαβάζει από μια μεταβλητή περιβάλλοντος. Αυτό διατηρεί τις τιμές που αφορούν το περιβάλλον εκτός του πηγαίου κώδικα και σας επιτρέπει να επαναχρησιμοποιήσετε τη δέσμη ενεργειών σε χώρους εργασίας ή μηχανήματα χωρίς να την επεξεργαστείτε.

Πριν εκτελέσετε το σενάριο, δημιουργήστε μια μεταβλητή περιβάλλοντος με το όνομα FABRIC_WORKSPACE_ID.

Ορισμός της μεταβλητής περιβάλλοντος στα Windows

Στα Windows, μπορείτε να ορίσετε τη μεταβλητή από οποιοδήποτε τερματικό που υποστηρίζει μεταβλητές περιβάλλοντος—PowerShell, Windows PowerShell, PowerShell ή Γραμμή εντολών windows που είναι ενσωματωμένα στο Visual Studio και το Visual Studio Code, το Τερματικό Windows και πολλά άλλα άλλα κοχύλια.

Εκτελέστε τα ακόλουθα στο PowerShell ή στο ενσωματωμένο τερματικό VS Code:

$env:FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Για να επιβεβαιώσετε ότι έχει οριστεί η μεταβλητή:

echo $env:FABRIC_WORKSPACE_ID

Με αυτόν τον τρόπο ορίζεται η μεταβλητή μόνο για την τρέχουσα περίοδο λειτουργίας τερματικού.

Ορισμός μόνιμης μεταβλητής περιβάλλοντος (Windows)

Για να είναι διαθέσιμη η μεταβλητή σε μελλοντικές συνεδρίες, χρησιμοποιήστε ένα από τα εξής:

• PowerShell (one-liner): Εκτέλεση setx FABRIC_WORKSPACE_ID "<WORKSPACE_ID>"αρχείου . Η setx εντολή εγγράφεται στο περιβάλλον χρήστη αλλά δεν ενημερώνει το τρέχον τερματικό—κλείστε και ανοίξτε ξανά το τερματικό (ή ανοίξτε ένα νέο) πριν εκτελέσετε το σενάριο.
• GUI:
  1. Ανοίξτε τις Ιδιότητες συστήματος.
  2. Επιλέξτε Ρυθμίσεις συστήματος για προχωρημένους.
  3. Επιλέξτε Μεταβλητές περιβάλλοντος.
  4. Στην περιοχή Μεταβλητές χρήστη, επιλέξτε Νέα.
  5. Εισαγάγετε:
     • Όνομα: FABRIC_WORKSPACE_ID
     • Τιμή: το αναγνωριστικό του χώρου εργασίας σας
  6. Επιλέξτε ΟΚ για αποθήκευση.
  7. Κλείστε και ανοίξτε ξανά το τερματικό σας πριν εκτελέσετε ξανά το σενάριο.

Ορισμός της μεταβλητής περιβάλλοντος σε macOS ή Linux

Σε macOS και Linux, μπορείτε να ορίσετε τη μεταβλητή από οποιοδήποτε κέλυφος που υποστηρίζει export—Bash, Zsh (η προεπιλογή στο σύγχρονο macOS), Fish (με ελαφρώς διαφορετική σύνταξη) και τα ενσωματωμένα τερματικά στο Visual Studio Code και σε άλλους επεξεργαστές.

Τρέξιμο:

export FABRIC_WORKSPACE_ID="<WORKSPACE_ID>"

Για να επιβεβαιώσετε ότι έχει οριστεί η μεταβλητή:

echo $FABRIC_WORKSPACE_ID

Με αυτόν τον τρόπο ορίζεται η μεταβλητή μόνο για την τρέχουσα συνεδρία κελύφους.

Ορίστε μια μόνιμη μεταβλητή περιβάλλοντος (macOS ή Linux)

Για να κάνετε τη μεταβλητή διαθέσιμη σε μελλοντικές συνεδρίες, προσθέστε τη export γραμμή στο προφίλ του κελύφους σας:

• Zsh (προεπιλογή στο macOS): ~/.zshrc
• Bash: ~/.bashrc (Linux) ή ~/.bash_profile (macOS)
• Fish: εκτέλεση set -Ux FABRIC_WORKSPACE_ID "<WORKSPACE_ID>" αντί για επεξεργασία αρχείου

Μετά την ενημέρωση του προφίλ, είτε ανοίξτε ένα νέο τερματικό είτε εκτελέστε source ~/.zshrc (ή το κατάλληλο αρχείο), ώστε η αλλαγή να τεθεί σε ισχύ.

Βήμα 4—Προσθήκη βοηθητικών λειτουργιών

Σε αυτό το βήμα, συνυπολογίζετε τις οριζόντιες ανησυχίες—έλεγχος ταυτότητας, κατασκευή κεφαλίδας, polling μακροχρόνιας λειτουργίας και λογική επανάληψης—σε ένα μικρό σύνολο επαναχρησιμοποιήσιμων βοηθών που μπορεί να καλέσει κάθε συνάρτηση βήματος.

Η συγκέντρωση αυτών των ανησυχιών στους βοηθούς - αντί να τις εντάσσετε σε κάθε τοποθεσία κλήσης - σας δίνει τρία συγκεκριμένα πλεονεκτήματα:

• Μοναδική πηγή αλήθειας για οριζόντιες ανησυχίες: Ο έλεγχος ταυτότητας, οι κεφαλίδες και η ψηφοφορία LRO απαιτούνται σχεδόν σε κάθε κλήση API. Η συγκέντρωσή τους διατηρεί κάθε λειτουργία βήματος εστιασμένη στον δικό της πόρο αντί να εφαρμόζει εκ νέου τη λογική απόκτησης διακριτικών και επανάληψης.
• Ανθεκτικότητα χωρίς ακαταστασία: Οι βοηθοί απορροφούν παροδικές συνθήκες—ασύγχρονη παροχή, καθυστέρηση μετάδοσης παρασκηνίου, αποτυχίες αποστολής με δυνατότητα επανάληψης—έτσι ώστε οι συναρτήσεις βημάτων να παραμένουν σύντομες και να διαβάζονται σαν λίστα ελέγχου.
• Πιο εύκολο στη διδασκαλία και την τροποποίηση: Κάθε βοηθός εισάγεται μία φορά και επαναχρησιμοποιείται. Εάν το Fabric αλλάξει ένα μοτίβο LRO ή ένα εύρος ελέγχου ταυτότητας, το διορθώνετε σε ένα σημείο.

Οι βοηθοί που προσθέτετε σε αυτό το βήμα είναι:

• Βοηθοί ελέγχου ταυτότητας: δημιουργία κεφαλίδων για Fabric API REST (και Power BI τελικά σημεία LRO συμπλέγματος)
• FabricClient: ελαφρύ περιτύλιγμα για συνεπείς κλήσεις API
• LRO handler: poll μακροχρόνιες λειτουργίες χρησιμοποιώντας Location / x-ms-operation-id / Retry-After, συμπεριλαμβανομένων των αποκρίσεων 200-with-Running, των τελικών σημείων συμπλέγματος Power BI και των ωφέλιμων φορτίων ολοκλήρωσης μόνο κατάστασης (επιλύεται από το displayName)
• Βοηθός ωφέλιμου φορτίου ορισμού: κωδικοποίηση base64 map.json για ενσωματωμένους ορισμούς
• Βοηθός σύνδεσης ροής συμβάντων: ζητά την προσαρμοσμένη συμβολοσειρά σύνδεσης τελικού σημείου
• Βοηθός σπόρων: στέλνει τα αρχικά συμβάντα με λογική επανάληψης για να διασφαλίσει την επιτυχία της κατάποσης
• KQL βοηθός ετοιμότητας βάσης δεδομένων: περιμένει να γίνει διαθέσιμη η βάση δεδομένων KQL στους Χάρτες Fabric

Δημιουργία βοηθητικών λειτουργιών ελέγχου ταυτότητας

Κάθε Fabric κλήση REST που πραγματοποιεί αυτό το σεμινάριο φέρει ένα διακριτικό πρόσβασης Microsoft Entra (διακριτικό κομιστή) στην κεφαλίδα Authorization. Αντί να αποκτά διακριτικά ad hoc, αυτό το βήμα τυλίγεται DefaultAzureCredential σε ένα μικρό TokenProvider και εκθέτει ένα πρόγραμμα δημιουργίας κεφαλίδων για συγκεκριμένο κοινό για κάθε οικογένεια τελικού σημείου που καλεί το σενάριο.

Η συγκέντρωση της απόκτησης διακριτικών και της κατασκευής κεφαλίδων σε βοηθούς — αντί να αποκτάτε διακριτικά σε κάθε τοποθεσία κλήσης — σας δίνει τρία συγκεκριμένα πλεονεκτήματα:

• Κεντρικό διαπιστευτήριο: Ένα μεμονωμένο DefaultAzureCredential τυλίγεται στο TokenProvider και επαναχρησιμοποιείται για κάθε κλήση API, επομένως η ανακάλυψη ταυτότητας (Azure CLI, VS Code, διαχειριζόμενη ταυτότητα κ.λπ.) γίνεται μία φορά.
• Διακριτικά με επίγνωση κοινού: Τα τελικά σημεία συμπλέγματος Fabric, Kusto και Power BI απορρίπτουν διακριτικά που έχουν εκδοθεί για λάθος κοινό. Ένα ξεχωριστό εργαλείο δημιουργίας κεφαλίδων ανά κοινό διατηρεί το σωστό εύρος ακριβώς δίπλα στον ιστότοπο κλήσης, επομένως είναι προφανές σε ποιο τελικό σημείο στοχεύει κάθε συνάρτηση.
• Φρέσκο σε κάθε αίτημα: Οι δημιουργοί κεφαλίδων κατασκευάζουν την Authorization κεφαλίδα κατ' απαίτηση αντί να αποθηκεύουν οι ίδιοι το διακριτικό στην προσωρινή μνήμη. Τα υποκείμενα διαπιστευτήρια ανανεώνονται με διαφάνεια, επομένως οι ιστότοποι κλήσεων δεν χρειάζεται ποτέ να σκεφτούν τη λήξη.

Αυτό το εκπαιδευτικό βοήθημα καλεί Fabric API REST χρησιμοποιώντας ανατεθειμένα εύρη όπως Item.ReadWrite.All.

Προσθέστε τα ακόλουθα μετά το Config μάθημα:

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

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

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


_tokens = TokenProvider()


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

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


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


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

Δημιουργία περιτυλίγματος προγράμματος-πελάτη Fabric

Οι περισσότερες κλήσεις REST Fabric σε αυτό το σεμινάριο στέλνουν τις ίδιες κεφαλίδες Authorization και Content-Type. Αντί να τα επαναλαμβάνει σε κάθε τοποθεσία κλήσης, αυτό το σεμινάριο τυλίγεται httpx.Client σε ένα μικρό FabricClient που επισυνάπτει αυτόματα τις κεφαλίδες ενώ εξακολουθεί να επιστρέφει το ακατέργαστο httpx.Response , ώστε κάθε καλών να μπορεί να επιθεωρήσει τους κωδικούς κατάστασης (για παράδειγμα, για διάκριση 201 από 202το ).

Το περιτύλιγμα με αυτόν τον τρόπο — αντί να περνάτε httpx.Clientheaders=_fabric_headers() σε κάθε τοποθεσία κλήσης — σας δίνει δύο συγκεκριμένα πλεονεκτήματα:

• Κεφαλίδες σε ένα μέρος: Κάθε τοποθεσία κλήσης λαμβάνει αυτόματα την πιο πρόσφατη _fabric_headers() , επομένως δεν είναι δυνατή η κατά λάθος αποστολή μιας νέας αίτησης χωρίς την Authorization κεφαλίδα.
• Οι κωδικοί κατάστασης παραμένουν ορατοί: request() επιστρέφει το ακατέργαστο httpx.Response αντί για το αποκωδικοποιημένο JSON, έτσι ώστε οι ιστότοποι κλήσεων να μπορούν ακόμα να διακλαδίζονται στην κατάσταση (201 vs 202) και να επιθεωρούν κεφαλίδες όπως Location ή Retry-After για χειρισμό LRO.

Προσθέστε τα ακόλουθα μετά τις λειτουργίες του βοηθού ελέγχου ταυτότητας:

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

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

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

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

Δημιουργία βοηθητικής συνάρτησης LRO

Πολλά Fabric API REST που χρησιμοποιούνται σε αυτό το εκπαιδευτικό βοήθημα — όπως η Δημιουργία Eventhouse, η Δημιουργία ροής συμβάντων και η Δημιουργία χάρτη — υποστηρίζουν μακροχρόνιες λειτουργίες (LRO).

Αυτά τα API μπορούν να επιστρέψουν απαντήσεις σε διάφορα μοτίβα:

• 201 Created με το σώμα του πόρου ενσωματωμένο (σύγχρονο)
• 202 Accepted με μια Location κεφαλίδα που δείχνει σε μια διεύθυνση URL κατάστασης λειτουργίας (ασύγχρονη)
• 202 Accepted x-ms-operation-id με κεφαλίδα αντί για Location (ασύγχρονη, εναλλακτική μορφή)
• 200 OK με status: "Running" ή status: "NotStarted" κατά τη διάρκεια της ψηφοφορίας (ακόμα σε εξέλιξη)
• 200 OKμε status: "Succeeded" αλλά χωρίς αναγνωριστικό πόρου στο σώμα (επιτυχής, επίλυση με καταχώριση και αντιστοίχιση)displayName

Για να χειριστείτε όλα αυτά με συνέπεια, δημιουργείτε μια μοναδική βοηθητική συνάρτηση που:

1. Επιστρέφει αμέσως το αναγνωριστικό πόρου εάν η αρχική απόκριση το περιέχει ήδη.
2. Διαφορετικά, σταθμίζει τη διεύθυνση URL της λειτουργίας (που δημιουργήθηκε από ένα ή Locationx-ms-operation-id) χρησιμοποιώντας Retry-After.
3. Αντιμετωπίζει 200 OK με status: "Running" / "NotStarted" όπως είναι ακόμα σε εξέλιξη και συνεχίζει τη δημοσκόπηση.
4. Σε περίπτωση επιτυχίας, επιστρέφει το αναγνωριστικό πόρου από το σώμα ή επιστρέφει στην καταχώριση πόρων και στην αντιστοίχιση κατά displayName (με επαναλήψεις) όταν το σώμα είναι μόνο για κατάσταση.
5. Χρησιμοποιεί _pbi_headers() όταν η διεύθυνση URL ανίχνευσης βρίσκεται σε ένα σύμπλεγμα Power BI (*.analysis.windows.net) και Fabric κεφαλίδες διαφορετικά.

Αυτός ο μοναδικός βοηθός αντικαθιστά την ανάγκη για βοηθούς "επίλυσης με όνομα" ανά πόρο — κάθε create_* συνάρτηση σε αυτό το σεμινάριο καλεί _handle_lro με την κατάλληλη list_url και match_display_name.

Προσθέστε τα ακόλουθα μετά το FabricClient μάθημα:

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

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

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

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

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

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

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

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

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

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

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

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

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

Βοηθός ωφέλιμου φορτίου ορισμού

Όταν δημιουργείτε έναν χάρτη με δημόσιο ορισμό, το Create map REST API αναμένει ότι κάθε τμήμα θα definition.parts φέρει ένα ωφέλιμο φορτίο με κωδικοποίηση base64 με "payloadType": "InlineBase64". Ο βοηθός _json_to_b64 κωδικοποιεί ένα Python dict (το map.json) σας σε αυτήν τη μορφή, ώστε το create_map να μπορεί να το ρίξει κατευθείαν στο σώμα του αιτήματος.

Προσθέστε τα ακόλουθα μετά τη _handle_lro λειτουργία:

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

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

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

Δημιουργία βοηθού για την παροχή συμβολοσειρά σύνδεσης eventstream

Για να στείλετε συμβάντα στο προσαρμοσμένο τελικό σημείο της ροής συμβάντων, η δέσμη ενεργειών χρειάζεται ένα συμβολοσειρά σύνδεσης για αυτό το τελικό σημείο.

Σε αντίθεση με τα Fabric API REST που καλείτε μέχρι στιγμής (τα οποία είναι λειτουργίες επιπέδου ελέγχου για τη δημιουργία και τη διαχείριση πόρων), η πρόσληψη ροής συμβάντων χρησιμοποιεί ένα τελικό σημείο επιπέδου δεδομένων συμβατό με Κέντρα συμβάντων και αυτό το τελικό σημείο πραγματοποιεί έλεγχο ταυτότητας με ένα συμβολοσειρά σύνδεσης που βασίζεται σε SAS και όχι με ένα διακριτικό Microsoft Entra. Η συμβολοσειρά σύνδεσης δημιουργείται όταν προσθέτετε την προσαρμοσμένη προέλευση τελικού σημείου και δεν εκτίθεται από το Fabric REST API, επομένως πρέπει να αντιγραφεί από την πύλη Fabric.

get_eventhub_connection_string_interactive είτε επαναχρησιμοποιεί μια τιμή από τη μεταβλητή EVENTHUB_CONNECTION_STRING περιβάλλοντος (χρήσιμη σε επαναλαμβανόμενες εκτελέσεις) είτε σας ζητά να την αναζητήσετε κατά το χρόνο εκτέλεσης και, στη συνέχεια, την cfg αποθηκεύει στην προσωρινή μνήμη, ώστε τα επόμενα βήματα να μπορούν να την επαναχρησιμοποιήσουν χωρίς να σας ζητηθεί ξανά.

Προσθέστε τα ακόλουθα μετά τη _json_to_b64 λειτουργία:

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

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

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

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

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

    return cfg.eventhub_connection_string

Δημιουργία βοηθού για τη δημιουργία αρχικών συμβάντων από CSV

Για να διασφαλιστεί ότι ο χάρτης εμφανίζει δεδομένα αμέσως μετά τη δημιουργία του, το σενάριο στέλνει ένα μικρό σύνολο αρχικών συμβάντων στη ροή συμβάντων πριν από τη δημιουργία του χάρτη.

Χωρίς αυτό το βήμα, ο πίνακας Eventhouse ενδέχεται να μην περιέχει ακόμα δεδομένα και ο χάρτης μπορεί να εμφανίζεται κενός κατά την πρώτη φόρτωση.

Αυτή η βοηθητική συνάρτηση διαβάζει δεδομένα από ένα τοπικό αρχείο CSV και στέλνει κάθε γραμμή ως συμβάν JSON στη ροή συμβάντων χρησιμοποιώντας το πρωτόκολλο EventHub.

Επειδή οι πόροι Eventstream παρέχονται ασύγχρονα, το προσαρμοσμένο τελικό σημείο ενδέχεται να μην είναι άμεσα έτοιμο να δεχτεί συμβάντα μετά τη δημιουργία. Για να το χειριστείτε αυτό, η βοηθητική συνάρτηση περιλαμβάνει ενσωματωμένη λογική επανάληψης που επιχειρεί αυτόματα να στείλει συμβάντα μέχρι να είναι διαθέσιμο το τελικό σημείο. Αυτό διασφαλίζει ότι η διαδικασία προετοιμασίας είναι αξιόπιστη και επαναλαμβανόμενη και δεν απαιτεί χειροκίνητες ρυθμίσεις χρονισμού.

Αυτή η προσέγγιση αντικατοπτρίζει τα μοτίβα κατάποσης του πραγματικού κόσμου:

• Τα δεδομένα παράγονται εξωτερικά (για παράδειγμα, συσκευές ή εφαρμογές IoT)
• Τα συμβάντα μεταδίδονται σε ροή στο Eventstream
• Το Eventstream παρέχει δεδομένα στο Eventhouse για υποβολή ερωτημάτων και απεικόνιση

Με τη σπορά αρχικών συμβάντων, προσομοιώνετε αυτήν τη ροή πρόσληψης και διασφαλίζετε ότι:

• Συμπληρώνεται ο πίνακας προορισμού
• Η συνάρτηση KQL έχει δεδομένα για επιστροφή
• Ο χάρτης αποδίδεται αμέσως μετά τη δημιουργία

Προσθέστε τον ακόλουθο κώδικα μετά τη get_eventhub_connection_string_interactive() συνάρτηση:

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

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

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

Περιμένετε τη διαθεσιμότητα της βάσης δεδομένων KQL

Όταν υπάρχει η βάση δεδομένων, η βάση δεδομένων KQL και η συνάρτηση KQL, η βάση δεδομένων KQL ενδέχεται να μην είναι άμεσα επιλύσιμη από άλλα τελικά σημεία Fabric REST. Οι υπηρεσίες Fabric εκτελούνται σε κατανεμημένα backend, επομένως ένας πόρος που δημιουργήθηκε πρόσφατα μπορεί να χρειαστεί λίγο χρόνο για να μεταδοθεί σε αυτά.

Εάν καλέσετε τη Δημιουργία αντιστοίχισης αμέσως μετά την εγκατάσταση της συνάρτησης KQL, η Δημιουργία αντιστοίχισης ενδέχεται να αποτύχει να επιλύσει την προέλευση δεδομένων και να επιστρέψει ένα σφάλμα όπως Η βάση δεδομένων Kusto δεν βρέθηκε.

Το wait_for_kql_database_ready ανιχνεύει το τελικό σημείο REST Fabric για τη βάση δεδομένων KQL και επιστρέφει μόλις απαντήσει 200 OK. Είναι μια πύλη βέλτιστης προσπάθειας — μια επιτυχημένη απόκριση σε αυτό το τελικό σημείο επιπέδου ελέγχου είναι ένας ισχυρός διακομιστής μεσολάβησης που οι Χάρτες μπορούν επίσης να επιλύσουν τη βάση δεδομένων — και αυξάνεται RuntimeError μετά εάν max_attempts η βάση δεδομένων δεν γίνει ποτέ ορατή.

Προσθέστε τα ακόλουθα μετά τη seed_eventstream_from_csv λειτουργία:

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

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

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

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

Δημιουργία κύριων συναρτήσεων

Στη συνέχεια, προσθέτετε τις κύριες συναρτήσεις που καθορίζουν τη ροή εργασίας. Όλα αυτά ονομάζονται από main().

Οι συναρτήσεις προστίθενται με τη σειρά που ορίζονται στον κώδικα. main() τα καλεί με ελαφρώς διαφορετική σειρά, ώστε ο πίνακας KQL να υπάρχει πριν συνδεθεί η ροή συμβάντων σε αυτόν και έτσι τα δεδομένα σποράς να είναι διαθέσιμα πριν από την εκτέλεση της επαλήθευσης.

• Δημιουργία υπηρεσίας συμβάντων
• Δημιουργία του πίνακα KQL
• Επαλήθευση κατάποσης (ονομάζεται μετά τη σπορά)
• Δημιουργία ροής συμβάντων
• Δημιουργία συνάρτησης KQL
• Δημιουργία του ορισμού χάρτη (map.json)
• Δημιουργία των μεταδεδομένων πλατφόρμας (.platform)
• Δημιουργήστε τον χάρτη

Δημιουργία υπηρεσίας συμβάντων

create_eventhouse δημιουργεί μια βάση δεδομένων συμβάντων στον χώρο εργασίας και επιστρέφει το αναγνωριστικό στοιχείου της. Το Create Eventhouse REST API μπορεί να απαντήσει στην ίδια κλήση με τρεις διαφορετικούς τρόπους:

• 201 Created με το ενσωματωμένο αναγνωριστικό αποθήκης συμβάντων (σύγχρονο).
• 202 Accepted με URL λειτουργίας LRO (ασύγχρονη).
• 409 Conflict με x-ms-public-api-error-code οριστεί σε ένα από τα δύο ItemDisplayNameNotAvailableYet (το προηγούμενο όνομα εξακολουθεί να είναι δεσμευμένο στο παρασκήνιο) ή ItemDisplayNameAlreadyInUse (μια αποθήκη συμβάντων με αυτό το όνομα υπάρχει ήδη στον χώρο εργασίας).

Για να χειριστείτε και τα τρία αξιόπιστα: create_eventhouse

• Εκπρόσωποι 201 και 202 απαντήσεις στο _handle_lro, το οποίο καλύπτει ήδη ομοιόμορφα τη σύγχρονη και την ασύγχρονη ολοκλήρωση.
• Τιμές Retry-After και επαναλήψεις (έως πέντε προσπάθειες) σε ItemDisplayNameNotAvailableYet.
• Επαναχρησιμοποιεί την υπάρχουσα βάση συμβάντων ενεργοποιημένη ItemDisplayNameAlreadyInUse καταχωρώντας τις μονάδες συμβάντων στον χώρο εργασίας και αντιστοιχίζοντας με displayName.
• Επιστρέφει σε ένα μονοσήμαντο εμφανιζόμενο όνομα (με κατάληξη ένα σύντομο UUID) εάν το όνομα δεν γίνει ποτέ διαθέσιμο μετά την εξάντληση του προϋπολογισμού επανάληψης, ώστε το σενάριο να μπορεί να συνεχίσει να προχωρά.

Προσθέστε τα ακόλουθα μετά τη wait_for_kql_database_ready λειτουργία:

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

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

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

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

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

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

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

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

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

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

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

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

Δημιουργία πίνακα KQL

create_kql_table_if_missing Εξασφαλίζει ότι ο πίνακας προορισμού υπάρχει στη βάση δεδομένων KQL πριν ξεκινήσει η εγγραφή της ροής συμβάντων σε αυτόν. Ο προορισμός ροής συμβάντων που δημιουργείτε αργότερα έχει ρυθμιστεί με ProcessedIngestion και ένα σταθερό tableName, επομένως ο πίνακας πρέπει να υπάρχει ήδη κατά την άφιξη συμβάντων — διαφορετικά η πρόσληψη αποτυγχάνει.

Η συνάρτηση εκδίδει μια .create-merge table εντολή στο τελικό σημείοqueryServiceUri + /v1/rest/mgmt διαχείρισης Kusto (). .create-merge είναι ανίκανο: δημιουργεί τον πίνακα εάν δεν υπάρχει και συγχωνεύει το σχήμα εάν υπάρχει. Αυτό καθιστά ασφαλή την κλήση σε κάθε τρέξιμο.

Πριν από την έκδοση της εντολής, η συνάρτηση διαβάζει τις ιδιότητες της βάσης συμβάντων που θα λάβει queryServiceUri και το αναγνωριστικό στοιχείου βάσης δεδομένων KQL και, στη συνέχεια, επιλύει τη βάση δεδομένων displayName , ώστε το ωφέλιμο mgmt φορτίο να την αναφέρει με το όνομα και όχι με το αναγνωριστικό.

Προσθέστε τα ακόλουθα μετά τη create_eventhouse λειτουργία:

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

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

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

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

    kql_database_item_id = databases_item_ids[0]

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

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

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

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

Επαλήθευση πρόσληψης δεδομένων

verify_eventhouse_data επιβεβαιώνει ότι τα συμβάντα που έχουν τοποθετηθεί στη ροή συμβάντων προσγειώθηκαν στην πραγματικότητα στον πίνακα eventhouse. Σταθμίζει ένα <table> | count ερώτημα στο τελικό σημείο ερωτήματος Kusto (queryServiceUri + /v1/rest/query) του eventhouse έως ότου το πλήθος που επιστρέφεται είναι μεγαλύτερο από μηδέν ή αποτύχει μετά από ένα χρονικό όριο. Η απορρόφηση ροής συμβάντων διαρκεί μερικά δευτερόλεπτα για να μεταδοθεί από το προσαρμοσμένο τελικό σημείο στον πίνακα, επομένως η ανίχνευση — και όχι ένα μεμονωμένο ερώτημα — είναι αυτό που δίνει ένα σίγουρο σήμα επιτυχίας/αποτυχίας.

Ορίζεται δίπλα στο create_kql_table_if_missing επειδή και οι δύο βοηθοί αναζητούν τις ίδιες ιδιότητες του eventhouse (queryServiceUri, databasesItemIds) και επιλύουν τη βάση δεδομένων displayNameKQL . Καλείται main(), ώστε τα seeded events να έχουν την ευκαιρία να ρέουν μέσα από seed_eventstream_from_csv το eventstream και να φτάσουν στο τραπέζι πριν ξεκινήσει η καταμέτρηση.

Η εκτέλεση αυτού του ελέγχου εκ των προτέρων εντοπίζει νωρίς την εσφαλμένη ρύθμιση παραμέτρων της πρόσληψης — για παράδειγμα, έναν προορισμό ροής συμβάντων που συνδέεται με λάθος όνομα πίνακα — αντί να τον αφήσετε να εμφανιστεί αργότερα ως κενός χάρτης.

Προσθέστε τα ακόλουθα μετά τη create_kql_table_if_missing λειτουργία:

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

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

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

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

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

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

    db_id = db_ids[0]

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

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

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

    max_attempts = 12
    delay_seconds = 5

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

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

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

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

        if attempt < max_attempts:
            time.sleep(delay_seconds)

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

Δημιουργία ροής συμβάντων με ορισμό

create_eventstream_with_definition δημιουργεί μια ροή συμβάντων στον χώρο εργασίας με την πλήρη τοπολογία της ενσωματωμένη στην αίτηση και, στη συνέχεια, επιστρέφει το αναγνωριστικό στοιχείου της ροής συμβάντων. Η χρήση ενός δημόσιου ορισμού σάς επιτρέπει να παρέχετε τη ροή συμβάντων και να καλωδιώνετε τις πηγές, τις ροές και τους προορισμούς της σε μία μόνο κλήση, αντί να δημιουργείτε πρώτα τη ροή συμβάντων και μετά να επιδιορθώνετε τον ορισμό της.

Πριν από την αποστολή της αίτησης, η συνάρτηση διαβάζει τις ιδιότητες της υπηρεσίας συμβάντων για να λάβει το αναγνωριστικό στοιχείου βάσης δεδομένων KQL και επιλύει το αναγνωριστικό της βάσης δεδομένων displayName , ώστε ο προορισμός να την αναφέρει με βάση το όνομα και όχι με το αναγνωριστικό. Στη συνέχεια, δημιουργεί ένα γράφημα ροής συμβάντων με μια CustomEndpoint πηγή, ένα DefaultStream, και έναν Eventhouse προορισμό διαμορφωμένο με ProcessedIngestion και το σταθερό tableName από cfg, base64-κωδικοποιεί το γράφημα ως eventstream.json τμήμα και το POST για να δημιουργήσει ροή συμβάντων.

Το Create Eventstream REST API μπορεί να απαντήσει με 201 Created (σύγχρονο, ενσωματωμένο σώμα), 202 Accepted (ασύγχρονο LRO μέσω Location ή x-ms-operation-id) ή 200 OK με ένα ωφέλιμο φορτίο ολοκλήρωσης μόνο κατάστασης, όπου η ροή συμβάντων δεν είναι ακόμα ορατή στην απόκριση Λίστα ροών συμβάντων λόγω καθυστέρησης μετάδοσης παρασκηνίου. _handle_lro καλύπτει όλες αυτές τις περιπτώσεις — συμπεριλαμβανομένης της καταχώρισης και της αντιστοίχισης κατά displayName — επομένως αυτή η συνάρτηση αναθέτει τον πλήρη χειρισμό απόκρισης σε αυτήν σε μία μόνο κλήση.

Προσθέστε τα ακόλουθα μετά τη verify_eventhouse_data λειτουργία:

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

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

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

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

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

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

    kql_database_item_id = databases_item_ids[0]

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

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

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

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

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

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

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

Δημιουργία συνάρτησης KQL

create_kql_function δημιουργεί (ή ενημερώνει) μια αποθηκευμένη συνάρτηση Kusto στη βάση δεδομένων KQL της βάσης δεδομένων KQL και επιστρέφει το αναγνωριστικό στοιχείου της βάσης δεδομένων KQL, ώστε ο καλών να μπορεί να συνδέσει την προέλευση δεδομένων του χάρτη σε αυτήν. Η συνάρτηση — LatestVehicleLocations από προεπιλογή — επιστρέφει την πιο πρόσφατη γραμμή ανά VehicleId μέσω arg_max(EventTime, *), προβάλλοντας Latitude, Longitude, VehicleId και EventTime ώστε οι Χάρτες Fabric να μπορούν να δεσμεύσουν τις στήλες γεωγραφικού πλάτους και μήκους του επιπέδου.

Όπως create_kql_table_if_missing, αυτός ο βοηθός εκτελείται ενάντια στο τελικό σημείο διαχείρισης Kusto του eventhouse (queryServiceUri + /v1/rest/mgmt) και είναι ανίκανος: .create-or-alter function δημιουργεί τη συνάρτηση εάν δεν υπάρχει και αντικαθιστά το σώμα του εάν υπάρχει, έτσι ώστε ο βοηθός να είναι ασφαλής να καλεί σε κάθε εκτέλεση.

Η εντολή αποστέλλεται με skipvalidation=true επειδή το σώμα της συνάρτησης αναφέρεται στον πίνακα προορισμού και table("<name>") όχι ως γυμνό αναγνωριστικό. Η φόρμα αναβάλλει την ανάλυση ονόματος στον χρόνο ερωτήματος, επομένως η table() επικύρωση κατά τη δημιουργία θα αποτύγχανε διαφορετικά εάν ο πίνακας δεν έχει λάβει ακόμη δεδομένα και το σχήμα του δεν είναι πλήρως ορατό στον επικυρωτή. Η skipvalidation=true σύζευξη με table("...") επιτρέπει τη δημιουργία της συνάρτησης πριν η πρόσληψη συμπληρώσει τον πίνακα, η οποία είναι η σειρά με την οποία εκτελείται αυτό το σεμινάριο.

Προσθέστε τα ακόλουθα μετά τη create_eventstream_with_definition λειτουργία:

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

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

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

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

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

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

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

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

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

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

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

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

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

Κατασκευάστε map.json

Το build_map_json δημιουργεί και επιστρέφει το φορτίο «map.json που καθορίζει τα περιεχόμενα του Fabric χάρτη. Το ωφέλιμο φορτίο ακολουθεί το σχήμα ορισμού στοιχείων χάρτη και αποτελείται από τέσσερις ενότητες: dataSources (από πού προέρχονται τα δεδομένα), iconSources (προαιρετικοί προσαρμοσμένοι δείκτες), layerSources (τι ερωτάται και πόσο συχνά) και layerSettings (πώς αποδίδεται το αποτέλεσμα στον χάρτη).

Για αυτό το σεμινάριο, dataSources δείχνει τη βάση δεδομένων KQL (itemType: "KqlDatabase") που δημιουργήθηκε νωρίτερα και η μοναδική καταχώρηση είναι layerSources ένα επίπεδο που υποστηρίζεται από Kusto (type: "kusto", queryType: "function") του οποίου query καλεί την αποθηκευμένη συνάρτηση LatestVehicleLocations(). refreshIntervalMs διαβάζεται από cfg.refresh_interval_ms (5000 ms από προεπιλογή), επομένως το επίπεδο εκτελεί ξανά τη λειτουργία σε ένα χρονόμετρο και ο χάρτης αντικατοπτρίζει τη νέα πρόσληψη σε σχεδόν πραγματικό χρόνο.

Η αντίστοιχη layerSettings καταχώριση συνδέει τις στήλες αποτελεσμάτων του επιπέδου με τον χάρτη μέσω latitudeColumnName: "Latitude" και longitudeColumnName: "Longitude", αποδίδει κάθε γραμμή ως bubble σημείο και επιφάνειες VehicleId και EventTime επεξηγήσεις εργαλείων. Η συνάρτηση εκτυπώνει το συναρμολογημένο ωφέλιμο φορτίο, ώστε να μπορείτε να επιθεωρήσετε το ακριβές JSON που στέλνει η κλήση Δημιουργία χάρτη.

Για περισσότερες πληροφορίες σχετικά με τον ορισμό χάρτη REST API, ανατρέξτε στην ενότητα Ορισμός στοιχείου χάρτη.

Προσθέστε τα ακόλουθα μετά τη create_kql_function λειτουργία:

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

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

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

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

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

        "iconSources": [],

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

            }
        ]
    }

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

Build .platform (μεταδεδομένα πλατφόρμας)

build_platform_json δημιουργεί και επιστρέφει ένα προαιρετικό .platform τμήμα που μπορεί να περιλαμβάνει η κλήση Δημιουργία αντιστοίχισης όταν map.json θέλετε να ορίσετε μη προεπιλεγμένα μεταδεδομένα στοιχείου στον χάρτη. Δεν απαιτείται η συμπερίληψη ενός τμήματος .platform — Fabric εφαρμόζει προεπιλεγμένα μεταδεδομένα όταν το τμήμα παραλείπεται — αλλά αυτό το σεμινάριο δείχνει πώς να συντάξετε ένα, ώστε να μπορείτε να χρησιμοποιήσετε ξανά το μοτίβο όταν χρειάζεστε ρητό έλεγχο του τύπου στοιχείου, του εμφανιζόμενου ονόματος, της περιγραφής ή ενός σταθερού λογικού αναγνωριστικού.

Το ωφέλιμο φορτίο ακολουθεί το σχήμα platform-properties και έχει δύο ενότητες: metadata (type: "Map", displayName, description) και config (version, logicalId). Το logicalId δημιουργείται ως ένα νέο UUID εδώ, το οποίο είναι εντάξει για μια δημιουργία μίας λήψης. Εάν σκοπεύουμε να αναπτύξουμε ξανά τον ίδιο χάρτη μέσω ενσωμάτωσης Git ή επαναλαμβανόμενων εκτελέσεων, καρφιτσώστε logicalId σε μια σταθερή τιμή, ώστε οι ενημερώσεις να στοχεύουν το ίδιο στοιχείο.

Για περισσότερες πληροφορίες, δείτε Αντιστοίχιση ορισμού στοιχείου και Επισκόπηση ορισμού στοιχείου.

Προσθέστε τα ακόλουθα μετά τη build_map_json λειτουργία:

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

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

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

Δημιουργία χάρτη με ενσωματωμένο ορισμό

create_map δημιουργεί τον χάρτη δημοσιεύοντας τον ενσωματωμένο ορισμό που έχετε συγκεντρώσει και επιστρέφει το αναγνωριστικό στοιχείου του νέου χάρτη. Το αίτημα φέρει τρία μέρη με κωδικοποίηση base64 κάτω από payloadType: "InlineBase64"το : map.json (ο απαιτούμενος ορισμός πυρήνα), τα προαιρετικά .platform μεταδεδομένα που δημιουργήσατε στο προηγούμενο βήμα και ένα αρχείο ερωτήματος Kusto με όνομα queries/layerSource-<layerSourceId>.kql που περιέχει την κλήση στην αποθηκευμένη συνάρτηση KQL. Η ομαδοποίηση και των τριών μερών σε μία κλήση παρέχει τον χάρτη και συνδέει το επίπεδο δεδομένων του στη λειτουργία KQL ατομικά, επομένως δεν απαιτείται περαιτέρω getDefinition / updateDefinition ταξίδι μετ' επιστροφής.

Το όνομα του αρχείου ερωτήματος έχει σημασία: Fabric επιλύει το ερώτημα ενός επιπέδου αντιστοιχίζοντας το queries/layerSource-<layerSourceId>.kql με το id της αντίστοιχης καταχώρησης στο layerSources, έτσι ώστε η συνάρτηση να τραβήξει το αναγνωριστικό πηγής επιπέδου από το map_json["layerSources"][0]["id"] για να κατασκευάσει τη διαδρομή. map.json και .platform κωδικοποιούνται με base64 μέσω _json_to_b64; το κείμενο του ερωτήματος κωδικοποιείται απευθείας με base64 επειδή είναι συμβολοσειρά και dictόχι .

Το Create Map REST API μπορεί να απαντήσει με 201 Created (σύγχρονο, ενσωματωμένο αναγνωριστικό), 202 Accepted (ασύγχρονο LRO μέσω Location ή x-ms-operation-id) ή 200 OK με ωφέλιμο φορτίο ολοκλήρωσης μόνο κατάστασης, όπου ο χάρτης δεν είναι ακόμη ορατός στους χάρτες λίστας λόγω καθυστέρησης διάδοσης του backend. _handle_lro καλύπτει όλες αυτές τις περιπτώσεις — συμπεριλαμβανομένης της καταχώρισης και της αντιστοίχισης κατά displayName — επομένως αυτή η συνάρτηση αναθέτει τον πλήρη χειρισμό απόκρισης σε αυτήν σε μία μόνο κλήση.

Για περισσότερες πληροφορίες, δείτε Ορισμός στοιχείου αντιστοίχισης.

Προσθέστε τα ακόλουθα μετά τη build_platform_json λειτουργία:

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


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

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

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

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

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

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

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

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

Ενορχηστρώστε τη ροή εργασίας

main είναι το μοναδικό σημείο εισόδου που εκτελεί το σεμινάριο από άκρο σε άκρο. Δημιουργεί Config, ανοίγει ένα httpx.Client επαναχρησιμοποιημένο σε κάθε βοηθό, το τυλίγει σε ένα FabricClient, και στη συνέχεια καλεί κάθε συνάρτηση βήματος με σειρά εξάρτησης: create_eventhouse → create_kql_table_if_missing (πρέπει να υπάρχει πριν συνδεθεί η ροή συμβάντων σε αυτήν) → create_eventstream_with_definition → seed_eventstream_from_csv → verify_eventhouse_data (πιάνει εσφαλμένη διαμόρφωση πρόσληψης πριν από οποιαδήποτε εργασία χάρτη) → create_kql_function → wait_for_kql_database_ready (πύλη βέλτιστης προσπάθειας, ώστε η Δημιουργία χάρτη να μπορεί να επιλύσει τη βάση δεδομένων KQL) → build_map_json → build_platform_json → create_map.

Η ταξινόμηση έχει σημασία επειδή τα περισσότερα βήματα καταναλώνουν κάτι που δημιουργήθηκε από ένα προηγούμενο βήμα — create_eventstream_with_definition χρειάζεται το όνομα της συνάρτησης KQL και databasesItemIds χρειάζεται το όνομα της συνάρτησης create_mapKQL και το αναγνωριστικό στοιχείου βάσης δεδομένων KQL. Το τελικό μπλοκ print εμφανίζει τα αναγνωριστικά κάθε πόρου που δημιουργείται, ώστε να μπορείτε να τα βρείτε στην πύλη Fabric.

Προσθέστε τα ακόλουθα μετά τη create_map λειτουργία:

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

if __name__ == "__main__":
    main()

Εκτελέστε την εφαρμογή

Κατά την εκτέλεση της δέσμης ενεργειών, σας ζητείται να επικολλήσετε τη συμβολοσειρά σύνδεσης string eventstream.

Για να ανακτήσετε αυτήν την τιμή:

1. Άνοιγμα του χώρου εργασίας Fabric
2. Άνοιγμα της ροής συμβάντων που δημιουργήθηκε από τη δέσμη ενεργειών
3. Επιλέξτε την προέλευση προσαρμοσμένου τελικού σημείου
4. Ανοίξτε τον έλεγχο ταυτότητας κλειδιού SAS
5. Αντιγραφή συμβολοσειράς σύνδεσης-πρωτεύον κλειδί

Επικολλήστε την τιμή στην κονσόλα όταν σας ζητηθεί.

Εκτελέστε το σενάριο:

python create_realtime_map.py

Βεβαιωθείτε ότι δημιουργήθηκαν όλα τα στοιχεία:

Σε αυτό το σημείο, δημιουργούνται και ρυθμίζονται όλοι οι πόροι.

Για να προσομοιώσετε τη συνεχή ροή και να παρακολουθήσετε την ενημέρωση του χάρτη σε σχεδόν πραγματικό χρόνο, συνεχίστε με την παρακολούθηση Tutorial: Προσομοίωση απορρόφησης δεδομένων σε πραγματικό χρόνο για έναν χάρτη χρησιμοποιώντας REST API και Python. Βασίζεται απευθείας σε αυτό το σεμινάριο και επαναχρησιμοποιεί το eventhouse, τη ροή συμβάντων, τη συνάρτηση KQL και τον χάρτη που δημιουργήσατε.

Σύνοψη

Σε αυτό το πρόγραμμα εκμάθησης, παρείχατε τους πόρους που απαιτούνται για μια γεωχωρική λύση σε πραγματικό χρόνο στο Microsoft Fabric, χρησιμοποιώντας Fabric REST API και Python.

Καταφέρατε τα εξής:

• Δημιουργήθηκε μια βάση δεδομένων eventhouse και KQL χρησιμοποιώντας το Fabric REST API
• Δημιουργία ροής συμβάντων με προσαρμοσμένο τελικό σημείο για πρόσληψη συμβάντων ροής
• Όρισε μια συνάρτηση KQL για την υποβολή ερωτημάτων και τη διαμόρφωση δεδομένων σε πραγματικό χρόνο για οπτικοποίηση χάρτη
• Δημιουργία και ανάπτυξη χάρτη Fabric με ενσωματωμένο ορισμό αναφορά σε δεδομένα βάσης συμβάντων
• Σπορά της ροής συμβάντων με αρχικά συμβάντα, ώστε ο χάρτης να εμφανίζει δεδομένα αμέσως

Αυτή η αρχιτεκτονική παρουσιάζει ένα κοινό μοτίβο ανάλυσης σε πραγματικό χρόνο στο Fabric:

• Εξωτερικοί παραγωγοί στέλνουν συμβάντα στο Eventstream
• Η ροή συμβάντων δρομολογεί και προσλαμβάνει δεδομένα στην αποθήκη συμβάντων
• Οι συναρτήσεις KQL μετασχηματίζουν τα δεδομένα
• Οι Χάρτες υποβάλλουν ερωτήματα στην υπηρεσία Eventhouse και ανανεώνονται αυτόματα για να αντικατοπτρίζουν νέα συμβάντα

Με την αυτοματοποίηση της δημιουργίας πόρων χρησιμοποιώντας Python και REST API, έχετε πλέον μια επαναλαμβανόμενη προσέγγιση για τη δημιουργία χωρικών εφαρμογών σε πραγματικό χρόνο χωρίς χειροκίνητη διαμόρφωση. Για να οδηγήσετε συνεχή δεδομένα στον χάρτη, συνεχίστε με το σεμινάριο προσομοιωτή παρακολούθησης.

Επόμενα βήματα

Τώρα που καταλαβαίνετε τη ροή από άκρο σε άκρο, μπορείτε να επεκτείνετε αυτήν τη λύση για να ενσωματώσετε έναν προσομοιωτή σε πραγματικό χρόνο.

Για ένα σεμινάριο που δείχνει τη δημιουργία ενός προσομοιωτή σε πραγματικό χρόνο για τον χάρτη που μόλις δημιουργήσατε χρησιμοποιώντας REST API, δείτε: