Obsługa Private Link dla łącznika Spark dla Microsoft Fabric Data Warehouse

Łącznik Spark dla Fabric Data Warehouse obsługuje operacje odczytu i zapisu w środowiskach, w których Private Link jest włączona na poziomie dzierżawy lub obszaru roboczego. Operacje zapisu wykorzystują strategie wsadowego wstawiania JDBC, które przesyłają dane bezpośrednio przez połączenie JDBC bez korzystania z zewnętrznego obszaru przejściowego.

Prerequisites

Użyj tych instrukcji importu na początku notesu lub przed rozpoczęciem korzystania z łącznika:

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

Strategie wstawiania zapisu wsadowego JDBC

Łącznik udostępnia opcję writeStrategy z dwiema strategiami opartymi na JDBC:

Strategy Opis Zduplikowane bezpieczeństwo
MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT Zapisy wielowątkowe z tabelami pośrednimi Pełny
NO_DUPLICATES_BATCH_INSERT Jednowątkowe zapisywanie z użyciem tabel etapowych Pełny

Obie strategie stosują protokół zatwierdzania dwufazowego. Dane są najpierw zapisywane w tymczasowych tabelach pośrednich, a następnie scalane atomowo z tabelą docelową. Takie podejście gwarantuje brak duplikatów nawet wtedy, gdy Spark ponawia nieudane zadania.

Wybór strategii automatycznej

W środowiskach z obsługą linków prywatnych łącznik automatycznie wybiera pozycję MULTI_THREAD_NO_DUPLICATES_BATCH_INSERT. Nie jest wymagana żadna konfiguracja.

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

Wybór strategii jawnej

Strategię można ręcznie zastąpić za pomocą opcji 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>")

Jak to działa

Faza 1. Zapisywanie w tabelach przejściowych

Każda partycja Spark zapisuje swoje wiersze do dedykowanych tymczasowych tabel pośrednich przy użyciu instrukcji INSERT INTO … VALUES obejmujących wiele wierszy. Tabele przejściowe są opróżniane przed każdym zapisem, więc jeśli Spark ponawia wykonanie zadania, dane zostają ponownie wstawione bez duplikatów.

Strategia wielowątkowa pozwala zrównoleglić ten etap, ponieważ każda partycja używa wielu wątków zapisujących, z których każdy zapisuje dane do własnej tabeli przejściowej.

Faza 2: Scalanie atomowe

Po zakończeniu przetwarzania wszystkich partycji sterownik scala każdą tabelę przejściową z obiektem docelowym w ramach jednej transakcji atomowej. Takie podejście gwarantuje, że albo wszystkie dane trafiają do tabeli docelowej, albo żadne. Wszystkie tabele przejściowe są czyszczone automatycznie.

Wybieranie strategii

Skorzystaj z poniższych wskazówek, aby wybrać strategię:

  • MULTI_THREAD_NO_DUPLICATES_BATCH_INSERTZacznij od . Jest to wartość domyślna w środowiskach łącza prywatnego i zapewnia najlepszą przepływność.
  • Przełącz na NO_DUPLICATES_BATCH_INSERT, jeśli wystąpią błędy OutOfMemoryError w egzekutorze. Ta strategia przesyła wiersze strumieniowo bez buforowania całych partycji, kosztem przepustowości, aby zmniejszyć zużycie pamięci.

Obie strategie zapewniają identyczne gwarancje zerowe duplikatów.

Zwiększ liczbę wątków zapisu w celu przyspieszenia zapisu

Przepływność można zwiększyć, zwiększając liczbę wątków zapisywania na zadanie platformy Spark.

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

W poniższej tabeli opisano szczegóły konfiguracji wątków zapisywania:

Setting Szczegóły
Domyślne wątki na zadanie 3
Efekt większej liczby wątków Więcej współbieżnych połączeń JDBC z magazynem
Rozważenie Wyższe użycie pamięci wykonawczej; zbyt wiele wątków może powodować błędy OutOfMemoryErrors

Przykłady kodu

Zapisywanie danych za pomocą automatycznego wykrywania

Poniższy przykład zapisuje ramkę danych platformy Spark w tabeli magazynu. Łącznik automatycznie wybiera odpowiednią strategię zapisu.

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

Dołączanie danych do istniejącej tabeli

Poniższy przykład dołącza nowe wiersze do istniejącej tabeli magazynu:

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

Zapisywanie danych na dużą skalę

W poniższym przykładzie przedstawiono operację zapisu na dużą skalę z ponownym partycjonowaniem:

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

Zagadnienia do rozważenia

Obecnie konektor obsługuje strategie wsadowego wstawiania JDBC:

  • Obsługuje wszystkie tryby zapisu DataFrame (ErrorIfExists, Ignore, Overwrite, Append).
  • Jest bezpieczny dzięki ponownym próbom platformy Spark. Tabele przejściowe są opróżniane przy ponowieniu próby.
  • Nie obsługuje operacji aktualizacji, usuwania ani operacji upsert.

Treści powiązane

  • Ostatnia aktualizacja o