Microsoft pilote JDBC pour l'ingénierie des données de Microsoft Fabric

JDBC (Java Database Connectivity) est une norme largement adoptée qui permet aux applications clientes de se connecter et d’utiliser des données à partir de bases de données et de plateformes Big Data.

Le pilote Microsoft JDBC pour l’ingénierie des données Fabric vous permet de connecter, interroger et gérer des charges de travail Spark dans Fabric avec la fiabilité et la simplicité de la norme JDBC. Basé sur les API Livy de Fabric, le pilote offre une connectivité SQL Spark sécurisée et flexible à vos applications Java et outils BI. Cette intégration vous permet de soumettre et d’exécuter du code Spark directement sans avoir besoin de créer des éléments de définition de notebooks ou de tâches Spark séparés. Le pilote est compatible avec les clients JDBC populaires tels que DbVisualizer et DBeaver, ainsi que les outils décisionnels qui prennent en charge la connectivité JDBC, y compris Tableau.

Principales fonctionnalités

• Conforme JDBC 4.2 : implémentation complète de la spécification JDBC 4.2
• Authentification d’ID Microsoft Entra : flux d’authentification multiples, notamment les informations d’identification interactives, les informations d’identification du client et l’authentification basée sur des certificats
• Intégration HikariCP : Utilisez HikariCP pour gérer et réutiliser les connexions JDBC pour des applications de production
• Prise en charge des requêtes natives Spark SQL : exécution directe d’instructions Spark SQL sans traduction
• Prise en charge complète des types de données : prise en charge de tous les types de données Spark SQL, y compris les types complexes (ARRAY, MAP, STRUCT)
• Prérécupération asynchrone du jeu de résultats : chargement des données en arrière-plan pour améliorer les performances
• Modèle disjoncteur : protection contre les défaillances en cascade avec nouvelle tentative automatique
• Reconnexion automatique : récupération de session transparente sur les échecs de connexion
• Logique de nouvelle tentative avancée : nouvelle tentative avec recul exponentiel et récupération de session pour améliorer la résilience
• Prise en charge du proxy : configuration du proxy HTTP et SOCKS pour les environnements d’entreprise

Prerequisites

Avant d’utiliser le pilote Microsoft JDBC pour Fabric Data Engineering, assurez-vous d’avoir :

• Kit de développement Java (JDK) : version 11 ou ultérieure (Java 21 recommandé)
• Fabric Access : Accès à un espace de travail Fabric
• informations d’identification Microsoft Entra ID : informations d’identification appropriées pour l’authentification
• Identifiants d’espace de travail et de Lakehouse : identificateurs GUID pour votre espace de travail Fabric et votre Lakehouse.

Téléchargement et installation

Le pilote Microsoft JDBC pour Fabric Data Engineering version 1.0.0 prend en charge Java 11, 17 et 21. Nous améliorons continuellement la prise en charge de la connectivité Java et vous recommandons d’utiliser la dernière version du pilote Microsoft JDBC.

• Téléchargez le pilote Microsoft JDBC pour Fabric Data Engineering (zip)
• Téléchargez le pilote Microsoft JDBC pour Fabric Data Engineering (tar)

1. Téléchargez le fichier zip ou tar à partir des liens ci-dessus.
2. Extrayez le fichier téléchargé pour accéder aux fichiers JAR du pilote.
3. Sélectionnez le fichier JAR qui correspond à votre version JRE :
   • Pour Java 11 : ms-sparksql-jdbc-1.0.1.jre11.jar
   • Pour Java 17 : ms-sparksql-jdbc-1.0.1.jre17.jar
   • Pour Java 21 : ms-sparksql-jdbc-1.0.1.jre21.jar
4. Ajoutez le fichier JAR sélectionné au chemin de classe de votre application.
5. Pour les clients JDBC, configurez la classe de pilote JDBC : com.microsoft.spark.livy.jdbc.LivyDriver

Exemple de démarrage rapide

Cet exemple montre comment se connecter à Fabric et exécuter une requête en utilisant le pilote Microsoft JDBC pour Fabric Data Engineering. Avant d’exécuter ce code, vérifiez que vous avez rempli les prérequis et installé le pilote.

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

Format de chaîne de connexion

Chaîne de connexion de base

Le pilote Microsoft JDBC pour l’ingénierie des données Fabric utilise le format de chaîne de connexion suivant :

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

Composants de chaîne de connexion

Exemples de chaînes de connexion

Connexion de base (authentification interactive basée sur un navigateur)

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

Avec la configuration des ressources 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

Avec les propriétés de session 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

Authentication

Le pilote Microsoft JDBC pour l’ingénierie des données Fabric prend en charge plusieurs méthodes d’authentification via Microsoft Entra ID (anciennement Azure Active Directory). L’authentification est configurée à l’aide du AuthFlow paramètre dans la chaîne de connexion.

Flux d’authentification

Authentification interactive du navigateur

Idéal pour : Développement et applications interactives

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

Paramètres :

• AuthFlow=1: spécifie l’authentification interactive du navigateur
• AuthTenantID (facultatif) : ID client Microsoft Entra
• AuthClientID (facultatif) : ID d’application (client)

Comportement:

• Ouvre une fenêtre de navigateur pour l’authentification utilisateur
• Les informations d’identification sont mises en cache pour les connexions suivantes jusqu’à ce qu’elles expirent
• Adapté aux applications mono-utilisateur

Informations d’identification du client ou authentification du principal de service

Idéal pour : services automatisés et travaux en arrière-plan

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

Paramètres obligatoires :

• AuthFlow=3: spécifie l’authentification des informations d’identification du client
• AuthClientID: ID d’application (client) de Microsoft Entra ID
• AuthClientSecret: Secret client de l'ID Microsoft Entra
• AuthTenantID : ID de locataire Microsoft Entra

Meilleures pratiques :

• Stocker les secrets en toute sécurité (Azure Key Vault, variables d’environnement)
• Utiliser des identités managées lorsque cela est possible
• Changer régulièrement les secrets

Authentification par certificat

Idéal pour : les applications d’entreprise nécessitant une authentification basée sur des certificats

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

Paramètres obligatoires :

• AuthFlow=4: spécifie l’authentification basée sur un certificat
• AuthClientID: ID d’application (client)
• AuthCertificatePath: chemin d’accès au fichier de certificat PFX/PKCS12
• AuthCertificatePassword: Mot de passe du certificat
• AuthTenantID : ID de locataire Microsoft Entra

Authentification par jeton d’accès

Idéal pour : scénarios d’authentification personnalisés

// 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);

Mise en cache de l’authentification

Le pilote met automatiquement en cache les jetons d’authentification pour améliorer les performances :

// 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);

Paramètres de configuration

Paramètres requis

Ces paramètres doivent être présents dans chaque chaîne de connexion :

Paramètres facultatifs

Configuration de la version de l’API

Configuration de l’environnement

Configuration de l’étincelle

Configuration des ressources de session

Configurez les ressources de session Spark pour des performances optimales :

Exemple :

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

Propriétés de session Spark personnalisées

Tout paramètre avec le préfixe spark. est automatiquement appliqué à la session Spark :

Exemples de configurations Spark :

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

Moteur d’exécution natif (NEE) :

spark.nee.enabled=true

Exemple complet :

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

Paramètres de connexion client HTTP

Configurez les connexions de transport HTTP du pilote pour des performances réseau optimales. Ces paramètres ne configurent ni ne gèrent le pooling de connexions JDBC :

Exemple :

HttpMaxTotalConnections=200;HttpMaxConnectionsPerRoute=100;HttpConnectionTimeoutInSeconds=60

Configuration du serveur proxy

Configurez les paramètres de proxy HTTP et SOCKS pour les environnements d’entreprise :

Exemple de proxy HTTP :

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

Exemple de proxy SOCKS :

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

Configuration de la journalisation

Exemple :

LogLevel=DEBUG

Emplacement du journal par défaut :

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

Configuration du journal personnalisé : Utilisez un fichier log4j2.xml ou logback.xml personnalisé dans votre classpath.

Exemples d’utilisation

Connexion de base

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

Exécution de requêtes

Requête 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);
        }
    }
}

Requête avec filtre

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

Requête avec limite

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

Utilisation des jeux de résultats

Navigation dans les jeux de résultats

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

Traitement des jeux de résultats volumineux

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
}

Utilisation de requêtes préparées

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

Opérations de traitement par lots

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

Regroupement de connexions avec HikariCP

Configurez HikariCP avec l’URL JDBC afin qu’il crée des connexions via LivyDriver. Créez une instance HikariCP et réutilisez-la pendant la durée de vie de l’application.

Dépendance de Maven

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

Configurez HikariCP avec l’URL JDBC

Configurez l’URL JDBC et la classe de pilote pour que HikariCP crée des connexions physiques via LivyDriver. Définir explicitement la classe de pilote est optionnel lorsque la découverte automatique du fournisseur de services 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"));
            }
        }
    }
}

Utilisez les conseils suivants lorsque vous configurez le pool :

Pour la liste complète des options de configuration HikariCP, voir :

• Référence de l’API HikariConfig
• Options de configuration HikariCP

Mappage de type de données

Le pilote mappe les types de données Spark SQL aux types JDBC SQL et java :

Contenu connexe

• Environnements d'exécution Apache Spark dans Fabric
• Fabric Runtime 1.3
• Présentation de l’API Livy pour l’ingénierie des données