Private Link Unterstützung für den Spark-Connector für Microsoft Fabric Data Warehouse

Der Spark-Connector für Fabric Data Warehouse unterstützt Lese- und Schreibvorgänge in Umgebungen, in denen Private Link auf Mandanten- oder Arbeitsbereichsebene aktiviert ist. Schreibvorgänge verwenden JDBC-Batch-Insert-Strategien, die Daten direkt über die JDBC-Verbindung ohne externes Staging senden.

Voraussetzungen

Verwenden Sie diese Importanweisungen am Anfang Ihres Notizbuchs oder bevor Sie mit der Verwendung des Connectors beginnen:

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

JDBC-Batch-Insert-Strategien

Der Connector bietet die Option writeStrategy mit zwei JDBC-basierten Strategien:

Strategie Description Doppelte Sicherheit
MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT Multithread-Schreibvorgänge mit Stagingtabellen Vollständig
NO_DUPLICATES_BATCH_INSERT Single-Thread-Schreibvorgänge mit Staging-Tabellen Vollständig

Beide Strategien verwenden einen Zwei-Phasen-Commit-Ansatz. Daten werden zuerst in temporäre Stagingtabellen geschrieben und dann atomisch in die Zieltabelle zusammengeführt. Dieser Ansatz garantiert, dass keine Duplikate entstehen, selbst wenn Spark fehlgeschlagene Tasks wiederholt.

Automatische Strategieauswahl

In Umgebungen mit aktiviertem Private Link wählt der Connector automatisch MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT aus. Es ist keine Konfiguration erforderlich.

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

Explizite Strategieauswahl

Sie können die Strategie manuell überschreiben, indem Sie die writeStrategy Option verwenden.

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

So funktioniert es

Phase 1: Schreiben in Staging-Tabellen

Jede Spark-Partition schreibt ihre Zeilen in dedizierte temporäre Stagingtabellen mithilfe von Mehrzeilenanweisungen INSERT INTO … VALUES . Stagingtabellen werden vor jedem Schreibvorgang geleert, sodass die Daten, falls Spark einen Task wiederholt, sauber neu eingefügt werden, ohne Duplikate zu erzeugen.

Die Multithread-Strategie parallelisiert diesen Schritt, indem jede Partition mehrere Writer-Threads verwendet, die jeweils in eine eigene Stagingtabelle geschrieben werden.

Phase 2: Atomzusammenführung

Nachdem alle Partitionen abgeschlossen sind, führt der Treiber jede Stagingtabelle in einer einzigen atomalen Transaktion in die Zieltabelle zusammen. Dieser Ansatz garantiert, dass alle Daten entweder in der Zieltabelle landen oder nicht. Alle Stagingtabellen werden automatisch bereinigt.

Auswählen einer Strategie

Verwenden Sie die folgenden Anleitungen, um eine Strategie auszuwählen:

  • Beginnen Sie mit MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT. Es ist die Standardeinstellung in privaten Linkumgebungen und bietet den besten Durchsatz.
  • Wechseln Sie zu NO_DUPLICATES_BATCH_INSERT, wenn OutOfMemory-Fehler im Executor auftreten. Diese Strategie streamt Zeilen, ohne ganze Partitionen zwischenzupuffern, und tauscht dabei Durchsatz gegen eine geringere Speichernutzung ein.

Beide Strategien bieten identische Null-Duplikatgarantien.

Verbessern von Writer-Threads für schnellere Schreibvorgänge

Sie können den Durchsatz erhöhen, indem Sie Writer-Threads pro Spark-Aufgabe erhöhen.

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

In der folgenden Tabelle werden die Konfigurationsdetails für Writer-Threads beschrieben:

Setting Einzelheiten
Standardanzahl von Threads pro Aufgabe 3
Wirkung weiterer Threads Mehr gleichzeitige JDBC-Verbindungen zum Data Warehouse
Überlegung Höhere Speicherauslastung des Ausführungsmoduls; zu viele Threads verursachen möglicherweise OutOfMemoryErrors

Code-Beispiele

Daten mit automatischer Erkennung schreiben

Im folgenden Beispiel wird ein Spark DataFrame in eine Warehouse-Tabelle geschrieben. Der Konnektor wählt automatisch die entsprechende Schreibstrategie aus.

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

Anfügen von Daten an eine vorhandene Tabelle

Im folgenden Beispiel werden neue Zeilen an eine vorhandene Lagertabelle angefügt:

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

Daten in großem Maßstab schreiben

Das folgende Beispiel zeigt eine umfangreiche Schreiboperation mit Neupartitionierung:

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

Überlegungen

Derzeit der Konnektor mit JDBC-Batch-Einfügestrategien:

  • Unterstützt alle DataFrame-Speichermodi (ErrorIfExists, Ignore, Overwrite, Append).
  • Ist mit Spark-Wiederholungen sicher. Stagingtabellen werden beim Wiederholen abgeschnitten.
  • Unterstützt keine Aktualisierungs-, Lösch- oder Upsert-Vorgänge.

Verwandte Inhalte

  • Zuletzt aktualisiert um