Υποβολή ερωτήματος στη ροή συμβάντων στο Microsoft Fabric από ένα σημειωματάριο με δομημένη ροή Spark

Το Eventstream στο Microsoft Fabric είναι μια πλήρως διαχειριζόμενη υπηρεσία πρόσληψης και ροής συμβάντων που επιτρέπει την επεξεργασία δεδομένων και την ανάλυση δεδομένων σε πραγματικό χρόνο. Μπορείτε να ενσωματώσετε τη ροή συμβάντων με σημειωματάρια Fabric χρησιμοποιώντας το Spark Structured Streaming για να επεξεργαστείτε και να αναλύσετε δεδομένα ροής σε πραγματικό χρόνο.

Αυτή η ενοποίηση σάς επιτρέπει να εξερευνάτε ροές συμβάντων και άλλες πηγές σε πραγματικό χρόνο μέσω του Real-Time hub, απευθείας από τα σημειωματάρια Fabric σας. Μπορείτε επίσης να δημιουργήσετε νέες ροές συμβάντων και να αρχίσετε να απορροφάτε δεδομένα από σχεδόν 30 (και αυξανόμενες) πηγές ροής, συμπεριλαμβανομένων βάσεων δεδομένων με δυνατότητα CDC, μεσιτών μηνυμάτων, υπηρεσιών ροής και δημόσιων ροών. Το νέο μοντέλο προγραμματισμού σάς επιτρέπει να συνδέεστε εύκολα σε ροή συμβάντων, να διαβάζετε δεδομένα ροής και να εκτελείτε αναλύσεις σε πραγματικό χρόνο χρησιμοποιώντας γνωστά API Spark χωρίς συμβολοσειρές σύνδεσης ή διαπιστευτήρια, καθιστώντας πιο ασφαλή τη δημιουργία εφαρμογών σε πραγματικό χρόνο

Προαπαιτούμενα στοιχεία

Πριν ξεκινήσετε, βεβαιωθείτε ότι έχετε τις ακόλουθες προϋποθέσεις:

  • Ένας χώρος εργασίας Fabric με ενεργοποιημένη τη ροή συμβάντων.
  • Ένα περιβάλλον σημειωματαρίου που έχει ρυθμιστεί στο Fabric.
  • Βασικές γνώσεις Spark Structured Streaming και PySpark ή Scala.

Βήματα για την υποβολή ερωτήματος στη ροή συμβάντων από ένα σημειωματάριο

  1. Ανακαλύψτε και προσθέστε ροή συμβάντων στο Σημειωματάριο:

    1. Ανοίξτε το σημειωματάριό σας Microsoft Fabric.

    2. Επιλέξτε Προσθήκη στοιχείων δεδομένων στην καρτέλα Στοιχεία δεδομένων και, στη συνέχεια, επιλέξτε Από Real-Time διανομέα από το αναπτυσσόμενο μενού.

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

    4. Προσθέστε τη ροή συμβάντων στο περιβάλλον σημειωματαρίου σας.

      Στιγμιότυπο οθόνης που δείχνει πώς μπορείτε να ανακαλύψετε τη ροή συμβάντων στο κέντρο Real-Time.

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

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

  2. Ανάγνωση δεδομένων ροής από τη ροή συμβάντων:

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

      Στιγμιότυπο οθόνης που δείχνει πώς μπορείτε να διαβάσετε τη ροή συμβάντων με το Spark από το μενού περιβάλλοντος.

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

      from pyspark.sql import SparkSession
from pyspark.sql.functions import col
from pyspark.sql.types import StringType
from pyspark.sql.dataframe import DataFrame

eventstream_options = {
    "eventstream.itemid": __in_eventstream_item_id,
    "eventstream.datasourceid": __in_eventstream_datasource_id
}

# Read from Kafka using the config map
df_raw = spark.readStream.format("kafka").options(**eventstream_options).load()

decoded_df = df_raw.select(
    col("key").cast(StringType()).alias("key"),
    col("value").cast(StringType()).alias("value"),
    col("partition"),
    col("offset")
)

def showDf(x:DataFrame, y:int):
    x.show()

# Print messages to the console
query = decoded_df.writeStream.foreachBatch(showDf).outputMode("append").start()
query.awaitTermination()

      Η τιμή των eventstream.itemid επιλογών και eventstream.datasourceid συμπληρώνεται αυτόματα στο κελί κωδικού παραμέτρου πάνω από αυτό το κελί κωδικού.

      Σημείωμα

      Για κάθε επιλογή στο Read with Spark, δημιουργείται ένα νέο κελί κώδικα με τις απαραίτητες παραμέτρους που συμπληρώνονται αυτόματα και δημιουργείται ένας νέος προορισμός Notebook για το στοιχείο Eventstream, όπου το eventstream.datasourceid είναι μοναδικό για κάθε προορισμό Notebook. Για ένα στοιχείο σημειωματαρίου, συνιστάται να το ορίσετε μία φορά ως προορισμό για ένα στοιχείο Eventstream για να αποφύγετε τη δημιουργία πολλών προορισμών.

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

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

  3. Προσθήκη σημειωματαρίου ως προορισμού: Μπορείτε επίσης να προσθέσετε το σημειωματάριο ως προορισμό για τη ροή συμβάντων για να ξεκινήσει η πρόσληψη δεδομένων στη ροή συμβάντων μέσα από το σημειωματάριο. Για περισσότερες πληροφορίες, ανατρέξτε στο θέμα Προσθήκη προορισμού σημειωματαρίου Spark σε μια ροή συμβάντων.

Συμπέρασμα

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

Σχετικό περιεχόμενο

Για να μάθετε περισσότερα σχετικά με την εργασία με ροές συμβάντων και σημειωματάρια στο Fabric, ανατρέξτε στους παρακάτω πόρους

  • Τελευταία ενημέρωση στις