Łącznik Spark dla baz danych SQL

Konektor Spark dla baz danych SQL to wydajna biblioteka, która pozwala czytać i zapisywać do SQL Server, baz danych Azure SQL oraz baz SQL w Fabric. Łącznik oferuje następujące możliwości:

  • Używaj Sparka do wykonywania dużych operacji zapisu i odczytu na Azure SQL Database, Azure SQL Managed Instance, SQL Server na Azure VM oraz bazach SQL w Fabric.
  • W przypadku korzystania z tabeli lub widoku łącznik obsługuje modele zabezpieczeń ustawione 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, więc nie musisz go instalować osobno.

Authentication

Uwierzytelnianie Microsoft Entra jest zintegrowane z Fabric.

  • Po zalogowaniu się do obszaru roboczego Fabric poświadczenia są automatycznie przekazywane do aparatu SQL dla celów uwierzytelniania i autoryzacji.
  • Wymaga włączenia i skonfigurowania Microsoft Entra ID w aparacie bazy danych SQL.
  • Jeśli skonfigurowana jest usługa Microsoft Entra ID, nie jest wymagana dodatkowa konfiguracja w kodzie platformy Spark. Poświadczenia są automatycznie mapowane.

Możesz również użyć metody uwierzytelniania SQL (określając nazwę użytkownika i hasło SQL) lub jednostkę usługi (podając token dostępu platformy Azure na potrzeby uwierzytelniania opartego na aplikacji).

Permissions

Aby korzystać z łącznika Spark, tożsamość — niezależnie od tego, czy jest to użytkownik, czy aplikacja — musi mieć niezbędne uprawnienia bazy danych dla docelowego aparatu SQL. Te uprawnienia są wymagane do odczytu lub zapisu w tabelach i widokach.

W przypadku usług Azure SQL Database, Azure SQL Managed Instance i SQL Server na maszynie wirtualnej platformy Azure:

  • Tożsamość uruchamiająca operację zazwyczaj wymaga db_datawriter oraz db_datareader uprawnień, a opcjonalnie db_owner w celu pełnej kontroli.

Dla bazy danych SQL w Fabric:

  • Tożsamość zwykle wymaga uprawnień db_datawriter i db_datareader, a opcjonalnie db_owner.
  • Ta tożsamość musi również mieć co najmniej uprawnienia do odczytu bazy danych SQL w usłudze Fabric na poziomie elementu.

Uwaga / Notatka

Jeśli używasz jednostki usługi, może ona działać jako aplikacja (bez kontekstu użytkownika) lub jako użytkownik, jeśli personifikacja użytkownika jest włączona. Principal usługi musi mieć wymagane uprawnienia bazy danych dla operacji, które chcesz wykonać.

Przykłady użycia i kodu

W tej sekcji przedstawiono przykłady kodu, aby pokazać, jak efektywnie używać łącznika Spark dla baz danych SQL. Te przykłady obejmują różne scenariusze, w tym odczytywanie i zapisywanie w tabelach SQL oraz konfigurowanie opcji łącznika.

Uwaga / Notatka

Przed zapisem zbiorczym wszystkie przychodzące dane Spark muszą pasować do docelowych typów danych SQL. Gdy nadpisujesz lub tworzysz tabelę, łącznik mapuje wartości TimestampType i TimestampNTZType w Spark na wartości SQL datetime2 zamiast datetime. Typy znaczników czasowych Spark obsługują do sześciu cyfr z dokładnością ułamkowej sekundy, natomiast SQL datetime obsługuje trzy, co może powodować niedopasowanie.

Obsługiwane opcje

Minimalna wymagana opcja to url jako "jdbc:sqlserver://<server>:<port>;database=<database>;" lub ustawienie spark.mssql.connector.default.url.

  • Po podaniu elementu url :

    • Zawsze używaj url jako pierwszej preferencji.
    • Jeśli spark.mssql.connector.default.url nie zostanie ustawiony, łącznik ustawi go i będzie go używać w przyszłości.

  • url Jeśli element nie jest dostarczony:

    • Jeśli spark.mssql.connector.default.url jest ustawiony, łącznik używa wartości z konfiguracji Spark.
    • Jeśli spark.mssql.connector.default.url nie jest ustawiona, zostanie zgłoszony błąd, ponieważ wymagane szczegóły nie są dostępne.

Ten łącznik obsługuje opcje zdefiniowane tutaj: Opcje JDBC źródła danych SQL

Łącznik obsługuje również następujące opcje:

Option Wartość domyślna Description
reliabilityLevel NAJLEPSZY WYSIŁEK Steruje niezawodnością operacji wstawiania. Możliwe wartości: BEST_EFFORT (wartość domyślna, najszybsza, może spowodować zduplikowanie wierszy w przypadku ponownego uruchomienia funkcji wykonawczej) NO_DUPLICATES (wolniejsze, gwarantuje, że żadne zduplikowane wiersze nie zostaną wstawione nawet w przypadku ponownego uruchomienia funkcji wykonawczej). Wybierz zależnie od tolerancji dla duplikatów i potrzeb dotyczących wydajności.
isolationLevel "READ_COMMITTED" Ustawia poziom izolacji transakcji dla operacji SQL. Możliwe wartości: READ_COMMITTED (wartość domyślna uniemożliwia odczytywanie niezatwierdzonych danych), READ_UNCOMMITTED, REPEATABLE_READ, SNAPSHOT, . SERIALIZABLE Wyższe poziomy izolacji mogą zmniejszyć współbieżność, ale poprawić spójność danych.
tableLock „false” Określa, czy podczas operacji wstawiania jest używana wskazówka blokady na poziomie tabeli programu SQL Server TABLOCK. Możliwe wartości: true (włącza funkcję TABLOCK, która może poprawić wydajność zapisu zbiorczego), false (wartość domyślna nie używa funkcji TABLOCK). Ustawienie na true może zwiększyć przepustowość dużych wstawień, ale może zmniejszyć równoległość dla innych operacji na tabeli.
schemaCheckEnabled „prawda” Określa, czy jest wymuszana ścisła weryfikacja schematu między platformą Spark DataFrame a tabelą SQL. Możliwe wartości: true (wartość domyślna, wymusza ścisłe dopasowywanie schematów), false (umożliwia większą elastyczność i może pominąć niektóre kontrole schematu). Ustawienie false może pomóc w rozbieżnościach schematów, ale może skutkować nieoczekiwanymi wynikami, jeśli struktury znacznie się różnią.

Inne opcje interfejsu API zbiorczejDataFrame można ustawić jako opcje dla elementu i są przekazywane do API kopiowania zbiorczego podczas zapisu.

Przykład zapisu i odczytu

Poniższy kod wykorzystuje automatyczne uwierzytelnianie Microsoft Entra ID do demonstracji tych operacji:

  • Zapisz ramkę danych: df.write.option("...", "...").mssql("<schema>.<table>").
  • Przeczytaj tabelę: spark.read.option("...", "...").mssql("<schema>.<table>").
  • Uruchom niestandardowe zapytanie: spark.read.option("...", "...").option("query", "<your-custom-query>").mssql().

Wskazówka

Dane są tworzone na bieżąco do celów demonstracyjnych. W scenariuszu produkcyjnym zazwyczaj odczytujesz dane z istniejącego źródła lub tworzysz bardziej złożony DataFrame element.

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)
df.write.mode("overwrite").option("url", url).mssql("dbo.publicExample")
spark.read.option("url", url).mssql("dbo.publicExample").show()
spark.read.option("url", url).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

url = "jdbc:sqlserver://<server>:<port>;database=<database2>;" # different database
df.write.mode("overwrite").option("url", url).mssql("dbo.tableInDatabase2") # default url is updated
spark.read.mssql("dbo.tableInDatabase2").show() # no url option specified and will use database2

Możesz również wybrać kolumny, zastosować filtry i użyć innych opcji podczas odczytywania danych z aparatu bazy danych SQL.

Przykłady uwierzytelniania

W poniższych przykładach pokazano, jak używać metod uwierzytelniania innych niż Microsoft Entra ID, takich jak jednostka usługi (token dostępu) i uwierzytelnianie SQL.

Uwaga / Notatka

Jak wspomniano wcześniej, uwierzytelnianie za pomocą identyfikatora Entra firmy Microsoft jest obsługiwane automatycznie podczas logowania się do obszaru roboczego Fabric, dlatego musisz używać tych metod tylko wtedy, gdy jest to wymagane w danym scenariuszu.

import com.microsoft.sqlserver.jdbc.spark
url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
row_data = [("Alice", 1),("Bob", 2),("Charlie", 3)]
column_header = ["Name", "Age"]
df = spark.createDataFrame(row_data, column_header)

from azure.identity import ClientSecretCredential
credential = ClientSecretCredential(tenant_id="", client_id="", client_secret="") # service principal app
scope = "https://database.windows.net/.default"
token = credential.get_token(scope).token

df.write.mode("overwrite").option("url", url).option("accesstoken", token).mssql("dbo.publicExample")
spark.read.option("accesstoken", token).mssql("dbo.publicExample").show()
spark.read.option("accesstoken", token).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

Obsługiwane tryby zapisywania ramki danych

Podczas zapisywania danych z platformy Spark do baz danych SQL można wybrać spośród kilku trybów zapisywania. Tryby zapisywania określają sposób zapisywania danych, gdy tabela docelowa już istnieje i może mieć wpływ na schemat, dane i indeksowanie. Zrozumienie tych trybów pomaga uniknąć nieoczekiwanej utraty lub zmian danych.

Ten łącznik obsługuje opcje zdefiniowane tutaj: Funkcje zapisywania platformy Spark

  • ErrorIfExists (domyślny tryb zapisywania): jeśli tabela docelowa istnieje, zapis jest przerwany i zwracany jest wyjątek. W przeciwnym razie zostanie utworzona nowa tabela z danymi.

  • Ignoruj: jeśli tabela docelowa istnieje, zapis ignoruje żądanie i nie zwraca błędu. W przeciwnym razie zostanie utworzona nowa tabela z danymi.

  • Zastąp: jeśli tabela docelowa istnieje, zostanie ona usunięta, utworzona ponownie, a następnie dołączone zostaną nowe dane.

    Uwaga / Notatka

    Gdy używasz overwrite, tracisz oryginalny schemat tabeli (szczególnie typowe dane wyłącznie dla MSSQL) oraz indeksy tabel. Schemat jest zastępowany schematem wywnioskowanym z Twojego Spark DataFrame. Aby uniknąć utraty schematu i indeksów, dodaj .option("truncate", true).

  • Dołącz: jeśli tabela docelowa istnieje, do niej są dołączane nowe dane. W przeciwnym razie zostanie utworzona nowa tabela z danymi.

Troubleshoot

Po zakończeniu procesu dane wyjściowe operacji odczytu platformy Spark są wyświetlane w obszarze danych wyjściowych komórki. Błędy z com.microsoft.sqlserver.jdbc.SQLServerException pochodzą bezpośrednio z programu SQL Server. Szczegółowe informacje o błędach można znaleźć w dziennikach aplikacji platformy Spark.

Zapisy masowe wymagają napływających danych, aby pasowały do docelowych typów danych SQL. Jeśli dane się nie zgadzają, możesz otrzymać ten błąd:

Caused by: com.microsoft.sqlserver.jdbc.SQLServerException: The service has encountered an error processing your request. Please try again. Error code 4815.

Na przykład ten błąd może wystąpić, gdy docelowa tabela SQL używa datetime, ale przychodząca wartość typu TimestampType w Spark, taka jak 2025-01-01 10:30:00.123456, ma większą precyzję, niż obsługuje datetime.

Aby rozwiązać błąd, zastosuj jedno z następujących sposobów:

  • Konwertuj, przycinaj lub przekształcaj dane wejściowe, aby pasowały do docelowego typu danych SQL. Na przykład skróć wartość do trzech cyfr części ułamkowej sekundy: 2025-01-01 10:30:00.123000.
  • Pozwól łącznikowi odtworzyć schemat tabeli, ustawiając .option("truncate", false). Konektor mapuje typ znacznika czasu Spark na SQL datetime2.

Treści powiązane

  • Last updated on