Daten ingestieren mit Notebooks

Notizbücher in Azure Databricks bieten einen flexiblen, codegesteuerten Ansatz zum Aufnehmen von Daten aus verschiedenen Quellen. Wenn grafische Tools Ihre Anforderungen nicht erfüllen oder wenn Sie benutzerdefinierte Logik für die Datentransformation benötigen, bieten Notebooks die vollständige Kontrolle über den Aufnahmeprozess. Sie können Python (PySpark), SQL, Scala oder R verwenden, um eine Verbindung mit Datenquellen herzustellen, Transformationen anzuwenden und Daten in Unity-Katalogtabellen zu laden.

In dieser Lektion erfahren Sie, wie Sie Daten mithilfe von Notizbüchern für Batch- und Streamingszenarien aufnehmen.

Batchdaten mit DataFrames aufnehmen

DataFrames sind die Grundlage für die Batchdatenaufnahme in Databricks. Ein DataFrame ist eine verteilte Sammlung von Daten, die in benannte Spalten organisiert sind, ähnlich einer Tabelle in einer relationalen Datenbank. Sie verwenden die spark.read API, um Daten aus verschiedenen Quellen in einen DataFrame zu laden.

Im folgenden Beispiel wird eine CSV-Datei in einem DataFrame gelesen:

df = (spark.read
    .format("csv")
    .option("header", "true")
    .option("inferSchema", "true")
    .load("/Volumes/catalog/schema/volume/data.csv"))

Die format() Methode gibt den Dateityp an, während option() Methoden konfigurieren, wie die Daten analysiert werden. Die header Option weist Spark an, die erste Zeile als Spaltennamen zu verwenden, und inferSchema erkennt automatisch Datentypen.

Lesen aus verschiedenen Dateiformaten

Azure Databricks unterstützt mehrere Dateiformate. Sie können das Format und die Optionen basierend auf Ihrer Datenquelle anpassen:

# Read JSON files
df_json = spark.read.format("json").load("/path/to/data.json")

# Read Parquet files (columnar format, efficient for analytics)
df_parquet = spark.read.format("parquet").load("/path/to/data.parquet")

# Read XML files (requires rowTag option)
df_xml = (spark.read
    .format("xml")
    .option("rowTag", "record")
    .load("/path/to/data.xml"))

<data>
  <record><id>1</id><name>Alice</name></record>
  <record><id>2</id><name>Bob</name></record>
</data>

Parkettdateien speichern Daten in einem spaltenaren Format, wodurch die Abfrageleistung für analytische Workloads verbessert wird. JSON-Dateien unterstützen geschachtelte Strukturen und sind für API-Antworten üblich. XML-Dateien erfordern die rowTag Option, um anzugeben, welches Element jeder Zeile zugeordnet ist.

Herstellen einer Verbindung mit externen Datenbanken

Verwenden Sie für Daten, die in relationalen Datenbanken gespeichert sind, JDBC-Verbindungen, um direkt aus der Quelle zu lesen.

df_jdbc = (spark.read
    .format("jdbc")
    .option("url", "jdbc:sqlserver://server.database.windows.net:1433;database=mydb")
    .option("dbtable", "schema.tablename")
    .option("user", dbutils.secrets.get(scope="jdbc", key="username"))
    .option("password", dbutils.secrets.get(scope="jdbc", key="password"))
    .load())

Speichern Sie Anmeldeinformationen in Geheimschlüsseln von Databricks, anstatt sie in Notizbüchern zu codieren. Sie können Abfragen auch mithilfe einer Unterabfrage in der dbtable-Option an die Datenbank übertragen.

pushdown_query = "(SELECT id, name, amount FROM orders WHERE order_date > '2024-01-01') AS filtered_orders"
df = spark.read.format("jdbc").option("dbtable", pushdown_query).load()

Schreiben von Daten in Unity-Katalogtabellen

Nachdem Sie Ihre Daten geladen und transformiert haben, schreiben Sie sie in eine Unity-Katalogtabelle. Azure Databricks verwendet standardmäßig das Delta Lake-Format, das ACID-Transaktionen und Versionsverwaltung bereitstellt:

df.write.mode("overwrite").saveAsTable("catalog.schema.table_name")

Die mode() Option steuert, wie vorhandene Daten behandelt werden:

• überschreiben: Ersetzt den Tabelleninhalt vollständig.
• append: Fügt der vorhandenen Tabelle Zeilen hinzu
• ignore: Führt nichts aus, wenn die Tabelle bereits vorhanden ist
• Fehler: Löst eine Exception aus, wenn die Tabelle existiert (Standard)

Aufnahme von Streamingdaten mit Structured Streaming

Mit strukturiertem Streaming können Sie Daten kontinuierlich verarbeiten, sobald sie eintreffen. Statt spark.read verwenden Sie spark.readStream, um einen Streaming-DataFrame zu erstellen, der neue Daten inkrementell verarbeitet.

Im folgenden Beispiel werden Streamingdaten aus Kafka gelesen:

df_stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "broker:9092")
    .option("subscribe", "topic-name")
    .option("startingOffsets", "latest")
    .load())

Die startingOffsets Option bestimmt, wo beim Starten einer neuen Abfrage mit dem Lesen begonnen werden soll. Verwenden Sie latest nur für neue Daten oder earliest, um vom Anfang des Themas zu lesen. Sobald Sie eine Prüfpunkt-Position im Schreibstrom konfiguriert haben, werden nachfolgende Läufe automatisch ab dem zuletzt verarbeiteten Offset fortgesetzt, wobei startingOffsets ignoriert wird.

Verarbeiten dateibasierter Datenströme

Sie können auch aus Dateiquellen streamen, wenn neue Dateien in einem Verzeichnis ankommen:

df_file_stream = (spark.readStream
    .format("csv")
    .option("header", "true")
    .schema(defined_schema)
    .load("/Volumes/catalog/schema/volume/incoming/"))

Für Streamingdateiquellen sollten Sie das Schema explizit definieren, anstatt inferSchema zu verwenden. Die Schema-Ableitung wird für Streamingquellen nicht unterstützt, da das gesamte Dataset gelesen werden muss.

Streamingdaten in Tabellen schreiben

Verwenden Sie writeStream, um Streamingdaten in Unity-Katalogtabellen zu persistieren.

(df_stream
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/checkpoints/my-stream")
    .toTable("catalog.schema.streaming_table"))

Der Ausgabemodus bestimmt, welche Datensätze geschrieben werden:

• append: Nur neue Zeilen werden geschrieben (Standard für die meisten Vorgänge)
• update: Geänderte Zeilen werden geschrieben
• vollständig: Alle Zeilen werden neu geschrieben (verwendet bei Aggregationen)

Der Checkpoint-Speicherort speichert Fortschrittsinformationen, sodass der Datenstrom an der Stelle fortgesetzt werden kann, an der er unterbrochen wurde.

Verarbeiten halbstrukturierter Daten

Beim Arbeiten mit halbstrukturierten Daten wie JSON bietet der VARIANT-Datentyp Flexibilität beim Speichern und Abfragen geschachtelter Strukturen:

from pyspark.sql.functions import parse_json, variant_get, col

df = spark.read.text("/path/to/data.json")
df_variant = df.select(parse_json(col("value")).alias("data"))

Anschließend können Sie bestimmte Felder mithilfe der variant_get Funktions- oder SQL-Syntax abfragen:

df_variant.select(variant_get(col("data"), "$.customer.name", "string")).display()

Dieser Ansatz funktioniert gut, wenn das JSON-Schema zwischen Datensätzen variiert oder sich im Laufe der Zeit weiterentwickelt.