Łącznik Spark dla magazynu danych Microsoft Fabric

Łącznik Spark dla usługi Fabric Data Warehouse umożliwia deweloperom platformy Spark i analitykom danych uzyskiwanie dostępu do danych z magazynu i punktu końcowego analizy SQL w usłudze Lakehouse oraz pracę z nimi. Łącznik oferuje następujące możliwości:

• Możesz pracować z danymi z magazynu lub punktu końcowego analizy SQL w tym samym obszarze roboczym lub w wielu obszarach roboczych.
• Endpoint analityki SQL w domku nad jeziorem jest automatycznie wykrywany w zależności od kontekstu przestrzeni roboczej.
• Łącznik ma uproszczony interfejs API platformy Spark, abstrahuje podstawową złożoność i działa przy użyciu tylko jednego wiersza kodu.
• Podczas uzyskiwania dostępu do tabeli lub widoku łącznik podtrzymuje modele zabezpieczeń zdefiniowane na poziomie aparatu SQL. Te modele obejmują zabezpieczenia na poziomie obiektu (OLS), zabezpieczenia na poziomie wiersza (RLS) i zabezpieczenia na poziomie kolumny (CLS).
• Łącznik jest wstępnie zainstalowany w środowisku uruchomieniowym Fabric, co eliminuje potrzebę osobnej instalacji.

Uwierzytelnianie

Uwierzytelnianie Microsoft Entra to zintegrowane podejście do uwierzytelniania. Użytkownicy logują się do przestrzeni roboczej Fabric, a ich dane uwierzytelniające są automatycznie przekazywane do silnika SQL w celu uwierzytelniania i autoryzacji. Poświadczenia są automatycznie mapowane, a użytkownicy nie muszą udostępniać określonych opcji konfiguracji.

Uprawnienia

Aby nawiązać połączenie z silnikiem SQL, użytkownicy muszą mieć co najmniej uprawnienia odczytu (podobne do uprawnień CONNECT w programie SQL Server) w magazynie lub punkcie końcowym analizy SQL na poziomie poszczególnych elementów. Użytkownicy potrzebują również szczegółowych uprawnień na poziomie obiektu, aby odczytywać dane z określonych tabel lub widoków. Aby dowiedzieć się więcej, zobacz Security in Fabric Data Warehouse.

Szablony kodu i przykłady

Użyj sygnatury metody

Następujące polecenie przedstawia sygnaturę synapsesql metody dla żądania odczytu. Trzyczęściowy tableName argument jest wymagany do uzyskiwania dostępu do tabel lub widoków z magazynu danych oraz do punktu końcowego analityki SQL typu lakehouse. Zaktualizuj argument przy użyciu następujących nazw w zależności od scenariusza:

• Część 1. Nazwa magazynu lub jeziora.
• Część 2. Nazwa schematu.
• Część 3. Nazwa tabeli lub widoku.

synapsesql(tableName:String="<Part 1.Part 2.Part 3>") => org.apache.spark.sql.DataFrame

Oprócz bezpośredniego odczytywania z tabeli lub widoku, ten łącznik umożliwia również określenie niestandardowego lub przekazywanego zapytania, które przekazuje zapytanie do silnika SQL, który zwraca wynik do Spark.

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

Chociaż ten łącznik automatycznie odnajduje punkt końcowy dla określonego magazynu/magazynu typu lakehouse, jeśli chcesz go jawnie określić, możesz to zrobić.

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

Odczytywanie danych w tym samym obszarze roboczym

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

%%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 

Poniższy kod to przykład odczytu danych z tabeli lub widoku w ramce danych platformy Spark:

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

Poniższy kod to przykład odczytu danych z tabeli lub widoku w ramce danych platformy Spark z limitem liczby wierszy 10:

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

Poniższy kod to przykład odczytu danych z tabeli lub widoku w ramce danych platformy Spark po zastosowaniu filtru:

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

Poniższy kod to przykład odczytu danych z tabeli lub widoku w ramce danych platformy Spark tylko dla wybranych kolumn:

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

Odczytywanie danych z różnych przestrzeni roboczych

Aby uzyskać dostęp do danych z magazynu lub lakehouse w różnych obszarach roboczych i odczytywać je, możesz określić identyfikator obszaru roboczego, w którym znajduje się magazyn lub lakehouse, a następnie identyfikator elementu lakehouse lub magazynu. Poniższy wiersz zawiera przykład odczytu danych z tabeli lub widoku w Spark DataFrame z warehouse lub lakehouse, z wykorzystaniem określonego identyfikatora obszaru roboczego oraz identyfikatora lakehouse/warehouse.

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

Tworzenie tabeli lakehouse na podstawie danych z magazynu

Te wiersze kodu stanowią przykład odczytu danych z tabeli lub widoku w ramce danych platformy Spark i używają ich do tworzenia tabeli typu lakehouse:

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

Zapisywanie danych ramki danych platformy Spark w tabeli magazynu

Ten konektor wykorzystuje dwufazowy proces zapisu w tabeli Fabric DW. Początkowo dane ramki danych Spark są umieszczane w pamięci pośredniej, a następnie używane jest polecenie COPY INTO do załadowania danych do tabeli Fabric DW. Takie podejście zapewnia skalowalność przy zwiększaniu ilości danych.

Obsługiwane tryby zapisywania ramki danych

Następujące tryby zapisu są obsługiwane podczas zapisywania danych źródłowych ramki danych do tabeli docelowej w hurtowni danych:

• ErrorIfExists (domyślny tryb zapisywania): jeśli tabela docelowa istnieje, zapis zostanie przerwany, a wyjątek zwrócony do wywołującego. W przeciwnym razie zostanie utworzona nowa tabela z danymi.
• Ignoruj: jeśli tabela docelowa istnieje, zapis ignoruje żądanie zapisu bez zwracania błędu. W przeciwnym razie zostanie utworzona nowa tabela z danymi.
• Zastąp: jeśli tabela docelowa istnieje, istniejące dane w miejscu docelowym są zastępowane danymi. W przeciwnym razie zostanie utworzona nowa tabela z danymi.
• Dołącz: Jeśli tabela docelowa istnieje, nowe dane są do niej dołączane. W przeciwnym razie zostanie utworzona nowa tabela z danymi.

Poniższy kod przedstawia przykłady zapisywania danych DataFrame Spark do tabeli 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>")

Równoleglanie odczytów dla poprawy wydajności

Ten łącznik obsługuje równoległe odczyty w celu zwiększenia wydajności zapytań podczas ładowania dużych tabel. Podobnie jak spark.read.jdbc, można włączyć równoległość, określając kolumnę partycji i jej zakres wartości. Platforma Spark podzieli operację odczytu na wiele partycji, które są przetwarzane współbieżnie.

Poniższy kod pokazuje, jak włączyć równoległe odczyty przy użyciu notatnika 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>)

Rozwiązywanie problemów

Po zakończeniu fragment odpowiedzi z odczytu pojawia się w danych wyjściowych komórki. Błąd w bieżącej komórce powoduje również anulowanie kolejnych wykonań komórek w notesie. Szczegółowe informacje o błędzie są dostępne w dziennikach aplikacji platformy Spark.

Zagadnienia dotyczące korzystania z tego łącznika

Obecnie łącznik:

• Obsługuje pobieranie lub odczytywanie danych z magazynów Fabric i punktów końcowych analizy SQL elementów lakehouse.
• Obsługuje zapisywanie danych do tabeli magazynowej poprzez różne tryby zapisu – funkcja ta jest dostępna tylko w najnowszym środowisku uruchomieniowym GA, Runtime 2.0.
• Po włączeniu Private Link operacje odczytu i zapisu są możliwe zarówno na poziomie dzierżawy, jak i obszaru roboczego. Operacje zapisu wykorzystują strategie wsadowego wstawiania JDBC. Aby uzyskać więcej informacji, zobacz obsługę funkcji Private Link dla łącznika Spark.
• Usługa Fabric DW obsługuje Time Travel teraz jednak ten łącznik nie działa w przypadku zapytania ze składnią podróży w czasie.
• Zachowuje sygnaturę użycia, taką jak ta dostarczana z platformą Apache Spark dla usługi Azure Synapse Analytics w celu zapewnienia spójności. Jednak nie jest to kompatybilne wstecz, więc nie można nawiązać połączenia ani pracować z dedykowaną pulą SQL w usłudze Azure Synapse Analytics.
• Nazwy kolumn ze znakami specjalnymi będą obsługiwane przez dodanie znaku ucieczki przed wysłaniem zapytania na podstawie 3 części nazwy tabeli/widoku. W przypadku odczytu niestandardowego lub opartego na zapytaniu przekazywanym użytkownicy muszą unikać nazw kolumn zawierających znaki specjalne.

Powiązana zawartość

• Obsługa łącza prywatnego dla łącznika spark
• Środowiska uruchomieniowe platformy Apache Spark w Fabric
• Omówienie monitorowania platformy Apache Spark
• Bezpieczeństwo w Fabric Data Warehouse
• Uprawnienia szczegółowe SQL w Fabric