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

• Compatible con JDBC 4.2: implementación completa de la especificación JDBC 4.2
• 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
• Integración con HikariCP: Utiliza HikariCP para gestionar y reutilizar conexiones JDBC para aplicaciones de producción
• Compatibilidad con consultas nativas de Spark SQL: ejecución directa de instrucciones 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:

• Kit de desarrollo de Java (JDK):versión 11 o posterior (se recomienda 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 1.0.0 soporta Java 11, 17 y 21. Estamos mejorando continuamente la compatibilidad con la conectividad de Java y le recomendamos que trabaje con la versión más reciente del controlador JDBC de Microsoft.

• 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-1.0.1.jre11.jar
   • Para Java 17: ms-sparksql-jdbc-1.0.1.jre17.jar
   • Para Java 21: ms-sparksql-jdbc-1.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 usando el controlador JDBC de Microsoft para Fabric Data Engineering. Antes de ejecutar este código, asegúrese de que ha completado los requisitos previos e instalado el controlador.

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;" +  // Azure CLI based authentication
                     "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

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;" +  
             "AuthTenantID=<tenant-id>;" +  // Optional
             "LogLevel=INFO";

Connection conn = DriverManager.getConnection(url);

Parámetros:

• AuthFlow=1: especifica la autenticación interactiva del explorador.
• AuthTenantID (opcional): id. de inquilino de Microsoft Entra
• AuthClientID (opcional): Id. de aplicación (cliente)

Comportamiento:

• Abre una ventana del explorador para la autenticación de usuario
• Las credenciales se almacenan en caché para 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.pfx;" +
             "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 de acceso al archivo de certificado PFX/PKCS12
• 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);

Almacenamiento en caché de autenticación

El controlador almacena automáticamente en caché los tokens de autenticación para mejorar el rendimiento:

// Enable/disable caching (enabled by default)
String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=2;" +
             "AuthEnableCaching=true;" +  // Enable token caching
             "AuthCacheTTLMS=3600000";    // Cache TTL: 1 hour

Connection conn = DriverManager.getConnection(url);

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

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/driver.log

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

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

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

Navegar por conjuntos de resultados

public void navigateResultSet(Connection conn) throws SQLException {
    String sql = "SELECT id, name, amount FROM orders";
    
    try (Statement stmt = conn.createStatement(
            ResultSet.TYPE_SCROLL_INSENSITIVE,
            ResultSet.CONCUR_READ_ONLY);
         ResultSet rs = stmt.executeQuery(sql)) {
        
        // Move to first row
        if (rs.first()) {
            System.out.println("First row: " + rs.getString("name"));
        }
        
        // Move to last row
        if (rs.last()) {
            System.out.println("Last row: " + rs.getString("name"));
            System.out.println("Total rows: " + rs.getRow());
        }
        
        // Move to specific row
        if (rs.absolute(5)) {
            System.out.println("Row 5: " + rs.getString("name"));
        }
    }
}

Procesamiento de grandes conjuntos de resultados

public void processLargeResultSet(Connection conn) throws SQLException {
    String sql = "SELECT * FROM large_table";
    
    try (Statement stmt = conn.createStatement()) {
        // Set fetch size for efficient memory usage
        stmt.setFetchSize(1000);
        
        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

public void executeBatchInsert(Connection conn) throws SQLException {
    String sql = "INSERT INTO logs (timestamp, level, message) VALUES (?, ?, ?)";
    
    try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
        conn.setAutoCommit(false);  // Disable auto-commit for batch
        
        // 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();
            
            // Execute batch every 100 statements
            if (i % 100 == 0) {
                pstmt.executeBatch();
                pstmt.clearBatch();
            }
        }
        
        // Execute remaining statements
        pstmt.executeBatch();
        conn.commit();
        
        System.out.println("Batch insert completed successfully");
    } catch (SQLException e) {
        conn.rollback();
        throw e;
    } finally {
        conn.setAutoCommit(true);
    }
}

Agrupación de conexiones con HikariCP

Configura HikariCP con la URL JDBC para que cree conexiones a través de LivyDriver. Crea una instancia de HikariCP y reutilízala durante toda la vida útil de la aplicación.

Dependencia de Maven

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

Configurar HikariCP con la URL JDBC

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 de controlador es opcional cuando el autodescubrimiento del proveedor de servicios JDBC está disponible:

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
                     "LivySessionTimeoutSeconds=600";  // Poll up to 10 minutes for session 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?