controlador JDBC de Microsoft para ingeniería de datos de Microsoft Fabric

JDBC (Conectividad de bases de datos java) es un estándar ampliamente adoptado que permite a las aplicaciones cliente conectarse a datos de bases de datos y plataformas de macrodatos y trabajar con ellos.

El controlador JDBC de Microsoft para Fabric Data Engineering te permite conectar, consultar y gestionar cargas de trabajo de Spark en Fabric con la fiabilidad y simplicidad del estándar JDBC. Construido sobre las APIs Livy de Fabric, el controlador proporciona conectividad SQL Spark segura y flexible a tus aplicaciones Java y herramientas de BI. Esta integración te permite enviar y ejecutar código de Spark directamente sin necesidad de crear elementos separados para la definición de cuadernos o trabajos de Spark. El controlador es compatible con clientes JDBC populares, como DbVisualizer y DBeaver, así como con herramientas de BI que admiten la conectividad JDBC, incluido Tableau.

Características clave

• API de JDBC 4.2: Implementa API de JDBC 4.2 para la conectividad de Spark SQL, con limitaciones de Spark y Livy, como la ausencia de transacciones JDBC y conjuntos de resultados de solo avance y solo lectura
• Autenticación de Id. de Entra de Microsoft: varios flujos de autenticación, como las credenciales interactivas, las credenciales de cliente y la autenticación basada en certificados
• Sesiones de alta concurrencia: uso compartido opcional de sesiones de Fabric para cargas de trabajo concurrentes
• Agrupación explícita de conexiones: Elige HikariCP, un gestor de pools estándar, o LivyBuiltInPooledDataSource; puntos de entrada directos nunca agrupan
• Soporte nativo de consultas SQL de Spark: Ejecución directa de sentencias SQL de Spark sin traducción
• Compatibilidad completa con tipos de datos: compatibilidad con todos los tipos de datos de Spark SQL, incluidos tipos complejos (ARRAY, MAP, STRUCT)
• Captura previa del conjunto de resultados asincrónico: carga de datos en segundo plano para mejorar el rendimiento
• Patrón Circuit Breaker: Protección contra errores en cascada con reintento automático
• Reconexión automática: recuperación de sesión transparente en errores de conexión
• Lógica de reintento avanzada: reintento con retroceso exponencial y recuperación de sesión para aumentar la resistencia
• Compatibilidad con proxy: configuración de proxy HTTP y SOCKS para entornos empresariales

Prerrequisitos

Antes de utilizar el controlador JDBC de Microsoft para Fabric Data Engineering, asegúrate de tener:

• Java Development Kit (JDK): Java 11, 17 y 21 (recomendado Java 21)
• Fabric Access: Acceso a un espacio de trabajo Fabric
• Microsoft Entra ID credenciales: credenciales adecuadas para la autenticación
• Identificadores de área de trabajo y Lakehouse: identificadores GUID para tu área de trabajo de Fabric y Lakehouse

Descarga e instalación

El controlador JDBC de Microsoft para Fabric Data Engineering versión 2.0.1 soporta Java 11, 17 y 21. Use la versión más reciente disponible.

• Descarga el controlador Microsoft JDBC para Fabric Data Engineering (zip)
• Descargar el controlador Microsoft JDBC para Fabric Data Engineering (tar)

1. Descargue el archivo zip o tar de los vínculos anteriores.
2. Extraiga el archivo descargado para acceder a los archivos JAR del controlador.
3. Seleccione el archivo JAR que coincida con la versión de JRE:
   • Para Java 11: ms-sparksql-jdbc-2.0.1.jre11.jar
   • Para Java 17: ms-sparksql-jdbc-2.0.1.jre17.jar
   • Para Java 21: ms-sparksql-jdbc-2.0.1.jre21.jar
4. Agregue el archivo JAR seleccionado a la ruta de clase de la aplicación.
5. Para los clientes JDBC, configure la clase de controlador JDBC: com.microsoft.spark.livy.jdbc.LivyDriver

Ejemplo de inicio rápido

Este ejemplo demuestra cómo conectarse a Fabric y ejecutar una consulta utilizando el controlador JDBC de Microsoft para Fabric Data Engineering. Antes de ejecutar este código, asegúrate de cumplir los requisitos previos e instalar el controlador. Al usar AuthFlow=2, el controlador usa DefaultAzureCredential. Para usar CLI de Azure como fuente de credenciales, instala CLI de Azure y ejecuta az login.

import java.sql.*;

public class QuickStartExample {
    public static void main(String[] args) {
        // Connection string with required parameters
        String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
                     "FabricWorkspaceID=<workspace-id>;" +
                     "FabricLakehouseID=<lakehouse-id>;" +
                     "AuthFlow=2;" +  // DefaultAzureCredential; Azure CLI can be one source
                     "LogLevel=INFO";
        
        try (Connection conn = DriverManager.getConnection(url)) {
            // Execute a simple query
            try (Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery("SELECT 'Hello from Fabric!' as message")) {
                
                if (rs.next()) {
                    System.out.println(rs.getString("message"));
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

Formato de cadena de conexión

Cadena de conexión básica

El controlador Microsoft JDBC para Fabric Data Engineering utiliza el siguiente formato de cadena de conexión:

jdbc:fabricspark://<hostname>[:<port>][;<parameter1>=<value1>;<parameter2>=<value2>;...]

Componentes de cadena de conexión

Cadenas de conexión de ejemplo

Conexión básica (autenticación basada en explorador interactivo)

jdbc:fabricspark://api.fabric.microsoft.com;FabricWorkspaceID=<workspace-id>;FabricLakehouseID=<lakehouse-id>;AuthFlow=1

Con la configuración de recursos de Spark

jdbc:fabricspark://api.fabric.microsoft.com;FabricWorkspaceID=<workspace-id>;FabricLakehouseID=<lakehouse-id>;DriverCores=4;DriverMemory=4g;ExecutorCores=4;ExecutorMemory=8g;NumExecutors=2;AuthFlow=2

Con las propiedades de la sesión de Spark

jdbc:fabricspark://api.fabric.microsoft.com;FabricWorkspaceID=<workspace-id>;FabricLakehouseID=<lakehouse-id>;spark.sql.adaptive.enabled=true;spark.sql.shuffle.partitions=200;AuthFlow=2

Con modo de alta concurrencia

El modo de alta concurrencia (HC) está disponible para las conexiones de Microsoft Fabric a partir de la versión 1.1.0 del controlador. Añade hcEnabled=true para obtener una sesión de HC en lugar de una sesión clásica de Livy:

jdbc:fabricspark://api.fabric.microsoft.com;FabricWorkspaceID=<workspace-id>;FabricLakehouseID=<lakehouse-id>;AuthFlow=2;hcEnabled=true

Opcionalmente puedes identificar la carga de trabajo con una etiqueta de sesión y seleccionar un entorno Fabric:

jdbc:fabricspark://api.fabric.microsoft.com;FabricWorkspaceID=<workspace-id>;FabricLakehouseID=<lakehouse-id>;AuthFlow=2;hcEnabled=true;sessionTag=orders_etl;FabricEnvironmentID=<environment-id>

El modo HC es opcional. Si hcEnabled se omite o se establece en false, el driver utiliza la ruta clásica de sesión de Livy.

Consulta la sección Modo de alta concurrencia (HC) para obtener una guía completa de configuración.

Autenticación

El controlador Microsoft JDBC para Fabric Data Engineering soporta múltiples métodos de autenticación a través de Microsoft Entra ID (anteriormente Azure Active Directory). La autenticación se configura mediante el AuthFlow parámetro en la cadena de conexión.

Flujos de autenticación

Autenticación interactiva del explorador

Ideal para: Desarrollo e aplicaciones interactivas

String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=1;" +
             "LogLevel=INFO";

Connection conn = DriverManager.getConnection(url);

Parámetros:

• AuthFlow=1: especifica la autenticación interactiva del explorador.

Comportamiento:

• Abre una ventana del explorador para la autenticación de usuario
• Almacena en caché las credenciales de las conexiones posteriores hasta que expiran
• Adecuado para aplicaciones de usuario único

Credenciales de cliente o autenticación de entidad de servicio

Mejor para: Servicios automatizados y trabajos en segundo plano

String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=3;" +  
             "AuthClientID=<client-id>;" +
             "AuthClientSecret=<client-secret>;" +
             "AuthTenantID=<tenant-id>;" +
             "LogLevel=INFO";

Connection conn = DriverManager.getConnection(url);

Parámetros necesarios:

• AuthFlow=3: especifica la autenticación de credenciales de cliente.
• AuthClientID: ID de aplicación (cliente) de Microsoft Entra ID
• AuthClientSecret: secreto de cliente de Microsoft Entra ID
• AuthTenantID: Identificador de inquilino de Microsoft Entra

Procedimientos recomendados:

• Almacenamiento de secretos de forma segura (Azure Key Vault, variables de entorno)
• Uso de identidades administradas siempre que sea posible
• Rotación periódica de secretos

Autenticación basada en certificados

Mejor para: Aplicaciones empresariales que requieren autenticación basada en certificados

String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=4;" +  
             "AuthClientID=<client-id>;" +
             "AuthCertificatePath=/path/to/certificate.pem;" +
             "AuthCertificatePassword=<certificate-password>;" +
             "AuthTenantID=<tenant-id>;" +
             "LogLevel=INFO";

Connection conn = DriverManager.getConnection(url);

Parámetros necesarios:

• AuthFlow=4: especifica la autenticación basada en certificados.
• AuthClientID: ID de aplicación (cliente)
• AuthCertificatePath: Ruta hacia un archivo de certificado PEM
• AuthCertificatePassword: contraseña de certificado
• AuthTenantID: Identificador de inquilino de Microsoft Entra

Autenticación de token de acceso

Mejor para: Escenarios de autenticación personalizados

// Acquire token through custom mechanism
String accessToken = acquireTokenFromCustomSource();

String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=5;" +  // Access token authentication
             "AuthAccessToken=" + accessToken + ";" +
             "LogLevel=INFO";

Connection conn = DriverManager.getConnection(url);

Caché de autenticación

El controlador almacena en caché los tokens de autenticación y los actualiza según la expiración del token. Las propiedades analizadas AuthEnableCaching y AuthCacheTTLMS no controlan este comportamiento en la versión 2.0.0.

Parámetros de configuración

Parámetros requeridos

Estos parámetros deben estar presentes en cada cadena de conexión:

Parámetros opcionales

Configuración de la versión de API

Configuración del entorno

Configuración de sesiones de alta concurrencia

Configura hcEnabled=true en modo HC. Las propiedades HC restantes se usan solo cuando el modo HC está activado.

Example:

hcEnabled=true;sessionTag=interactive_reporting;hcAcquireTimeoutSeconds=600;hcAcquirePollingIntervalMs=2000;FabricEnvironmentID=<environment-id>

• FabricEnvironmentID Se aplica tanto a las sesiones HC como a las clásicas
• EnvironmentID también se acepta como alias de cross-driver. Los nombres de propiedad no distinguen mayúsculas de minúsculas.

Consulta la sección Modo de alta concurrencia (HC) para obtener la guía completa de configuración.

Configuración de Spark

Configuración de recursos de sesión

Configure los recursos de sesión de Spark para obtener un rendimiento óptimo:

Example:

DriverCores=4;DriverMemory=4g;ExecutorCores=4;ExecutorMemory=8g;NumExecutors=2; SessionName=MySparkSession

Propiedades personalizadas de la sesión de Spark

Cualquier parámetro con el prefijo spark. se aplica automáticamente a la sesión de Spark:

Configuraciones de Spark de ejemplo:

spark.sql.adaptive.enabled=true
spark.sql.adaptive.coalescePartitions.enabled=true
spark.sql.shuffle.partitions=200
spark.sql.autoBroadcastJoinThreshold=10485760
spark.dynamicAllocation.enabled=true
spark.dynamicAllocation.minExecutors=1
spark.dynamicAllocation.maxExecutors=10
spark.executor.memoryOverhead=1g

Motor de ejecución nativo (NEE):

spark.nee.enabled=true

Ejemplo completo:

jdbc:fabricspark://api.fabric.microsoft.com;FabricWorkspaceID=<guid>;FabricLakehouseID=<guid>;DriverMemory=4g;ExecutorMemory=8g;NumExecutors=2;spark.sql.adaptive.enabled=true;spark.nee.enabled=true;AuthFlow=2

Configuración de conexión de cliente HTTP

Configura las conexiones de transporte HTTP del controlador para un rendimiento óptimo de la red. Estos ajustes no configuran ni gestionan la agrupación de conexiones JDBC:

Example:

HttpMaxTotalConnections=200;HttpMaxConnectionsPerRoute=100;HttpConnectionTimeoutInSeconds=60

Configuración del proxy

Configure las opciones de proxy HTTP y SOCKS para entornos empresariales:

Ejemplo de proxy HTTP:

UseProxy=true;ProxyTransport=http;ProxyHost=proxy.company.com;ProxyPort=8080;ProxyAuthEnabled=true;ProxyUsername=user;ProxyPassword=pass

Ejemplo de proxy SOCKS:

UseProxy=true;ProxyTransport=tcp;ProxyHost=socks.company.com;ProxyPort=1080;ProxySocksVersion=5

Configuración de registro

Example:

LogLevel=DEBUG

Ubicación de registro predeterminada:

${user.home}/.microsoft/livy-jdbc-driver/logs/driver.log

Configuración de registro personalizada: Use un archivo log4j2.xml o logback.xml personalizado en su ruta de clase.

Modo de alta concurrencia (HC)

El modo de alta concurrencia (HC) está diseñado para aplicaciones JDBC que abren varias conexiones al mismo espacio de trabajo de Fabric y lakehouse. En lugar de provisionar una sesión clásica separada de Spark para cada conexión, Fabric sirve a esas conexiones con capacidad compartida y gestionada por el servidor de Spark.

El modo HC puede reducir el tiempo de inicio de la conexión, evitar la provisión repetida de sesiones de Spark y utilizar la capacidad disponible de forma más eficiente para cargas de trabajo concurrentes. Cada conexión JDBC continúa ejecutando sentencias a través de su contexto asignado dentro de la sesión HC.

El modo HC es opcional y complementa la agrupación de conexiones JDBC. El modo HC gestiona las sesiones Fabric Spark en el servidor, mientras que la agrupación de conexiones JDBC gestiona las conexiones en la aplicación cliente. Si no activas el modo HC, el conductor crea una sesión clásica de Livio.

Elige entre HC y modo clásico

Usa el modo HC cuando:

• Un servicio de aplicación gestiona solicitudes concurrentes que ejecutan SQL de Spark a través de múltiples conexiones JDBC.
• Una BI o herramienta de informes abre varias conexiones al mismo espacio de trabajo y a la misma casa del lago.
• Las tareas ETL paralelas o los trabajos programados generan ráfagas de conexiones JDBC.
• Las conexiones utilizan configuraciones compatibles de Spark y pueden compartir la capacidad de Spark gestionada por el servidor.
• Reducir el tiempo de inicio de la conexión y la provisión repetida de sesiones de Spark es importante.

Usa el modo clásico cuando:

• Tu aplicación utiliza solo una o unas pocas conexiones JDBC de larga duración.
• Cada conexión requiere una sesión dedicada de Spark para una carga de trabajo estricta o aislamiento de recursos.
• Las conexiones requieren configuraciones de Spark o entornos Fabric sustancialmente diferentes.
• El endpoint de Fabric Livy para el espacio de trabajo objetivo y la casa del lago no soporta el modo HC.

Activar el modo de alta concurrencia

El modo HC es opcional. Si omites hcEnabled o lo estableces en false, el controlador usa la ruta de sesión clásica de Livy. Configura hcEnabled en verdadero para conseguir una sesión de HC:

hcEnabled=true;

Parámetros requeridos en modo HC

La activación de HC requiere los siguientes identificadores válidos:

Incluye SessionTag porque influye en la emparejación del servidor para las sesiones calientes.

Emparejamiento de sesiones

El modo HC no garantiza que todas las conexiones reutilicen una sesión existente. Para cada solicitud de adquisición de sesión, el controlador envía:

• El valor SessionTag.
• La configuración efectiva de Spark, incluida la configuración de conf.* y EnvironmentId.

El servicio Fabric Livy evalúa estos valores y decide si adjunta la conexión a una sesión caliente o crea una nueva. Para mejorar la reutilización de sesiones, utiliza valores consistentes entre conexiones que estén destinadas a compartir la capacidad de Spark gestionada por el servidor.

Desactivar el modo HC

Puedes desactivar la adquisición de nuevas sesiones de HC sin tener que volver a desplegar el controlador configurando la cadena de conexión de HC como "false" o eliminando los parámetros de HC:

hcEnabled=false;

Retrocesos de emergencia: Para retrocesos de emergencia, inicia la JVM con -Dcom.microsoft.fabric.jdbc.hc.disable=true. Este conmutador de toda la JVM obliga a usar el modo clásico incluso cuando una cadena de conexión contiene hcEnabled=true.

Comprender el comportamiento del runtime de HC

• HC bloquea la adquisición de sesiones hasta que la sesión esté lista o hcAcquireTimeoutSeconds expire.
• Cerrar la conexión mientras la adquisición está en curso interrumpe la adquisición y desencadena la limpieza de la sesión HC en modo best-effort.
• La autenticación, la ejecución de consultas, la cancelación, las sentencias preparadas, la gestión de resultados y FabricEnvironmentID funcionan a través de las interfaces estándar de JDBC en ambos modos.

Ejemplos de uso

Conexión básica

import java.sql.*;

public class BasicConnectionExample {
    public static void main(String[] args) {
        String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
                     "FabricWorkspaceID=<workspace-id>;" +
                     "FabricLakehouseID=<lakehouse-id>;" +
                     "AuthFlow=2";
        
        try (Connection conn = DriverManager.getConnection(url)) {
            System.out.println("Connected successfully!");
            System.out.println("Database: " + conn.getMetaData().getDatabaseProductName());
            System.out.println("Driver: " + conn.getMetaData().getDriverName());
            System.out.println("Driver Version: " + conn.getMetaData().getDriverVersion());
        } catch (SQLException e) {
            System.err.println("Connection failed: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

Conexión en modo de alta concurrencia

El siguiente ejemplo abre una conexión JDBC en modo HC, ejecuta una consulta y libera la conexión automáticamente:

String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=2;" +
             "hcEnabled=true;" +
             "sessionTag=dashboard_queries;" +
             "FabricEnvironmentID=<environment-id>";

try (Connection conn = DriverManager.getConnection(url);
     Statement stmt = conn.createStatement();
     ResultSet rs = stmt.executeQuery("SELECT current_timestamp()")) {
    while (rs.next()) {
        System.out.println(rs.getTimestamp(1));
    }
}

Ejecución de consultas

Consulta simple

public void executeSimpleQuery(Connection conn) throws SQLException {
    String sql = "SELECT current_timestamp() as now";
    
    try (Statement stmt = conn.createStatement();
         ResultSet rs = stmt.executeQuery(sql)) {
        
        if (rs.next()) {
            Timestamp now = rs.getTimestamp("now");
            System.out.println("Current timestamp: " + now);
        }
    }
}

Consulta con filtro

public void executeQueryWithFilter(Connection conn) throws SQLException {
    String sql = "SELECT * FROM sales WHERE amount > 1000 ORDER BY amount DESC";
    
    try (Statement stmt = conn.createStatement();
         ResultSet rs = stmt.executeQuery(sql)) {
        
        while (rs.next()) {
            int id = rs.getInt("id");
            double amount = rs.getDouble("amount");
            Date date = rs.getDate("sale_date");
            
            System.out.printf("ID: %d, Amount: %.2f, Date: %s%n", 
                            id, amount, date);
        }
    }
}

Consulta con límite

public void executeQueryWithLimit(Connection conn) throws SQLException {
    String sql = "SELECT * FROM customers LIMIT 10";
    
    try (Statement stmt = conn.createStatement();
         ResultSet rs = stmt.executeQuery(sql)) {
        
        ResultSetMetaData metaData = rs.getMetaData();
        int columnCount = metaData.getColumnCount();
        
        // Print column names
        for (int i = 1; i <= columnCount; i++) {
            System.out.print(metaData.getColumnName(i) + "\t");
        }
        System.out.println();
        
        // Print rows
        while (rs.next()) {
            for (int i = 1; i <= columnCount; i++) {
                System.out.print(rs.getString(i) + "\t");
            }
            System.out.println();
        }
    }
}

Trabajar con conjuntos de resultados

Recorrer conjuntos de resultados

public void navigateResultSet(Connection conn) throws SQLException {
    String sql = "SELECT id, name, amount FROM orders";
    
    try (Statement stmt = conn.createStatement();
         ResultSet rs = stmt.executeQuery(sql)) {

        while (rs.next()) {
            System.out.printf(
                "Order %d: %s, %.2f%n",
                rs.getInt("id"),
                rs.getString("name"),
                rs.getDouble("amount"));
        }
    }
}

Los conjuntos de resultados son de solo avance y de solo lectura. El controlador no soporta métodos como first(), last(), y absolute().

Procesamiento de grandes conjuntos de resultados

El controlador pagina los resultados por tamaño de respuesta. Configura LivyStatementPageSize en la URL de JDBC para establecer el tamaño de página solicitado en bytes. El Statement.setFetchSize() método no controla la recuperación ni el uso de memoria.

jdbc:fabricspark://api.fabric.microsoft.com;FabricWorkspaceID=<workspace-id>;FabricLakehouseID=<lakehouse-id>;AuthFlow=2;LivyStatementPageSize=1048576

public void processLargeResultSet(Connection conn) throws SQLException {
    String sql = "SELECT * FROM large_table";
    
    try (Statement stmt = conn.createStatement()) {
        try (ResultSet rs = stmt.executeQuery(sql)) {
            int rowCount = 0;
            while (rs.next()) {
                // Process row
                processRow(rs);
                rowCount++;
                
                if (rowCount % 10000 == 0) {
                    System.out.println("Processed " + rowCount + " rows");
                }
            }
            System.out.println("Total rows processed: " + rowCount);
        }
    }
}

private void processRow(ResultSet rs) throws SQLException {
    // Process individual row
}

Uso de instrucciones preparadas

public void usePreparedStatement(Connection conn) throws SQLException {
    String sql = "SELECT * FROM products WHERE category = ? AND price > ?";
    
    try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
        // Set parameters
        pstmt.setString(1, "Electronics");
        pstmt.setDouble(2, 100.0);
        
        try (ResultSet rs = pstmt.executeQuery()) {
            while (rs.next()) {
                String name = rs.getString("name");
                double price = rs.getDouble("price");
                System.out.printf("Product: %s, Price: $%.2f%n", name, price);
            }
        }
    }
}

Operaciones por lotes

El controlador no soporta transacciones JDBC. La ejecución por lotes no es transaccional y puede tener éxito parcial. Un fallo posterior no revierte declaraciones anteriores exitosas.

public void executeBatchInsert(Connection conn) throws SQLException {
    String sql = "INSERT INTO logs (timestamp, level, message) VALUES (?, ?, ?)";
    int pendingStatements = 0;
    
    try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
        // Add multiple statements to batch
        for (int i = 0; i < 1000; i++) {
            pstmt.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
            pstmt.setString(2, "INFO");
            pstmt.setString(3, "Log message " + i);
            pstmt.addBatch();
            pendingStatements++;
            
            // Execute batch every 100 statements
            if ((i + 1) % 100 == 0) {
                pstmt.executeBatch();
                pstmt.clearBatch();
                pendingStatements = 0;
            }
        }
        
        if (pendingStatements > 0) {
            pstmt.executeBatch();
        }
        
        System.out.println("Batch insert completed successfully");
    }
}

Elige un propietario de la conexión

Elige un componente para controlar el ciclo de vida de la conexión. No coloques un pool de conexiones dentro de otro.

ConnectionPoolEnabled tiene como valor predeterminado false tanto en el código del controlador como en las propiedades incluidas. No es un interruptor de activación voluntaria para los puntos de acceso actuales. Use LivyBuiltInPooledDataSource cuando el controlador deba gestionar el pool, o configure un pool externo como HikariCP. Los otros puntos de entrada crean conexiones directas que no son propiedad del pool.

Agrupación de conexiones con HikariCP

HikariCP puede crear conexiones a partir de LivyDataSource o de la URL JDBC mediante LivyDriver. Crea una instancia de HikariCP, reutilízala durante la vida útil de la aplicación y usa solo un conjunto de conexiones en cualquiera de las configuraciones.

Dependencia de Maven

<dependency>
    <groupId>com.zaxxer</groupId>
    <artifactId>HikariCP</artifactId>
    <version>5.0.1</version>
</dependency>

Utilice LivyDataSource

LivyDataSource crea conexiones físicas independientes y no posee un pool de conexiones ni un ciclo de vida cerrado separado. Entrégalo directamente a HikariCP para que HikariCP sea el único propietario del pool:

import com.microsoft.spark.livy.jdbc.LivyDataSource;
import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

public final class ConnectionPoolExample implements AutoCloseable {
    private final HikariDataSource hikariDataSource;

    public ConnectionPoolExample(String url) {
        LivyDataSource configuredDataSource = new LivyDataSource();
        configuredDataSource.setUrl(url);

        HikariConfig config = new HikariConfig();
        config.setDataSource(configuredDataSource);
        config.setMaximumPoolSize(2);
        config.setMinimumIdle(0);
        config.setConnectionTimeout(900000);     // Wait up to 15 minutes for a connection
        config.setInitializationFailTimeout(-1); // Connect on demand
        config.setIdleTimeout(600000);           // 10 minutes
        config.setMaxLifetime(1800000);          // 30 minutes
        config.setPoolName("FabricSparkPool");

        hikariDataSource = new HikariDataSource(config);
    }

    public Connection getConnection() throws SQLException {
        return hikariDataSource.getConnection();
    }

    @Override
    public void close() {
        hikariDataSource.close();
    }

    public static void main(String[] args) throws SQLException {
        String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
                     "FabricWorkspaceID=<workspace-id>;" +
                     "FabricLakehouseID=<lakehouse-id>;" +
                     "AuthFlow=AZURE_CLI;" +
                     "hcEnabled=true;" +
                     "sessionTag=pooled_app;" +
                     "hcAcquireTimeoutSeconds=600";

        try (ConnectionPoolExample pool = new ConnectionPoolExample(url);
             Connection conn = pool.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery("SELECT 'Pooled connection!' as message")) {

            if (rs.next()) {
                System.out.println(rs.getString("message"));
            }
        }
    }
}

El ejemplo activa el modo HC en la URL de JDBC. HikariCP gestiona las conexiones del lado del cliente, mientras que el modo HC gestiona las sesiones compartidas de Spark en Fabric. Elimine hcEnabled y sessionTag si la aplicación requiere sesiones clásicas.

Utiliza LivyDriver directamente

Configura la URL JDBC y la clase de controlador para que HikariCP cree conexiones físicas a través de LivyDriver. Configurar explícitamente la clase del controlador es opcional cuando hay disponible el autodescubrimiento del proveedor de servicios JDBC. Esta configuración no crea una instancia de LivyDataSource, así que su grupo integrado no interviene:

import com.zaxxer.hikari.HikariConfig;
import com.zaxxer.hikari.HikariDataSource;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;

public final class DriverConnectionPoolExample {
    public static HikariDataSource createPool(String url) {
        HikariConfig config = new HikariConfig();
        config.setDriverClassName("com.microsoft.spark.livy.jdbc.LivyDriver");
        config.setJdbcUrl(url);

        // Keep pool sizes small: each classic-mode physical connection owns a Livy session.
        config.setMaximumPoolSize(2);
        config.setMinimumIdle(0);
        config.setConnectionTimeout(900000);     // Wait up to 15 minutes to borrow a connection
        config.setInitializationFailTimeout(-1); // Skip startup validation; connect on demand
        config.setIdleTimeout(600000);           // 10 minutes
        config.setMaxLifetime(1800000);          // 30 minutes
        config.setPoolName("FabricSparkDriverPool");

        return new HikariDataSource(config);
    }

    public static void main(String[] args) throws SQLException {
        String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
                     "FabricWorkspaceID=<workspace-id>;" +
                     "FabricLakehouseID=<lakehouse-id>;" +
                     "AuthFlow=AZURE_CLI;" +            // Uses the DefaultAzureCredential chain
                     "hcEnabled=true;" +
                     "sessionTag=pooled_app;" +
                     "hcAcquireTimeoutSeconds=600";  // Poll up to 10 minutes for HC readiness

        // Reuse this pool for the application lifetime; main closes it at process exit.
        try (HikariDataSource pool = createPool(url);
             Connection conn = pool.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery("SELECT 'Pooled LivyDriver connection!' as message")) {

            if (rs.next()) {
                System.out.println(rs.getString("message"));
            }
        }
    }
}

Utiliza las siguientes indicaciones al configurar el pool:

Para la lista completa de opciones de configuración de HikariCP, véase:

• Referencia de la API HikariConfig
• Opciones de configuración de HikariCP

Asignación de tipos de datos

El controlador asigna los tipos de datos de Spark SQL a los tipos sql de JDBC y a los tipos de Java:

Contenido relacionado

• Runtimes de Apache Spark en Fabric
• Fabric Runtime 1.3
• ¿Qué es Livy API para ingeniería de datos?