Supporto di collegamento privato per il connettore Spark per Microsoft Fabric Data Warehouse

Il connettore Spark per Fabric Data Warehouse supporta operazioni di lettura e scrittura in ambienti in cui collegamento privato è abilitato a livello di tenant o area di lavoro. Le operazioni di scrittura utilizzano strategie di inserimento batch tramite JDBC che inviano i dati direttamente attraverso la connessione JDBC senza staging esterno.

Prerequisiti

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

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

Strategie di scrittura per inserimenti batch JDBC

Il connettore espone un'opzione writeStrategy con due strategie basate su JDBC:

Strategy Description Sicurezza duplicata
MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT Scritture multithread con tabelle di staging Completo
NO_DUPLICATES_BATCH_INSERT Operazioni di scrittura a thread singolo con tabelle di staging Completo

Entrambe le strategie utilizzano un approccio di commit a due fasi. I dati vengono prima scritti nelle tabelle di staging temporanee, quindi uniti in modo atomico nella tabella di destinazione. Questo approccio garantisce zero duplicati anche quando Spark ritenta le attività non riuscite.

Selezione automatica della strategia

Negli ambienti abilitati per i collegamenti privati, il connettore seleziona MULTI_THREAD_NO_DUPLICATES_BATCH_INSERTautomaticamente . Non è necessaria alcuna configurazione.

df.write \
  .mode("overwrite") \
  .synapsesql("<warehouse name>.<schema name>.<table name>")

Selezione esplicita della strategia

È possibile eseguire manualmente l'override della strategia usando l'opzione writeStrategy .

# Multi-threaded (recommended for faster throughput)
df.write \
  .mode("overwrite") \
  .option("writeStrategy", "MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT") \
  .synapsesql("<warehouse name>.<schema name>.<table name>")

# Single-threaded (use if you encounter OutOfMemoryErrors)
df.write \
  .mode("overwrite") \
  .option("writeStrategy", "NO_DUPLICATES_BATCH_INSERT") \
  .synapsesql("<warehouse name>.<schema name>.<table name>")

Come funziona

Fase 1: Scrivere nelle tabelle di staging

Ogni partizione Spark scrive le righe in tabelle di staging temporanee dedicate usando istruzioni su più righe INSERT INTO … VALUES . Le tabelle di staging vengono troncate prima di ogni scrittura, quindi, se Spark riesegue un task, i dati vengono reinseriti senza duplicati.

La strategia multi-thread parallelizza questo passaggio facendo sì che ogni partizione utilizzi più thread di scrittura, ciascuno dei quali scrive nella relativa tabella di staging.

Fase 2: Unione atomica

Dopo il completamento di tutte le partizioni, il driver unisce tutte le tabelle di staging nella tabella di destinazione in un'unica transazione atomica. Questo approccio garantisce che tutti i dati vengano inseriti nella tabella di destinazione oppure che non ne venga inserito alcuno. Tutte le tabelle di staging vengono ripulite automaticamente.

Scegliere una strategia

Usare le indicazioni seguenti per scegliere una strategia:

  • Iniziare con MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT. È l'impostazione predefinita negli ambienti di collegamento privato e offre la velocità effettiva migliore.
  • Passare a NO_DUPLICATES_BATCH_INSERT se si verificano errori OutOfMemory dell'executor. Questa strategia elabora le righe in streaming senza memorizzare intere partizioni nel buffer, sacrificando il throughput per ridurre l'utilizzo della memoria.

Entrambe le strategie forniscono garanzie identiche senza duplicati.

Aumenta i thread di scrittura per velocizzare le scritture

È possibile aumentare il throughput aumentando i thread di scrittura per ciascuna attività Spark.

df.write \
  .mode("overwrite") \
  .option("writeStrategy", "MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT") \
  .option("writerThreadsPerTask", "5") \
  .synapsesql("<warehouse name>.<schema name>.<table name>")

La tabella seguente descrive i dettagli di configurazione per i thread di scrittura:

Impostazione Dettagli
Thread predefiniti per attività 3
Effetto di un maggior numero di thread Più connessioni JDBC simultanee al magazzino
Considerazione Utilizzo elevato della memoria dell'executor; Troppi thread potrebbero causare OutOfMemoryErrors

Esempi di codice

Scrivere dati con rilevamento automatico

Nell'esempio seguente viene scritto un dataframe Spark in una tabella warehouse. Il connettore seleziona automaticamente la strategia di scrittura appropriata.

from pyspark.sql.functions import concat, lit, col, current_timestamp

df = spark.range(100000).toDF("id") \
  .withColumn("name", concat(lit("user_"), col("id"))) \
  .withColumn("score", (col("id") % 100).cast("int"))

df.write \
  .mode("overwrite") \
  .synapsesql("<warehouse name>.<schema name>.<table name>")

Accodare dati a una tabella esistente

Nell'esempio seguente vengono aggiunte nuove righe a una tabella warehouse esistente:

newData = spark.createDataFrame([
  (1001, "Alice", 95),
  (1002, "Bob", 87),
  (1003, "Charlie", 92)
], ["id", "name", "score"])

newData.write \
  .mode("append") \
  .synapsesql("<warehouse name>.<schema name>.<table name>")

Scrittura di dati su larga scala

L'esempio seguente illustra un'operazione di scrittura su larga scala con ripartizionamento:

from pyspark.sql.functions import concat, lit, col, current_timestamp

bigDf = spark.range(10000000).repartition(16).toDF("id") \
  .withColumn("name", concat(lit("user_"), col("id"))) \
  .withColumn("city", lit("Seattle")) \
  .withColumn("age", (col("id") % 100).cast("int")) \
  .withColumn("salary", (col("id") * 1.5).cast("double")) \
  .withColumn("active", col("id") % 2 == 0) \
  .withColumn("created_at", current_timestamp())

bigDf.write \
  .mode("overwrite") \
  .synapsesql("<warehouse name>.<schema name>.<table name>")

Considerazioni

Attualmente, il connettore con strategie di inserimento batch JDBC:

  • Supporta tutte le modalità di salvataggio del DataFrame (ErrorIfExists, Ignore, Overwrite, Append).
  • È sicuro con i tentativi di Spark. Le tabelle di staging vengono troncate in caso di nuovo tentativo.
  • Non supporta operazioni di aggiornamento, eliminazione o inserimento/aggiornamento.

Contenuti correlati

  • Ultimo aggiornamento: