Caricare i dati nelle pipeline

È possibile caricare dati da qualsiasi origine dati supportata da Apache Spark in Azure Databricks usando le pipeline. È possibile definire set di dati (tabelle e viste) in una pipeline su qualsiasi query che restituisca un dataframe Spark, inclusi i dataframe di streaming e Pandas per i dataframe Spark. Per le attività di inserimento dati, Databricks consiglia di usare le tabelle di streaming per la maggior parte dei casi d'uso. Le tabelle di streaming sono utili per l'inserimento di dati dall'archiviazione di oggetti cloud tramite il caricatore automatico o da bus di messaggi come Kafka. Per altre informazioni sulle tabelle di streaming, il tipo di set di dati primario per l'inserimento, vedere Tabelle di streaming.

Non tutte le origini dati supportano SQL per l'inserimento. Tuttavia, è possibile combinare origini SQL e Python nella stessa pipeline per usare Python dove necessario. Per informazioni dettagliate sull'uso delle librerie non in pacchetto con le pipeline per impostazione predefinita, vedere Gestire le dipendenze Python per le pipeline. Per informazioni generali sull'inserimento in Azure Databricks, vedere Connettori Standard in Lakeflow Connect.

Gli esempi seguenti illustrano alcuni modelli comuni di caricamento dei dati.

Identifica le tue fonti di dati e il percorso di connessione

Prima di scrivere codice per pipeline, inventa ogni luogo da cui provengono i dati. Per ogni sorgente, annota come espone i suoi dati (file, un database, un sistema software as a service (SaaS), un'API o un stream), quanto spesso cambia e quali credenziali e accesso alla rete necessita. Il metodo di connessione spesso determina se una sorgente è naturalmente batch o streaming, quindi farlo correttamente presto evita rilavori in seguito.

Distribuisci ogni sorgente in uno dei seguenti percorsi di connessione. La tabella seguente elenca il meccanismo preferito per ciascuna sorgente di pipeline:

Per ogni fonte, conferma quanto segue prima di costruire:

• Identità: come funziona la pipeline. Le pipeline possono funzionare come principale di servizio, quindi impostalo prima per evitare di dipendere da un account personale.
• Percorso di rete: Quale connettività serve la sorgente, come una credenziale di storage, una posizione esterna o la connettività gestita da Lakeflow Connect.
• Semantica delle modifiche: In che modo l'origine segnala gli aggiornamenti e le eliminazioni, se lo fa. Questo determina se è necessario usare CDC o se puoi considerare la sorgente come di sola aggiunta.

Scegli un formato file e un livello di archiviazione

Le pipeline prendono la maggior parte di queste decisioni per te. Per impostazione predefinita, ogni tabella di streaming e ogni vista materializzata create da una pipeline vengono archiviate come tabelle Delta, offrendo transazioni ACID, convalida ed evoluzione dello schema, funzionalità di time travel e governance e tracciabilità di Unity Catalog per ogni set di dati. Non scegli il formato per gli output della pipeline. Le tue vere decisioni si trovano ai due margini della pipeline:

• Formato di input grezzo: qualunque cosa produca la sorgente, come CSV, JSON o Parquet. Auto Loader e read_files() supportano direttamente questi. Specifica il formato con cloudFiles.format in Python o l'argomento format => in SQL. Se controlli i dati di origine, preferisci Parquet o Avro, perché includono lo schema e si comprimono meglio, accelerando l'ingestione e l'inferenza dello schema. Le pipeline gestiscono uno qualsiasi di questi formati, quindi non lasciare che il formato limiti la scelta della fonte.
• Percorso di archiviazione raw: Per i file, fai confluire i dati in un volume di Unity Catalog anziché in un percorso di bucket non gestito, così la tracciabilità e il controllo degli accessi si estendono fino alla landing zone. Consulta Che cosa sono i volumi di Unity Catalog?.

Per le tabelle generate da una pipeline, le opzioni rimanenti sono il catalogo di destinazione e lo schema, che definiscono il perimetro di governance e l'individuabilità, nonché l'organizzazione fisica delle tabelle di grandi dimensioni. Usa CLUSTER BY (cluster liquido) per mantenere buone prestazioni delle query man mano che le tabelle crescono senza dover ottimizzare manualmente le partizioni. Vedere Usare clustering liquido per le tabelle.

Caricare da una tabella esistente

Caricare dati da qualsiasi tabella esistente in Azure Databricks. È possibile trasformare i dati usando una query o caricare la tabella per un'ulteriore elaborazione nella pipeline.

Pitone

@dp.table(
  comment="A table summarizing counts of the top baby names for New York for 2021."
)
def top_baby_names_2021():
  return (
    spark.read.table("baby_names_prepared")
      .filter(expr("Year_Of_Birth == 2021"))
      .groupBy("First_Name")
      .agg(sum("Count").alias("Total_Count"))
      .sort(desc("Total_Count"))
  )

SQL

CREATE OR REFRESH MATERIALIZED VIEW top_baby_names_2021
COMMENT "A table summarizing counts of the top baby names for New York for 2021."
AS SELECT
  First_Name,
  SUM(Count) AS Total_Count
FROM baby_names_prepared
WHERE Year_Of_Birth = 2021
GROUP BY First_Name
ORDER BY Total_Count DESC

Caricare file dall'archiviazione di oggetti cloud

Databricks consiglia di usare il caricatore automatico nelle pipeline per la maggior parte delle attività di inserimento dati dall'archiviazione di oggetti cloud o da file in un volume di Unity Catalog. Il caricatore automatico e le pipeline sono progettati per caricare in modo incrementale e idempotente i dati in continua crescita man mano che arrivano nell'archiviazione cloud. Vedere Che cos'è il caricatore automatico? e Caricare i dati dall'archivio oggetti.

L'esempio seguente legge i dati dall'archiviazione cloud usando il caricatore automatico.

Pitone

@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json")
  )

SQL

CREATE OR REFRESH STREAMING TABLE sales
  AS SELECT *
  FROM STREAM read_files(
    'abfss://myContainer@myStorageAccount.dfs.core.windows.net/analysis/*/*/*.json',
    format => "json"
  );

Gli esempi seguenti usano Il caricatore automatico per creare set di dati da file CSV in un volume del catalogo Unity.

Pitone

@dp.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/Volumes/my_catalog/retail_org/customers/")
  )

SQL

CREATE OR REFRESH STREAMING TABLE customers
AS SELECT * FROM STREAM read_files(
  "/Volumes/my_catalog/retail_org/customers/",
  format => "csv"
)

Eseguire l'autenticazione nell'archiviazione cloud

Auto Loader utilizza le posizioni esterne di Unity Catalog per autenticarsi con l'archiviazione cloud. È necessario configurare un percorso esterno per il percorso di archiviazione da cui si vuole leggere e concedere il READ FILES privilegio all'utente in esecuzione.

Per inserire da Azure Data Lake Storage, configurare una posizione esterna supportata da credenziali di archiviazione che fanno riferimento a un contenitore di archiviazione. Per altre informazioni, vedere Connettersi all'archiviazione di oggetti cloud usando il catalogo unity.

Caricare dati da un bus di messaggi

È possibile configurare le pipeline per acquisire i dati dai bus di messaggi. Databricks consiglia di usare tabelle di streaming con esecuzione continua e scalabilità automatica avanzata per fornire l'inserimento più efficiente per il caricamento a bassa latenza dai bus di messaggi. Per altre informazioni, vedere Ottimizzare l'utilizzo del cluster della pipeline Lakeflow con la scalabilità automatica.

Ad esempio, il codice seguente configura una tabella di streaming per inserire dati da Kafka usando la funzione read_kafka .

Pitone

from pyspark import pipelines as dp

@dp.table
def kafka_raw():
  return (
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "kafka_server:9092")
      .option("subscribe", "topic1")
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE kafka_raw AS
  SELECT *
  FROM STREAM read_kafka(
    bootstrapServers => 'kafka_server:9092',
    subscribe => 'topic1'
  );

Inserimento da Google Pub/Sub

Nell'esempio seguente viene creata una tabella di streaming che legge da un argomento Google Pub/Sub usando la funzione read_pubsub .

Pitone

@dp.table
def pubsub_raw():
  auth_options = {
    "clientId": client_id,
    "clientEmail": client_email,
    "privateKey": private_key,
    "privateKeyId": private_key_id
  }
  return (
    spark.readStream
      .format("pubsub")
      .option("subscriptionId", "my-subscription")
      .option("topicId", "my-topic")
      .option("projectId", "my-project")
      .options(auth_options)
      .load()
  )

SQL

CREATE OR REFRESH STREAMING TABLE pubsub_raw
AS SELECT * FROM STREAM read_pubsub(
  subscriptionId => 'my-subscription',
  projectId => 'my-project',
  topicId => 'my-topic',
  clientEmail => secret('pubsub-scope', 'clientEmail'),
  clientId => secret('pubsub-scope', 'clientId'),
  privateKeyId => secret('pubsub-scope', 'privateKeyId'),
  privateKey => secret('pubsub-scope', 'privateKey')
);

Databricks consiglia di usare i segreti quando si forniscono opzioni di autorizzazione. Per tutte le opzioni di autenticazione, vedere Configurare l'accesso a Pub/Sub .

Per acquisire dati da altre sorgenti del bus di messaggi, vedere:

• Leggi Kinesis
• Pulsar: read_pulsar

Caricare dati da Hub eventi di Azure

Hub eventi di Azure è un servizio di streaming di dati che fornisce un'interfaccia compatibile con Apache Kafka. È possibile usare il connettore Structured Streaming Kafka, incluso nel runtime della pipeline, per caricare i messaggi da Hub eventi di Azure. Per altre informazioni sul caricamento e l'elaborazione dei messaggi da Hub eventi di Azure, vedere Usare Hub eventi di Azure come origine dati della pipeline.

Caricare dati da sistemi esterni

Le pipeline supportano il caricamento di dati da qualsiasi origine dati supportata da Azure Databricks. Vedere Connettersi a origini dati e servizi esterni. È possibile anche caricare dati esterni utilizzando Lakehouse Federation per le origini dati supportate . Poiché Lakehouse Federation richiede Databricks Runtime 13.3 LTS o versione successiva, per usare Lakehouse Federation, configurare la pipeline per utilizzare il canale di anteprima.

Alcune origini dati non hanno un supporto SQL equivalente. Se non è possibile usare Lakehouse Federation con una di queste origini dati, è possibile usare Python per inserire dati dall'origine. È possibile aggiungere file di origine Python e SQL alla stessa pipeline. Nell'esempio seguente viene dichiarata una vista materializzata per accedere allo stato corrente dei dati in una tabella PostgreSQL remota.

import dp

@dp.table
def postgres_raw():
  return (
    spark.read
      .format("postgresql")
      .option("dbtable", table_name)
      .option("host", database_host_url)
      .option("port", 5432)
      .option("database", database_name)
      .option("user", username)
      .option("password", password)
      .load()
  )

Caricare set di dati statici o di piccole dimensioni dall'archiviazione di oggetti cloud

È possibile caricare set di dati statici o di piccole dimensioni usando la sintassi di caricamento di Apache Spark. Le pipeline supportano tutti i formati di file supportati da Apache Spark in Azure Databricks. Per un elenco completo, vedere le opzioni di formato dati .

Gli esempi seguenti illustrano il caricamento di JSON per creare una tabella.

Pitone

@dp.table
def clickstream_raw():
  return (spark.read.format("json").load("/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"))

SQL

CREATE OR REFRESH MATERIALIZED VIEW clickstream_raw
AS SELECT * FROM read_files(
  "/databricks-datasets/wikipedia-datasets/data-001/clickstream/raw-uncompressed-json/2015_2_clickstream.json"
)

Caricare dati da un'origine dati personalizzata Python

Le origini dati personalizzate python consentono di caricare i dati in formati personalizzati. È possibile scrivere codice da cui leggere e scrivere in un'origine dati esterna specifica oppure usare il codice Python esistente per leggere i dati dai propri sistemi interni. Per altri dettagli sullo sviluppo di origini dati Python, vedere Origini dati personalizzate PySpark.

L'esempio seguente registra un'origine dati personalizzata con il nome my_custom_datasource del formato e ne legge in modalità batch e streaming.

from pyspark import pipelines as dp

# Assume `my_custom_datasource` is a custom Python custom data
# source that supports both batch and streaming reads, and has
# been registered using `spark.dataSource.register`.

# This creates a materialized view
@dp.table(name = "read_from_batch")
def read_from_batch():
    return spark.read.format("my_custom_datasource").load()

# This creates a streaming table
@dp.table(name = "read_from_streaming")
def read_from_streaming():
    return spark.readStream.format("my_custom_datasource").load()

Configurare una tabella di streaming per ignorare le modifiche in una tabella di streaming di origine

Per impostazione predefinita, le tabelle di streaming richiedono fonti a sola aggiunta. Se la tabella di streaming di origine richiede aggiornamenti o eliminazioni (ad esempio per gestire il «diritto all’oblio» ai sensi del GDPR), utilizzare il flag skipChangeCommits per ignorare tali modifiche. Questo flag funziona solo con la funzione spark.readStream e non può essere usato quando la tabella di streaming di origine è il target di una funzione option(). Per altre informazioni, vedere Gestire le modifiche alle tabelle Delta Lake di origine.

@dp.table
def b():
   return spark.readStream.option("skipChangeCommits", "true").table("A")

accedere in modo sicuro alle credenziali di archiviazione con chiavi segrete in una pipeline

È possibile usare i segreti di Azure Databricks per archiviare credenziali come chiavi di accesso o password. Per configurare il segreto nella pipeline, utilizzare una proprietà Spark nella configurazione del cluster della pipeline. Vedere Configurare il calcolo classico per le pipeline.

Nell'esempio seguente viene usato un segreto per archiviare una chiave di accesso necessaria per leggere i dati di input da un account di archiviazione Azure Data Lake Storage usando il caricatore automatico. È possibile usare questo stesso metodo per configurare qualsiasi segreto richiesto dalla pipeline, ad esempio le chiavi AWS per accedere a S3 o la password a un metastore Apache Hive.

Per ulteriori informazioni sull'uso di Azure Data Lake Storage, vedere Connettersi ad Azure Data Lake Storage e gestione rete virtuale di Azure.

{
  "id": "43246596-a63f-11ec-b909-0242ac120002",
  "storage": "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<path>",
  "clusters": [
    {
      "spark_conf": {
        "spark.hadoop.fs.azure.account.key.<storage-account-name>.dfs.core.windows.net": "{{secrets/<scope-name>/<secret-name>}}"
      },
      "autoscale": {
        "min_workers": 1,
        "max_workers": 5,
        "mode": "ENHANCED"
      }
    }
  ],
  "development": true,
  "continuous": false,
  "libraries": [
    {
      "notebook": {
        "path": "/Users/user@databricks.com/Pipeline Notebooks/pipeline quickstart"
      }
    }
  ],
  "name": "pipeline quickstart using ADLS2"
}

In questo esempio di codice sostituire i valori seguenti.

from pyspark import pipelines as dp

json_path = "abfss://<container-name>@<storage-account-name>.dfs.core.windows.net/<path-to-input-dataset>"
@dp.create_table(
  comment="Data ingested from an ADLS2 storage account."
)
def read_from_ADLS2():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load(json_path)
  )

In questo esempio di codice sostituire i valori seguenti.