Ingestire dati con notebook

I notebook in Azure Databricks offrono un approccio flessibile basato sul codice per l'inserimento di dati da varie origini. Quando gli strumenti grafici non soddisfano i requisiti o quando è necessaria una logica personalizzata per la trasformazione dei dati, i notebook offrono il controllo completo sul processo di inserimento. È possibile usare Python (PySpark), SQL, Scala o R per connettersi alle origini dati, applicare trasformazioni e caricare i dati nelle tabelle del catalogo Unity.

In questa unità si apprenderà come inserire dati usando i notebook per scenari batch e di streaming.

Inserire dati batch con DataFrames

I dataframe sono la base per l'inserimento di dati batch in Databricks. Un dataframe è una raccolta distribuita di dati organizzati in colonne denominate, analogamente a una tabella in un database relazionale. Usare l'API spark.read per caricare dati da varie origini in un dataframe.

L'esempio seguente legge un file CSV in un dataframe:

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

Il format() metodo specifica il tipo di file, mentre option() i metodi configurano la modalità di analisi dei dati. L'opzione header indica a Spark di usare la prima riga come nomi di colonna e inferSchema rileva automaticamente i tipi di dati.

Leggere da formati di file diversi

Azure Databricks supporta più formati di file. È possibile modificare il formato e le opzioni in base all'origine dati:

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

I file Parquet archiviano i dati in un formato a colonne, migliorando le prestazioni delle query per i carichi di lavoro analitici. I file JSON supportano strutture annidate e sono comuni per le risposte api. I file XML richiedono l'opzione rowTag per specificare quale elemento corrisponde a ogni riga.

Connessione a database esterni.

Per i dati archiviati in database relazionali, usare le connessioni JDBC per leggere direttamente dall'origine:

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

Archiviare le credenziali nei segreti di Databricks anziché inserirle nel codice nei notebook. È anche possibile eseguire il push delle query nel database usando una sottoquery nell'opzione dbtable :

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

Scrivere dati nelle tabelle del catalogo Unity

Dopo il caricamento e la trasformazione dei dati, scriverli in una tabella del catalogo Unity. Azure Databricks usa il formato Delta Lake per impostazione predefinita, che fornisce transazioni ACID e controllo delle versioni:

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

L'opzione mode() controlla la modalità di gestione dei dati esistenti:

• sovrascrivi: sostituisce completamente il contenuto della tabella
• append: aggiunge righe alla tabella esistente
• ignore: non esegue alcuna operazione se la tabella esiste già
• error: genera un'eccezione se la tabella esiste (impostazione predefinita)

Inserire dati di streaming con Structured Streaming

Structured Streaming consente di elaborare i dati in modo continuo man mano che arrivano. Anziché spark.read, viene utilizzato spark.readStream per creare un DataFrame di streaming che elabora i nuovi dati in modo incrementale.

L'esempio seguente legge i dati di streaming da Kafka:

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

L'opzione startingOffsets determina dove iniziare la lettura all'avvio di una nuova query. Usare latest solo per i nuovi dati o earliest per leggere dall'inizio dell'argomento. Dopo aver configurato un percorso del checkpoint nel flusso di scrittura, le esecuzioni successive riprendono automaticamente dall'ultimo offset elaborato, ignorando startingOffsets.

Elaborare flussi basati su file

È anche possibile eseguire lo streaming da origini file man mano che arrivano nuovi file in una directory:

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

Per le origini di file di streaming, è necessario definire lo schema in modo esplicito anziché usare inferSchema. L'inferenza dello schema non è supportata per le origini di streaming perché richiede la lettura dell'intero set di dati.

Scrivere dati di streaming in tabelle

Usare writeStream per rendere persistenti i dati di streaming nelle tabelle di Unity Catalog:

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

La modalità di output determina quali record vengono scritti:

• append: vengono scritte solo nuove righe (impostazione predefinita per la maggior parte delle operazioni)
• update: le righe modificate vengono scritte
• complete: tutte le righe vengono riscritte (usate con le aggregazioni)

La posizione del checkpoint archivia le informazioni sullo stato, consentendo al flusso di riprendere da dove è stato interrotto se interrotto.

Gestire dati semistrutturati

Quando si utilizzano dati semistrutturati come JSON, il tipo di dati VARIANT offre flessibilità per l'archiviazione e l'esecuzione di query su strutture annidate:

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

È quindi possibile eseguire query su campi specifici usando la funzione o la variant_get sintassi SQL:

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

Questo approccio funziona correttamente quando lo schema JSON varia tra i record o si evolve nel tempo.