Conector de Apache Spark: SQL Server y Azure SQL

El conector de Apache Spark para SQL Server y Azure SQL es un conector de alto rendimiento que puede usar para incluir datos transaccionales en el análisis de macrodatos y conservar los resultados para consultas ad hoc o informes. Mediante el conector, puede usar cualquier base de datos SQL, local o en la nube, como origen de datos de entrada o receptor de datos de salida para trabajos de Spark.

Esta biblioteca contiene el código fuente del conector de Apache Spark para las plataformas SQL Server y Azure SQL.

Apache Spark es un motor de análisis unificado para el procesamiento de datos a gran escala.

Importa el conector usando la coordenada que coincida con tu versión de Spark:

Los conectores Spark 3.3.x y Spark 3.4.x son versiones beta, y ninguno alcanzó disponibilidad general antes de que el proyecto fuera archivado. La versión 1.3.0-BETA no es compatible con el controlador JDBC de Microsoft para SQL Server 7.0.1, Spark 2.4 ni Spark 3.0. Las notas de la versión upstream 1.3.0 te dicen que importes la versión 1.3.0, pero esa versión nunca se publicó. Utilice 1.3.0-BETA en su lugar.

También puede compilar el conector desde el origen o descargar el archivo JAR desde la sección Versión de GitHub. Para el código fuente y el historial de versiones, consulta el repositorio archivado de SQL Spark Connector en GitHub.

Características compatibles

• Compatibilidad con todos los enlaces de Spark (Scala, Python, R)
• Compatibilidad con la autenticación básica y la pestaña de claves de Active Directory (AD)
• Compatibilidad con escritura reordenada dataframe
• Compatibilidad con la escritura en una instancia única de SQL Server y un grupo de datos en clústeres de macrodatos de SQL Server
• Compatibilidad de conectores fiables con Instancia Única de SQL Server

Opciones admitidas

El conector de Apache Spark para SQL Server y Azure SQL admite las opciones definidas en el artículo JDBC de SQL DataSource .

Además, el conector admite las siguientes opciones:

Establezca otras opciones de copia masiva como ajustes adicionales en el dataframe. El conector pasa estas opciones a las bulkcopy API en escritura.

Comparación del rendimiento

El conector apache Spark para SQL Server y Azure SQL es hasta 15 veces más rápido que el conector JDBC genérico para escribir en SQL Server. Las características de rendimiento varían según el tipo, el volumen de datos, las opciones usadas y podrían mostrar variaciones entre cada ejecución. Los siguientes resultados de rendimiento son el tiempo necesario para sobrescribir una tabla SQL con 143,9M filas en spark dataframe. Spark se construye leyendo tabla HDFS generada mediante dataframe. Se excluye el tiempo de lectura store_sales hasta dataframe . Los resultados se promedian en tres ciclos.

Configuración

• Configuración de Spark: num_executors = 20, executor_memory = "1664 m", executor_cores = 2
• Configuración de Data Gen: factor_de_escala=50, tablas_partitionadas=true
• store_sales Archivo de datos con número de filas 143.997.590

Medio ambiente

• Clúster de macrodatos de SQL Server CU5
• master + 6 nodos
• Cada nodo es un servidor gen-5, con 512 GB de RAM, NVM de 4 TB por nodo y NIC de 10 Gbps

Problemas a los que se suele enfrentar

java.lang.NoClassDefFoundError: com/microsoft/aad/adal4j/AuthenticationException

Este error se produce cuando se usa una versión anterior del mssql controlador en el entorno de Hadoop. El conector ahora incluye este controlador. Si anteriormente usaba el conector de Azure SQL y los controladores instalados manualmente en el clúster para la compatibilidad de autenticación de Microsoft Entra, quite esos controladores.

Para corregir el error:

1. Si usa un entorno genérico de Hadoop, compruebe y quite el mssql archivo JAR con el siguiente comando: rm $HADOOP_HOME/share/hadoop/yarn/lib/mssql-jdbc-6.2.1.jre7.jar. Si usa Databricks, agregue un script de inicialización global o de clúster para quitar las versiones anteriores del mssql controlador de la /databricks/jars carpeta o agregue esta línea a un script existente: rm /databricks/jars/*mssql*

2. Agregue los paquetes adal4j y mssql. Por ejemplo, puede usar Maven, pero cualquier manera debería funcionar.

   Precaución

   No instale el conector de SQL Spark de esta manera.

3. Agregue la clase de controlador a la configuración de conexión. Por ejemplo:

   connectionProperties = {
     `Driver`: `com.microsoft.sqlserver.jdbc.SQLServerDriver`
   }`
   

Para obtener más información, consulte la resolución a https://github.com/microsoft/sql-spark-connector/issues/26.

Comienza

El conector de Apache Spark para SQL Server y Azure SQL se basa en la API Spark DataSourceV1 y la API masiva de SQL Server. Usa la misma interfaz que el conector JDBC integrado Spark-SQL. Con esta integración, puede integrar fácilmente el conector y migrar los trabajos de Spark existentes mediante la actualización del parámetro format con com.microsoft.sqlserver.jdbc.spark.

Para incluir el conector en los proyectos, descargue este repositorio y compile el archivo JAR mediante SBT.

Escribir en una nueva tabla SQL

Si usa el modo overwrite sin la opción truncate al volver a crear la tabla, la operación quita los índices. Además, una tabla almacenada en columnas se convierte en una tabla amontonada. Para mantener los índices existentes, establezca la opción truncate a true. Por ejemplo: .option("truncate","true").

server_name = "jdbc:sqlserver://{SERVER_ADDR}"
database_name = "database_name"
url = server_name + ";" + "databaseName=" + database_name + ";"

table_name = "table_name"
username = "username"
password = "password123!#" # Please specify password here

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("overwrite") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Anexar a la tabla SQL

try:
  df.write \
    .format("com.microsoft.sqlserver.jdbc.spark") \
    .mode("append") \
    .option("url", url) \
    .option("dbtable", table_name) \
    .option("user", username) \
    .option("password", password) \
    .save()
except ValueError as error :
    print("Connector write failed", error)

Especificar el nivel de aislamiento

Este conector usa el READ_COMMITTED nivel de aislamiento de forma predeterminada cuando inserta datos de forma masiva en la base de datos. Para invalidar el nivel de aislamiento, use la mssqlIsolationLevel opción :

    .option("mssqlIsolationLevel", "READ_UNCOMMITTED") \

Leer desde la tabla SQL

jdbcDF = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("user", username) \
        .option("password", password).load()

Autenticación de Microsoft Entra

Ejemplo de Python con principal de servicio

context = adal.AuthenticationContext(authority)
token = context.acquire_token_with_client_credentials(resource_app_id_url, service_principal_id, service_principal_secret)
access_token = token["accessToken"]

jdbc_db = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("accessToken", access_token) \
        .option("encrypt", "true") \
        .option("hostNameInCertificate", "*.database.windows.net") \
        .load()

Ejemplo de Python con contraseña de Active Directory

jdbc_df = spark.read \
        .format("com.microsoft.sqlserver.jdbc.spark") \
        .option("url", url) \
        .option("dbtable", table_name) \
        .option("authentication", "ActiveDirectoryPassword") \
        .option("user", user_name) \
        .option("password", password) \
        .option("encrypt", "true") \
        .option("hostNameInCertificate", "*.database.windows.net") \
        .load()

Para autenticarse mediante Active Directory, instale la dependencia necesaria.

Cuando se usa ActiveDirectoryPassword, el user valor debe estar en el formato UPN, como username@domainname.com.

Para Scala, instale el com.microsoft.aad.adal4j artefacto.

Para Python, instale la adal biblioteca. Esta biblioteca está disponible a través de pip.

Para obtener ejemplos, consulte los cuadernos de ejemplo.

Contenido relacionado

• Repositorio de GitHub del conector de SQL Spark