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

• APIs JDBC 4.2 : Implémente les APIs JDBC 4.2 pour la connectivité à Spark SQL, avec des limitations propres à Spark et Livy, telles que l’absence de transactions JDBC et des ensembles de résultats à défilement vers l’avant uniquement et en lecture seule
• 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
• Sessions à forte concurrence : partage de sessions Fabric par consentement pour les charges de travail concurrentes
• Gestion explicite du pool de connexions : choisissez HikariCP, un gestionnaire de pool standard, ou LivyBuiltInPooledDataSource ; les points d’entrée directs ne sont jamais mis en pool
• Prise en charge des requêtes natives Spark SQL : exécution directe des instructions SQL Spark 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 :

• Java Development Kit (JDK) : Java 11, 17 et 21 (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 2.0.1 prend en charge Java 11, 17 et 21. Utilisez la dernière version disponible.

• 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-2.0.1.jre11.jar
   • Pour Java 17 : ms-sparksql-jdbc-2.0.1.jre17.jar
   • Pour Java 21 : ms-sparksql-jdbc-2.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 dé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, assurez-vous de remplir les prérequis et d’installer le pilote. En utilisant AuthFlow=2, le pilote utilise DefaultAzureCredential. Pour utiliser Azure CLI comme source de références, installez Azure CLI et exécutez 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();
        }
    }
}

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

Avec le mode de haute concurrence

Le mode haute concurrence (HC) est disponible avec les connexions Microsoft Fabric à partir de la version 1.1.0 du pilote. Ajoutez hcEnabled=true pour obtenir une session HC au lieu d’une session Livy classique :

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

Vous pouvez éventuellement identifier la charge de travail avec une balise session et sélectionner un environnement Fabric :

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

Le mode HC doit être activé manuellement. Si hcEnabled est omis ou défini sur false, le pilote utilise le chemin de session Livy classique.

Consultez la section mode haute concurrence (HC) pour obtenir des instructions complètes de configuration.

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;" +
             "LogLevel=INFO";

Connection conn = DriverManager.getConnection(url);

Paramètres :

• AuthFlow=1: spécifie l’authentification interactive du navigateur

Comportement:

• Ouvre une fenêtre de navigateur pour l’authentification utilisateur
• Met en cache les identifiants des connexions suivantes jusqu’à leur expiration
• 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.pem;" +
             "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 vers un fichier de certificat PEM
• 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 en cache les jetons d’authentification et les rafraîchit en fonction de l’expiration du jeton. Les propriétés analysées AuthEnableCaching et AuthCacheTTLMS ne contrôlent pas ce comportement dans la version 2.0.0.

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 session à forte concurrence

Réglé hcEnabled=true en mode HC. Les propriétés HC restantes ne sont utilisées que lorsque le mode HC est activé.

Exemple :

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

• FabricEnvironmentID Ça s’applique aussi bien aux sessions HC qu’aux sessions classiques
• EnvironmentID est également accepté en tant qu’alias inter-pilotes. Les noms de propriété sont insensibles à la casse.

Voir la section mode High-Concurrency (HC) pour obtenir des instructions de configuration complètes.

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; SessionName=MySparkSession

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

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

Mode haute concurrence (HC)

Le mode haute concurrence (HC) est conçu pour les applications JDBC qui ouvrent plusieurs connexions vers le même espace de travail Fabric et le même lakehouse. Au lieu de provisionner une session Spark classique distincte pour chaque connexion, Fabric sert ces connexions avec une capacité Spark partagée et gérée par le serveur.

Le mode HC peut réduire le temps de démarrage de la connexion, éviter le provisionnement répété des sessions Spark et utiliser la capacité disponible de manière plus efficace pour des charges de travail simultanées. Chaque connexion JDBC continue d’exécuter des instructions dans son contexte assigné au sein de la session HC.

Le mode HC est optionnel et complète le pooling de connexions JDBC. Le mode HC gère les sessions Fabric Spark sur le serveur, tandis que le pooling de connexions JDBC gère les connexions dans l’application client. Si vous n’activez pas le mode HC, le pilote crée une session classique de Livy.

Choisissez entre le mode HC et le mode classique

Utilisez le mode HC lorsque :

• Un service applicatif gère les requêtes concurrentes qui exécutent Spark SQL via plusieurs connexions JDBC.
• Un outil BI ou de reporting ouvre plusieurs connexions au même espace de travail et au même lac.
• Les tâches ETL parallèles ou les tâches planifiées créent des rafales de connexions JDBC.
• Les connexions utilisent des configurations compatibles de Spark et peuvent partager la capacité de Spark gérée par le serveur.
• Réduire le temps de démarrage de la connexion et le provisionnement répété des sessions Spark est important.

Utilisez le mode classique lorsque :

• Votre application n’utilise qu’une ou quelques connexions JDBC durables.
• Chaque connexion nécessite une session Spark dédiée pour une charge de travail stricte ou une isolation des ressources.
• Les connexions nécessitent des configurations Spark ou des environnements Fabric sensiblement différents.
• Le point de terminaison Livy de Fabric pour l’espace de travail cible et le lakehouse ne prend pas en charge le mode HC.

Activer le mode de haute concurrence

Le mode HC est sur inscription. Si vous omettez hcEnabled ou le définissez sur false, le pilote utilise le chemin de session Livy classique. Définissez hcEnabled à true pour obtenir une session HC :

hcEnabled=true;

Paramètres requis en mode HC

L’activation HC nécessite les identifiants valides suivants :

Incluez-le SessionTag car cela influence l’appariement côté serveur pour les sessions chaudes.

Correspondance de session

Le mode HC ne garantit pas que chaque connexion réutilise une session existante. Pour chaque demande d’acquisition de session, le pilote envoie :

• La valeur SessionTag.
• La configuration effective de Spark, y compris les paramètres conf.* et EnvironmentId.

Le service Fabric Livy évalue ces valeurs et décide s’il faut associer la connexion à une session chaude ou en créer une nouvelle. Pour améliorer la réutilisation des sessions, utilisez des valeurs cohérentes entre les connexions destinées à partager la capacité Spark gérée par le serveur.

Désactiver le mode HC

Vous pouvez désactiver les nouvelles acquisitions de session HC sans redéployer le pilote en mettant la chaîne de connexion HC en false ou en supprimant les paramètres HC :

hcEnabled=false;

Retour arrière d’urgence : Pour effectuer un retour arrière d’urgence, démarrez la JVM avec -Dcom.microsoft.fabric.jdbc.hc.disable=true. Ce commutateur à l’échelle de la JVM force le mode classique même lorsqu’une chaîne de connexion contient hcEnabled=true.

Comprendre le comportement du runtime HC

• L’acquisition de session HC est bloquée jusqu’à ce que la session soit prête ou que hcAcquireTimeoutSeconds expire.
• La fermeture d’une connexion alors qu’une acquisition est en cours interrompt l’acquisition et déclenche le nettoyage de la session HC dans la mesure du possible.
• Authentification, exécution de requêtes, annulation, instructions préparées, gestion des résultats, et FabricEnvironmentID fonctionnement via les interfaces JDBC standard dans les deux modes.

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

Connexion en mode de forte concurrence

L’exemple suivant ouvre une connexion JDBC en mode HC, lance une requête, puis libère automatiquement la connexion :

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

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

Parcourir les ensembles de résultats

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

Les ensembles de résultats sont en lecture avant uniquement et en lecture seule. Le pilote ne supporte pas des méthodes telles que first(), last(), et absolute().

Traitement des jeux de résultats volumineux

Le pilote paginage les résultats par taille de réponse. Configurez LivyStatementPageSize dans l’URL JDBC pour définir la taille de page demandée en octets. La Statement.setFetchSize() méthode ne contrôle ni la récupération ni l’utilisation de la mémoire.

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
}

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

Le pilote ne prend pas en charge les transactions JDBC. L’exécution par lots est non transactionnelle et peut partiellement réussir. Un échec ultérieur ne revient pas sur les déclarations antérieures réussies.

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

Choisissez un propriétaire de connexion

Choisissez un composant pour gérer le cycle de vie de la connexion. Ne placez pas un pool de connexion dans un autre.

ConnectionPoolEnabled a pour valeur par défaut false dans le code du pilote comme dans les propriétés fournies. Ce n’est pas une option d’activation pour les points d’entrée existants. Utilisez LivyBuiltInPooledDataSource lorsque le pilote doit posséder le pool, ou configurez un pool externe comme HikariCP. Les autres points d’entrée créent des connexions directes, non propriétaires du pool.

Regroupement de connexions avec HikariCP

HikariCP peut créer des connexions à partir de LivyDataSource ou de l’URL JDBC via LivyDriver. Créez une instance HikariCP, réutilisez-la pendant la durée de vie de l’application, et n’utilisez qu’un seul pool de connexion dans chaque configuration.

Dépendance de Maven

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

Utilisez LivyDataSource.

LivyDataSource crée des connexions physiques indépendantes et ne possède pas de pool de connexions ni de cycle de vie proche séparé. Fournissez-le directement à HikariCP afin que HikariCP soit le seul propriétaire du 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"));
            }
        }
    }
}

L’exemple active le mode HC dans l’URL JDBC. HikariCP gère les connexions côté client, tandis que le mode HC gère les sessions Spark partagées dans Fabric. Retirer hcEnabled et sessionTag si l’application nécessite des sessions classiques.

Utiliser LivyDriver directement

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. Cette configuration n’instancie pas LivyDataSource, donc son pool intégré n’entre pas en jeu :

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

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