Connettore Spark per Microsoft Fabric Data Warehouse

Il connettore Spark per Fabric Data Warehouse consente agli sviluppatori Spark e ai data scientist di accedere e lavorare con i dati di un warehouse e l'endpoint di analisi SQL di una lakehouse. Il connettore offre le funzionalità seguenti:

  • È possibile usare i dati di un warehouse o un endpoint di Analisi SQL nella stessa area di lavoro o in più aree di lavoro.
  • L'endpoint di analisi SQL di una casa sul lago viene scoperto automaticamente in base al contesto dello spazio di lavoro.
  • Il connettore ha un'API Spark semplificata, riassume la complessità sottostante e funziona con una sola riga di codice.
  • Mentre accedi a una tabella o a una vista, il connettore conferma i modelli di sicurezza definiti a livello del motore SQL. Questi modelli includono la protezione a livello di oggetto (OLS), la protezione a livello di riga (RLS) e la protezione a livello di colonna (CLS).
  • Il connettore è preinstallato all'interno del runtime di Fabric, eliminando la necessità di un'installazione separata.

Autenticazione

L'autenticazione di Microsoft Entra ID è un approccio di autenticazione integrato. Gli utenti accedono allo spazio di lavoro Fabric e le loro credenziali vengono automaticamente trasmesse al motore SQL per l'autenticazione e l'autorizzazione. Le credenziali vengono mappate automaticamente e non è necessario che l'utente fornisca opzioni di configurazione specifiche.

Nota

Il connettore Spark per Fabric Data Warehouse supporta solo l'autenticazione utente interattiva di Microsoft Entra. L'autenticazione del service principal non è supportata.

Autorizzazioni

Per connettersi al motore SQL, gli utenti devono disporre almeno dell'autorizzazione di lettura (simile all'autorizzazione CONNECT in SQL Server) per il warehouse o per l'endpoint di Analisi SQL (livello di elemento). Gli utenti devono disporre anche di autorizzazioni granulari a livello di oggetto per leggere i dati da tabelle o viste specifiche. Per saperne di più, consulta Security in Fabric Data Warehouse.

Modelli ed esempi di codice

Usare una firma del metodo

Il comando seguente mostra la firma del synapsesql metodo per la richiesta di lettura. L'argomento tableName in tre parti è necessario per accedere a tabelle o viste da un warehouse e dall'endpoint di Analisi SQL di un lakehouse. Aggiorna l'argomento con i nomi seguenti, in base allo scenario:

  • Parte 1: Nome del magazzino o del lakehouse.
  • Parte 2: nome dello schema.
  • Parte 3: Nome della tabella o della vista.
synapsesql(tableName:String="<Part 1.Part 2.Part 3>") => org.apache.spark.sql.DataFrame

Oltre a leggere direttamente da una tabella o una vista, questo connettore consente anche di specificare una query personalizzata o pass-through che viene passata al motore SQL e il risultato viene restituito a Spark.

spark.read.option(Constants.DatabaseName, "<warehouse/lakeshouse name>").synapsesql("<T-SQL Query>") => org.apache.spark.sql.DataFrame

Anche se questo connettore individua automaticamente l'endpoint per il warehouse/lakehouse specificato, è possibile anche specificarlo in modo esplicito.

//For warehouse
spark.conf.set("spark.datawarehouse.<warehouse name>.sqlendpoint", "<sql endpoint,port>")
//For lakehouse
spark.conf.set("spark.lakehouse.<lakeshouse name>.sqlendpoint", "<sql endpoint,port>")
//Read from table
spark.read.synapsesql("<warehouse/lakeshouse name>.<schema name>.<table or view name>")

Leggere i dati all'interno della stessa area di lavoro

Importante

Eseguire queste istruzioni di importazione all'inizio del notebook o prima di iniziare a usare il connettore:

import com.microsoft.spark.fabric
from com.microsoft.spark.fabric.Constants import Constants

Il codice seguente è un esempio per leggere i dati da una tabella o una vista in un DataFrame Spark:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>")

Il codice seguente è un esempio per leggere i dati da una tabella o una vista in un DataFrame Spark con un numero limite di righe pari a 10:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").limit(10)

Il codice seguente è un esempio per leggere i dati da una tabella o una vista in un DataFrame Spark dopo l'applicazione di un filtro:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").filter("column name == 'value'")

Il codice seguente è un esempio per leggere i dati da una tabella o una vista in un DataFrame Spark solo per le colonne selezionate:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>").select("column A", "Column B")

Leggere i dati tra aree di lavoro

Per accedere e leggere i dati da un magazzino o una lakehouse tra aree di lavoro, è possibile specificare l'ID dell'area di lavoro in cui si trova il magazzino o la lakehouse, e quindi l'ID dell'elemento del magazzino o della lakehouse. La linea seguente fornisce un esempio di lettura di dati da una tabella o una vista all'interno di un DataFrame Spark dal warehouse o lakehouse, utilizzando l'ID dell'area di lavoro e l'ID del lakehouse/warehouse specificati.

# For lakehouse
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").synapsesql("<lakehouse name>.<schema name>.<table or view name>")
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").option(Constants.LakehouseId, "<lakehouse item id>").synapsesql("<lakehouse name>.<schema name>.<table or view name>")

# For warehouse
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").synapsesql("<warehouse name>.<schema name>.<table or view name>")
df = spark.read.option(Constants.WorkspaceId, "<workspace id>").option(Constants.DatawarehouseId, "<warehouse item id>").synapsesql("<warehouse name>.<schema name>.<table or view name>")

Nota

Quando si esegue il notebook, per impostazione predefinita il connettore cerca il magazzino o la lakehouse specificati nell'area di lavoro della lakehouse associata al notebook. Per fare riferimento a un magazzino o una lakehouse da un'altra area di lavoro, specificare l'ID area di lavoro e lakehouse o l'ID articolo del magazzino come indicato in precedenza.

Creare una tabella lakehouse basata sui dati di un warehouse

Queste righe di codice forniscono un esempio per leggere i dati da una tabella o una vista in un dataframe Spark e usarli per creare una tabella lakehouse:

df = spark.read.synapsesql("<warehouse/lakehouse name>.<schema name>.<table or view name>")
df.write.format("delta").saveAsTable("<Lakehouse table name>")

Scrivere dati del dataframe Spark nella tabella del data warehouse

Questo connettore usa un processo di scrittura in due fasi in una tabella fabric DW. Inizialmente, esegue l'inserimento dei dati del dataframe Spark in un archivio intermedio, seguito dal comando COPY INTO per inserire i dati nella tabella DW di Fabric. Questo approccio garantisce la scalabilità con l'aumento del volume di dati.

Modalità supportate di salvataggio del DataFrame

Quando si scrivono dati di origine di un dataframe in una tabella di destinazione nel warehouse, sono supportate le modalità di salvataggio seguenti:

  • ErrorIfExists (modalità di salvataggio predefinita): Se la tabella di destinazione esiste, la scrittura viene interrotta con un'eccezione restituita all'utente chiamante. In caso contrario, viene creata una nuova tabella con i dati.
  • Ignora: se la tabella di destinazione esiste, la scrittura ignora la richiesta di scrittura senza restituire un errore. In caso contrario, viene creata una nuova tabella con i dati.
  • Sovrascrittura: se la tabella di destinazione esiste, i dati esistenti nella destinazione vengono sostituiti con i dati. In caso contrario, viene creata una nuova tabella con i dati.
  • Accodare: Se la tabella di destinazione esiste, i nuovi dati verranno aggiunti. In caso contrario, viene creata una nuova tabella con i dati.

Il codice seguente mostra esempi di scrittura dei dati del dataframe Spark in una tabella di Fabric DW:

df.write.synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>") # this uses default mode - errorifexists

df.write.mode("errorifexists").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("ignore").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("append").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")
df.write.mode("overwrite").synapsesql("<warehouse/lakehouse name>.<schema name>.<table name>")

Nota

Il connettore supporta la scrittura solo in una tabella Fabric DW, perché l'endpoint SQL analytics di un lakehouse è di sola lettura.

Parallelizzazione delle letture per migliorare le prestazioni

Questo connettore supporta letture parallelizzate per migliorare le prestazioni delle query durante il caricamento di tabelle di grandi dimensioni. Analogamente a spark.read.jdbc, è possibile abilitare il parallelismo specificando una colonna di partizione e il relativo intervallo di valori. Spark suddividerà quindi l'operazione di lettura in più partizioni elaborate contemporaneamente.

Il codice seguente illustra come abilitare le letture parallele usando un notebook Spark:

import com.microsoft.spark.fabric.tds.implicits.read.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.tds.implicits.write.FabricSparkTDSImplicits._
import com.microsoft.spark.fabric.Constants
import org.apache.spark.sql.SaveMode 

val df = spark.read.option("partitionColumn", <SomeColumn>)

        .option("lowerBound",  <ColumnValuesLowerLimit>)

        .option("upperBound", <ColumnValuesUpperLimit>)

        .option("numPartitions", <NumberOfPartitionsDesired>).synapsesql(<Table>)

Risoluzione dei problemi

Alla conclusione, l'estratto della risposta di lettura appare nell'output della cella. Un errore nella cella corrente annulla anche le esecuzioni delle celle successive del notebook. Informazioni dettagliate sugli errori sono disponibili nei log dell’applicazione Spark.

Considerazioni sull'uso di questo connettore

Attualmente il connettore:

  • Supporta il recupero o la lettura dei dati dai magazzini Fabric e dagli endpoint di analisi SQL degli elementi lakehouse.
  • Supporta la scrittura dei dati in una tabella warehouse utilizzando diverse modalità di salvataggio: questa funzione è disponibile solo con l'ultima runtime GA, Runtime 2.0.
  • Con collegamento privato abilitata, le operazioni di lettura e scrittura sono supportate sia a livello di tenant che di area di lavoro. Le operazioni di scrittura usano strategie di inserimento in batch JDBC. Per altre informazioni, vedere collegamento privato supporto per il connettore Spark.
  • Fabric DW ora supporta Time Travel, ma questo connettore non funziona per una query con sintassi di spostamento cronologico.
  • Mantiene la firma di utilizzo come quella fornita con Apache Spark per Azure Synapse Analytics per coerenza. Tuttavia, non è retrocompatibile connettersi e usare un pool SQL dedicato in Azure Synapse Analytics.
  • I nomi delle colonne con caratteri speciali sono gestiti aggiungendo un carattere di escape prima dell'invio della query, basato sul nome della tabella/vista a 3 parti. In una lettura personalizzata o basata su query pass-through, gli utenti devono eseguire l'escape dei nomi di colonna contenenti caratteri speciali.

Contenuto correlato

  • Ultimo aggiornamento: