Conector de Spark para bases de datos SQL

El conector Spark para bases de datos SQL es una biblioteca de alto rendimiento que te permite leer y escribir en bases de datos de SQL Server, Azure SQL y bases de datos SQL en Fabric. El conector ofrece las siguientes capacidades:

• Utiliza Spark para ejecutar grandes operaciones de escritura y lectura en Azure SQL Database, Azure SQL Managed Instance, SQL Server en Azure VM y bases de datos SQL en Fabric.
• Cuando se usa una tabla o una vista, el conector admite modelos de seguridad establecidos en el nivel del motor de SQL. Estos modelos incluyen seguridad de nivel de objeto (OLS), seguridad de nivel de fila (RLS) y seguridad de nivel de columna (CLS).

El conector está preinstalado en el entorno de ejecución de Fabric, por lo que no es necesario instalarlo por separado.

Autenticación

La autenticación de Microsoft Entra está integrada con Fabric.

• Al iniciar sesión en el área de trabajo de Fabric, las credenciales se pasan automáticamente al motor de SQL para la autenticación y autorización.
• Requiere que Microsoft Entra ID esté habilitado y configurado en el motor de base de datos SQL.
• No se necesita ninguna configuración adicional en el código de Spark si está configurado el identificador de Entra de Microsoft. Las credenciales se asignan automáticamente.

También puede usar el método de autenticación de SQL (especificando un nombre de usuario y una contraseña de SQL) o un principal de servicio (proporcionando un token de acceso de Azure para la autenticación de aplicaciones).

Permissions

Para usar el conector de Spark, la identidad (ya sea un usuario o una aplicación) debe tener los permisos de base de datos necesarios para el motor SQL de destino. Estos permisos son necesarios para leer o escribir en tablas y vistas.

Para Azure SQL Database, Azure SQL Managed Instance y SQL Server en una máquina virtual de Azure:

• La identidad que ejecuta la operación normalmente necesita permisos de db_datawriter y de db_datareader, y opcionalmente db_owner para el control total.

Para una base de datos SQL en Fabric:

• La identidad normalmente necesita db_datawriter y db_datareader permisos y, opcionalmente, db_owner.
• La identidad también necesita al menos permisos de lectura en la base de datos SQL de Fabric a nivel de elemento.

Ejemplos de uso y código

En esta sección, se proporcionan ejemplos de código para demostrar cómo usar el conector de Spark para bases de datos SQL de forma eficaz. En estos ejemplos se tratan varios escenarios, incluida la lectura y escritura en tablas SQL, y la configuración de las opciones del conector.

Opciones admitidas

La opción mínima requerida es url como "jdbc:sqlserver://<server>:<port>;database=<database>;" o configurar spark.mssql.connector.default.url.

• Cuando se proporciona url:

  • url Use siempre como primera preferencia.
  • Si spark.mssql.connector.default.url no se establece, el conector lo establecerá y lo reutilizará para su uso futuro.

• Cuando no se proporciona url:

  • Si spark.mssql.connector.default.url se establece, el conector usa el valor de la configuración de Spark.
  • Si spark.mssql.connector.default.url no se establece, se produce un error porque los detalles necesarios no están disponibles.

Este conector admite las opciones definidas aquí: Opciones JDBC de SQL DataSource

El conector también admite las siguientes opciones:

Otras opciones de API en masa se pueden establecer como opciones en DataFrame y se pasan a APIs de copia en masa al escribir.

Ejemplo de escritura y lectura

El siguiente código utiliza autenticación automática de Microsoft Entra ID para demostrar estas operaciones:

• Escribe un DataFrame: df.write.option("...", "...").mssql("<schema>.<table>").
• Lee una tabla: spark.read.option("...", "...").mssql("<schema>.<table>").
• Ejecuta una consulta personalizada: spark.read.option("...", "...").option("query", "<your-custom-query>").mssql().

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

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", 1),
  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
spark.read.option("url", url).option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show // Read with a custom query

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

También puede seleccionar columnas, aplicar filtros y usar otras opciones al leer datos del motor de base de datos SQL.

Ejemplos de autenticación

En los ejemplos siguientes se muestra cómo usar métodos de autenticación distintos de Microsoft Entra ID, como la entidad de servicio (token de acceso) y la autenticación de SQL.

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

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()
spark.read.option("user", "").option("password", "").option("query", "SELECT * FROM dbo.publicExample WHERE Age = 3").mssql().show() # Read with a custom query

Modos de guardado de DataFrame compatibles

Al escribir datos de Spark en bases de datos SQL, puede elegir entre varios modos de guardado. Los modos de guardado controlan cómo se escriben los datos cuando la tabla de destino ya existe y pueden afectar al esquema, los datos y la indexación. Comprender estos modos le ayuda a evitar cambios o pérdida de datos inesperados.

Este conector admite las opciones definidas aquí: Funciones de guardado de Spark

• ErrorIfExists (modo de guardado predeterminado): si existe la tabla de destino, se anula la escritura y se devuelve una excepción. De lo contrario, se crea una nueva tabla con datos.

• Omitir: si existe la tabla de destino, la escritura omite la solicitud y no devuelve un error. De lo contrario, se crea una nueva tabla con datos.

• Sobrescribir: Si la tabla de destino existe, se elimina, se recrea y se anexan nuevos datos.

  Nota:

  Cuando usas overwrite, pierdes el esquema original de la tabla (especialmente los tipos de datos exclusivos de MSSQL) y los índices de la tabla. El esquema se reemplaza por el esquema inferido de tu DataFrame de Spark. Para evitar perder el esquema y los índices, añadir .option("truncate", true).

• Anexar: si la tabla de destino existe, se anexan nuevos datos a ella. De lo contrario, se crea una nueva tabla con datos.

Troubleshoot

Cuando finaliza el proceso, la salida de la operación de lectura de Spark aparece en el área de salida de la celda. Los errores de com.microsoft.sqlserver.jdbc.SQLServerException proceden directamente de SQL Server. Puede encontrar información detallada sobre los errores en los registros de la aplicación Spark.

Las escrituras masivas requieren que los datos entrantes encajen con los tipos de datos SQL objetivo. Si los datos no encajan, puede que recibas este error:

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

Por ejemplo, este error puede producirse cuando la tabla SQL de destino usa datetime, pero un valor de Spark TimestampType de entrada, como 2025-01-01 10:30:00.123456, tiene más precisión de la que admite datetime.

Para resolver el error, utiliza uno de estos enfoques:

• Convierte, trunca o transforma los datos entrantes para ajustarlos al tipo de datos SQL de destino. Por ejemplo, trunque el valor a tres dígitos de precisión de fracción de segundo: 2025-01-01 10:30:00.123000.
• Permite que el conector recree el esquema de la tabla configurando .option("truncate", false). El conector mapea el tipo de marca de tiempo de Spark a SQL datetime2.

Contenido relacionado

• Azure SQL Database
• Base de datos SQL de Fabric
• Azure SQL Database: autenticación y autorización