Spark-Connector für SQL-Datenbanken (Vorschau)

Der Spark-Connector für SQL-Datenbanken ist eine Hochleistungsbibliothek, mit der Sie sql Server-, Azure SQL- und Fabric SQL-Datenbanken lesen und schreiben können. Der Connector bietet die folgenden Funktionen:

• Verwenden Sie Spark, um große Schreib- und Lesevorgänge für Azure SQL-Datenbank, azure SQL Managed Instance, SQL Server auf Azure VM und Fabric SQL-Datenbanken auszuführen.
• Wenn Sie eine Tabelle oder Ansicht verwenden, unterstützt der Connector Sicherheitsmodelle, die auf SQL-Modulebene festgelegt sind. Zu diesen Modellen gehören Sicherheit auf Objektebene (Object-Level Security, OLS), Sicherheit auf Zeilenebene (Row-Level Security, RLS) und Sicherheit auf Spaltenebene (Column-Level Security, CLS).

Der Connector ist in der Fabric-Laufzeit vorinstalliert, sodass Sie ihn nicht separat installieren müssen.

Authentifizierung

Die Microsoft Entra-Authentifizierung ist in Microsoft Fabric integriert.

• Wenn Sie sich beim Fabric-Arbeitsbereich anmelden, werden Ihre Anmeldeinformationen automatisch zur Authentifizierung und Autorisierung an das SQL-Modul übergeben.
• Erfordert, dass Microsoft Entra-ID auf Ihrem SQL-Datenbankmodul aktiviert und konfiguriert wird.
• Es ist keine zusätzliche Konfiguration in Ihrem Spark-Code erforderlich, wenn Die Microsoft Entra-ID eingerichtet ist. Die Anmeldeinformationen werden automatisch zugeordnet.

Sie können auch die SQL-Authentifizierungsmethode verwenden (indem Sie einen SQL-Benutzernamen und ein Kennwort angeben) oder einen Dienstprinzipal (indem Sie ein Azure-Zugriffstoken für die appbasierte Authentifizierung bereitstellen).

Erlaubnisse

Um den Spark-Connector zu verwenden, muss Ihre Identität – unabhängig davon, ob es sich um einen Benutzer oder eine App handelt – über die erforderlichen Datenbankberechtigungen für das SQL-Zielmodul verfügen. Diese Berechtigungen sind erforderlich, um aus Tabellen und Ansichten zu lesen oder in Tabellen und Ansichten zu schreiben.

Für Azure SQL-Datenbank, azure SQL Managed Instance und SQL Server auf azure VM:

• Die Identität, die den Vorgang ausführt, benötigt normalerweise db_datawriter- und db_datareader-Berechtigungen und optional db_owner für Vollzugriff.

Für Fabric SQL-Datenbanken:

• Die Identität benötigt in der Regel die Berechtigungen db_datawriter und db_datareader, und optional db_owner.
• Die Identität benötigt außerdem mindestens Leseberechtigungen für die Fabric SQL-Datenbank auf Elementebene.

Verwendungs- und Codebeispiele

In diesem Abschnitt stellen wir Codebeispiele bereit, um zu veranschaulichen, wie der Spark-Connector für SQL-Datenbanken effektiv verwendet wird. In diesen Beispielen werden verschiedene Szenarien behandelt, z. B. das Lesen von und Schreiben in SQL-Tabellen und das Konfigurieren der Connectoroptionen.

Unterstützte Optionen

Die minimale erforderliche Option ist url wie "jdbc:sqlserver://<server>:<port>;database=<database>;" oder festgelegt spark.mssql.connector.default.url.

• Wenn das url bereitgestellt wird:

  • Immer als erste Einstellung verwenden url .
  • Wenn spark.mssql.connector.default.url nicht festgelegt ist, wird der Connector es festlegen und für die zukünftige Verwendung wiederverwenden.

• Wenn dies url nicht angegeben wird:

  • Wenn spark.mssql.connector.default.url gesetzt ist, verwendet der Connector den Wert aus der Spark-Konfiguration.
  • Wenn spark.mssql.connector.default.url nicht festgelegt ist, wird ein Fehler zurückgegeben, da die benötigten Details nicht vorliegen.

Dieser Connector unterstützt die hier definierten Optionen: SQL DataSource JDBC Optionen

Der Connector unterstützt auch die folgenden Optionen:

Andere Massen-API-Optionen können als Optionen für das DataFrame festgelegt werden und beim Schreiben an Massenkopie-APIs übergeben werden.

Beispiel für Schreiben und Lesen

Der folgende Code zeigt, wie Daten unter Verwendung der mssql("<schema>.<table>")-Methode mit automatischer Microsoft Entra ID-Authentifizierung geschrieben und gelesen werden.

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

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

import com.microsoft.sqlserver.jdbc.spark.SparkSqlImplicits._
import org.apache.spark.sql.Row
import org.apache.spark.sql.types._
val url = "jdbc:sqlserver://<server>:<port>;database=<database>;"
val row_data = Seq(
  Row("Alice", 2),
  Row("Bob", 2),
  Row("Charlie", 3)
)
val schema = List(
  StructField("Name", StringType, nullable = true),
  StructField("Age", IntegerType, nullable = true)
)
val df = spark.createDataFrame(spark.sparkContext.parallelize(row_data), StructType(schema))
df.write.mode("overwrite").option("url", url).mssql("dbo.publicExample")
spark.read.option("url", url).mssql("dbo.publicExample").show

val 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

Sie können auch Spalten auswählen, Filter anwenden und andere Optionen verwenden, wenn Sie Daten aus dem SQL-Datenbankmodul lesen.

Authentifizierungsbeispiele

Die folgenden Beispiele zeigen, wie Sie andere Authentifizierungsmethoden als Microsoft Entra-ID verwenden, z. B. Dienstprinzipal (Zugriffstoken) und SQL-Authentifizierung.

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

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).option("user", "").option("password", "").mssql("dbo.publicExample")
spark.read.option("user", "").option("password", "").mssql("dbo.publicExample").show()

Unterstützte DataFrame-Speichermodi

Wenn Sie Daten aus Spark in SQL-Datenbanken schreiben, können Sie aus mehreren Speichermodi wählen. Speichermodi steuern, wie Daten geschrieben werden, wenn die Zieltabelle bereits vorhanden ist, und kann sich auf Schema, Daten und Indizierung auswirken. Wenn Sie diese Modi verstehen, können Sie unerwartete Datenverluste oder Änderungen vermeiden.

Dieser Connector unterstützt die hier definierten Optionen: Spark Save-Funktionen

• ErrorIfExists (Standardspeichermodus): Wenn die Zieltabelle vorhanden ist, wird der Schreibvorgang abgebrochen und eine Ausnahme zurückgegeben. Andernfalls wird eine neue Tabelle mit Daten erstellt.

• Ignorieren: Wenn die Zieltabelle existiert, ignoriert der Schreibvorgang die Anfrage und gibt keinen Fehler zurück. Andernfalls wird eine neue Tabelle mit Daten erstellt.

• Überschreiben: Wenn die Zieltabelle vorhanden ist, wird die Tabelle gelöscht, neu erstellt und neue Daten angefügt.

  Hinweis

  Bei Verwendung overwritegehen das ursprüngliche Tabellenschema (insbesondere MSSQL-exklusive Datentypen) und Tabellenindizes verloren und werden durch das Schema ersetzt, das von Ihrem Spark DataFrame abgeleitet wurde. Um zu vermeiden, dass Schema- und Indizes verloren gehen, verwenden Sie .option("truncate", true) anstelle von overwrite.

• Anfügen: Wenn die Zieltabelle existiert, werden neue Daten an sie angefügt. Andernfalls wird eine neue Tabelle mit Daten erstellt.

Troubleshoot

Nach Abschluss des Vorgangs wird die Ausgabe des Spark-Lesevorgangs im Ausgabebereich der Zelle angezeigt. Fehler stammen direkt aus com.microsoft.sqlserver.jdbc.SQLServerException SQL Server. Detaillierte Fehlerinformationen finden Sie in den Spark-Anwendungsprotokollen.

Verwandte Inhalte

• Azure SQL-Datenbank
• Fabric SQL-Datenbanken
• Azure SQL-Datenbank – Authentifizierung und Autorisierung