Driver JDBC di Microsoft per Microsoft Fabric Data Engineering

JDBC (Java Database Connectivity) è uno standard ampiamente adottato che consente alle applicazioni client di connettersi e lavorare con i dati dai database e dalle piattaforme Big Data.

Il driver Microsoft JDBC per Fabric Data Engineering ti permette di connettere, interrogare e gestire carichi di lavoro Spark in Fabric con l'affidabilità e la semplicità dello standard JDBC. Basato sulle API Livy di Fabric, il driver offre una connettività SQL Spark sicura e flessibile alle tue applicazioni Java e agli strumenti BI. Questa integrazione permette di inviare ed eseguire direttamente codice Spark senza dover creare elementi separati per notebook o definizione di job Spark. Il driver è compatibile con i client JDBC più diffusi, ad esempio DbVisualizer e DBeaver, oltre a strumenti bi che supportano la connettività JDBC, tra cui Tableau.

Funzionalità principali

• API di JDBC 4.2: Implementa le API di JDBC 4.2 per la connettività di Spark SQL, con limitazioni di Spark e Livy tra cui l'assenza di transazioni JDBC e set di risultati di sola lettura e solo in avanti
• Autenticazione con ID Microsoft Entra: più flussi di autenticazione, tra cui l'autenticazione interattiva, le credenziali del client e l'autenticazione basata su certificati
• Sessioni a elevata concorrenza: condivisione facoltativa delle sessioni di Fabric per carichi di lavoro concorrenti
• Pooling esplicito delle connessioni: Scegli HikariCP, un gestore di pool standard, oppure LivyBuiltInPooledDataSource; i punti di accesso diretti non usano mai il pooling
• Supporto nativo per query SQL di Spark: esecuzione diretta delle istruzioni SQL di Spark senza traduzione
• Supporto completo dei tipi di dati: supporto per tutti i tipi di dati SPARK SQL, inclusi tipi complessi (ARRAY, MAP, STRUCT)
• Prelettura del set di risultati asincrono: caricamento dei dati in background per migliorare le prestazioni
• Modello Circuit Breaker: protezione contro i guasti a catena con tentativo automatico
• Riconnessione automatica: ripristino sessione trasparente in caso di errori di connessione
• Logica avanzata di ripetizione dei tentativi: riprovare con backoff esponenziale e ripristino di sessione per migliorare la resilienza
• Supporto proxy: configurazione proxy HTTP e SOCKS per ambienti aziendali

Prerequisiti

Prima di utilizzare il driver Microsoft JDBC per Fabric Data Engineering, assicurati di avere:

• Java Development Kit (JDK): Java 11, 17 e 21 (consigliato Java 21)
• Fabric Access: Accesso a uno spazio di lavoro Fabric
• credenziali Microsoft Entra ID: credenziali appropriate per l'autenticazione
• ID area di lavoro e Lakehouse: identificatori GUID per il tuo workspace e lakehouse Fabric

Download e installazione

Il driver Microsoft JDBC per Fabric Data Engineering versione 2.0.1 supporta Java 11, 17 e 21. Usare la versione più recente disponibile.

• Scarica il driver Microsoft JDBC per Fabric Data Engineering (zip)
• Scarica il driver Microsoft JDBC per Fabric Data Engineering (tar)

1. Scaricare il file ZIP o tar dai collegamenti precedenti.
2. Estrarre il file scaricato per accedere ai file JAR del driver.
3. Selezionare il file JAR corrispondente alla versione JRE:
   • Per Java 11: ms-sparksql-jdbc-2.0.1.jre11.jar
   • Per Java 17: ms-sparksql-jdbc-2.0.1.jre17.jar
   • Per Java 21: ms-sparksql-jdbc-2.0.1.jre21.jar
4. Aggiungere il file JAR selezionato al classpath dell'applicazione.
5. Per dei client JDBC, configurare classe driver JDBC: com.microsoft.spark.livy.jdbc.LivyDriver

Esempio di avvio rapido

Questo esempio dimostra come connettersi a Fabric ed eseguire una query utilizzando il driver Microsoft JDBC per Fabric Data Engineering. Prima di eseguire questo codice, assicurati di completare i prerequisiti e di installare il driver. Utilizzando AuthFlow=2, il driver usa DefaultAzureCredential. Per usare interfaccia della riga di comando di Azure come sorgente delle credenziali, installa interfaccia della riga di comando di Azure ed esegui 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 stringa di connessione

Stringa di connessione di base

Il driver Microsoft JDBC per Fabric Data Engineering utilizza il seguente formato di stringa di connessione:

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

Componenti della stringa di connessione

Stringhe di connessione di esempio

Connessione di base (autenticazione basata su Interactive Browser)

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

Con la configurazione delle risorse 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 le proprietà della sessione 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 modalità ad alta concorrenza

La modalità a concorrenza elevata (HC) è disponibile nelle connessioni Microsoft Fabric nella versione 1.1.0 del driver e successive. Aggiungi hcEnabled=true per acquisire una sessione HC invece di una classica sessione di Livio:

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

Puoi opzionalmente identificare il carico di lavoro con un tag session e selezionare un ambiente Fabric:

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

La modalità HC è facoltativa. Se hcEnabled viene omesso o impostato su false, il driver utilizza il percorso classico della sessione di Livy.

Consulta la sezione modalità High-Concurrency (HC) per istruzioni complete sulla configurazione.

Authentication

Il driver Microsoft JDBC per Fabric Data Engineering supporta molteplici metodi di autenticazione tramite Microsoft Entra ID (precedentemente Azure Active Directory). L'autenticazione viene configurata usando il AuthFlow parametro nella stringa di connessione.

Flussi di autenticazione

Autenticazione interattiva del browser

Ideale per: sviluppo e applicazioni interattive

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

Connection conn = DriverManager.getConnection(url);

Parametri:

• AuthFlow=1: specifica l'autenticazione interattiva del browser

Comportamento:

• Apre una finestra del browser per l'autenticazione utente
• Memorizza le credenziali delle connessioni successive fino alla loro scadenza
• Adatto per applicazioni a utente singolo

Credenziali del client o autenticazione del principale del servizio

Ideale per: servizi automatizzati e processi in background

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

Parametri obbligatori:

• AuthFlow=3: specifica l'autenticazione delle credenziali client
• AuthClientID: ID dell'applicazione (client) di Microsoft Entra ID
• AuthClientSecret: segreto del client da Microsoft Entra ID
• AuthTenantID: ID tenant Microsoft Entra

Procedure consigliate:

• Archiviare i segreti in modo sicuro (Azure Key Vault, variabili di ambiente)
• Usare le identità gestite quando possibile
• Ruotare regolarmente i segreti

Autenticazione basata su certificati

Ideale per: applicazioni aziendali che richiedono l'autenticazione basata su certificati

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

Parametri obbligatori:

• AuthFlow=4: specifica l'autenticazione basata su certificati
• AuthClientID: ID applicazione (cliente)
• AuthCertificatePath: Percorso verso un file di certificato PEM
• AuthCertificatePassword: password del certificato
• AuthTenantID: ID tenant Microsoft Entra

Autenticazione del token di accesso

Ideale per: scenari di autenticazione personalizzati

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

Cache di autenticazione

Il driver memorizza i token di autenticazione e li aggiorna in base alla scadenza del token. Le proprietà analizzate AuthEnableCaching e AuthCacheTTLMS non controllano questo comportamento nella versione 2.0.0.

Parametri di configurazione

Parametri obbligatori

Questi parametri devono essere presenti in ogni stringa di connessione:

Parametri facoltativi

Configurazione della versione dell'API

Configurazione dell'ambiente

Configurazione della sessione a elevata concorrenza

Imposta hcEnabled=true sulla modalità HC. Le restanti proprietà HC vengono utilizzate solo quando la modalità HC è attivata.

Esempio:

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

• FabricEnvironmentID Si applica sia alle sessioni HC che a quelle classiche
• EnvironmentID è anche accettato come alias cross-driver. I nomi delle proprietà non sono sensibili alle maiuscole.

Consulta la sezione modalità High-Concurrency (HC) per istruzioni complete sulla configurazione.

Configurazione di Spark

Configurazione della risorsa sessione

Configurare le risorse della sessione Spark per ottenere prestazioni ottimali:

Esempio:

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

Proprietà personalizzate della sessione Spark

Qualsiasi parametro con il prefisso spark. viene applicato automaticamente alla sessione di Spark:

Configurazioni spark di esempio:

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

Motore di esecuzione nativo (NEE):

spark.nee.enabled=true

Esempio 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

Impostazioni di connessione client HTTP

Configura le connessioni di trasporto HTTP del driver per ottenere prestazioni ottimali della rete. Queste impostazioni non configurano né gestiscono il pool di connessione JDBC:

Esempio:

HttpMaxTotalConnections=200;HttpMaxConnectionsPerRoute=100;HttpConnectionTimeoutInSeconds=60

Configurazione proxy

Configurare le impostazioni proxy HTTP e SOCKS per gli ambienti aziendali:

Esempio di proxy HTTP:

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

Esempio di proxy SOCKS:

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

Configurazione del log

Esempio:

LogLevel=DEBUG

Percorso log predefinito:

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

Configurazione log personalizzata: Usare un file personalizzato log4j2.xml o logback.xml nel classpath.

Modalità ad alta concorrenza (HC)

La modalità ad alta concorrenza (HC) è progettata per applicazioni JDBC che aprono più connessioni allo stesso spazio di lavoro Fabric e allo stesso lakehouse. Invece di fornire una sessione Spark classica separata per ogni connessione, Fabric offre a queste connessioni una capacità Spark condivisa e gestita dal server.

La modalità HC può ridurre i tempi di avvio della connessione, evitare il ripetito provisioning di sessioni Spark e utilizzare la capacità disponibile in modo più efficiente per carichi di lavoro concorrenti. Ogni connessione JDBC continua a eseguire le istruzioni attraverso il contesto assegnato all'interno della sessione HC.

La modalità HC è opt-in e completa il pooling di connessioni JDBC. La modalità HC gestisce le sessioni Fabric Spark sul server, mentre il pool di connessione JDBC gestisce le connessioni nell'applicazione client. Se non attivi la modalità HC, il driver crea una classica sessione di Livy.

Scegli tra modalità HC e classica

Usa la modalità HC quando:

• Un servizio applicativo gestisce richieste concorrenti che eseguono SQL Spark tramite più connessioni JDBC.
• Uno strumento di BI o di creazione di report apre diverse connessioni alla stessa area di lavoro e allo stesso lakehouse.
• Le attività ETL parallele o i processi pianificati generano picchi di connessioni JDBC.
• Le connessioni utilizzano configurazioni Spark compatibili e possono condividere la capacità Spark gestita dal server.
• Ridurre i tempi di avvio della connessione e il provisioning ripetuto delle sessioni Spark è importante.

Usa la modalità classica quando:

• La tua applicazione utilizza solo una o poche connessioni JDBC di lunga durata.
• Ogni connessione richiede una sessione Spark dedicata per un carico di lavoro rigoroso o per l'isolamento delle risorse.
• Le connessioni richiedono configurazioni Spark o ambienti Fabric sostanzialmente diversi.
• L'endpoint Fabric Livy per lo spazio di lavoro target e lakehouse non supporta la modalità HC.

Abilita la modalità ad alta concorrenza

La modalità HC è facoltativa. Se ometti hcEnabled o lo imposti su false, il driver usa il percorso classico della sessione Livy. Imposta hcEnabled su vero per ottenere una sessione HC:

hcEnabled=true;

Parametri richiesti in modalità HC

L'attivazione HC richiede i seguenti identificatori validi:

Includi SessionTag perché influenza il matching lato server per le sessioni calde.

Abbinamento delle sessioni

La modalità HC non garantisce che ogni connessione riutilizzi una sessione esistente. Per ogni richiesta di acquisizione di sessione, il pilota invia:

• Il valore SessionTag.
• La configurazione effettiva di Spark, inclusi conf.* i parametri e EnvironmentId.

Il servizio Fabric Livy valuta questi valori e decide se collegare la connessione a una sessione calda o crearne una nuova. Per migliorare il riutilizzo delle sessioni, utilizza valori coerenti tra le connessioni che sono destinate a condividere la capacità di Spark gestita dal server.

Disabilita la modalità HC

Puoi disabilitare nuove acquisizioni di sessione HC senza ridistribuire il driver impostando la stringa di connessione HC su false oppure rimuovendo i parametri HC:

hcEnabled=false;

Rollback di emergenza: In caso di rollback di emergenza, avviare la JVM con -Dcom.microsoft.fabric.jdbc.hc.disable=true. Questa opzione, valida per l'intera JVM, forza la modalità classica anche quando una stringa di connessione contiene hcEnabled=true.

Comprendere il comportamento del runtime di HC

• L'acquisizione della sessione HC rimane bloccata finché la sessione non è pronta o hcAcquireTimeoutSeconds scade.
• La chiusura di una connessione mentre l'acquisizione è in corso interrompe l'acquisizione e attiva la pulizia della sessione HC per quanto possibile.
• Autenticazione, esecuzione delle query, annullamento, istruzioni preparate, gestione dei risultati e FabricEnvironmentID avvengono tramite le interfacce JDBC standard in entrambe le modalità.

Esempi di utilizzo

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

Connessione in modalità a elevata concorrenza

Il seguente esempio apre una connessione JDBC in modalità HC, esegue una query e rilascia automaticamente la connessione:

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

Esecuzione di query

Query semplice

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

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

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

Utilizzo dei set di risultati

Scorrere i set di risultati

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

I set di risultati sono a scorrimento solo in avanti e in sola lettura. Il driver non supporta metodi come first(), last(), e absolute().

Elaborazione di set di risultati di grandi dimensioni

Il driver pagina i risultati in base alla dimensione della risposta. Configura LivyStatementPageSize nell'URL JDBC per impostare la dimensione richiesta della pagina in byte. Il Statement.setFetchSize() metodo non controlla il recupero o l'uso della 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 di istruzioni preparate

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

Operazioni batch

Il driver non supporta transazioni JDBC. L'esecuzione batch è non transazionale e può parzialmente avere successo. Un fallimento successivo non annulla le dichiarazioni precedenti di successo.

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

Scegli un proprietario di connessione

Scegli un componente per gestire il ciclo di vita della connessione. Non mettere un pool di connessione dentro un altro.

ConnectionPoolEnabled è impostato per impostazione predefinita su false sia nel codice del driver sia nelle proprietà fornite. Non è un interruttore op-in per i punti di ingresso attuali. Utilizza LivyBuiltInPooledDataSource quando il driver deve gestire il pool, oppure configura un pool esterno come HikariCP. Gli altri punti di ingresso creano connessioni dirette, non proprietarie del pool.

Pool di connessioni con HikariCP

HikariCP può creare connessioni sia da LivyDataSource sia dall'URL JDBC tramite LivyDriver. Crea un'istanza HikariCP, riutilizzala per tutta la vita dell'applicazione e usa solo un pool di connessione in entrambe le configurazioni.

Dipendenza di Maven

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

Utilizzare LivyDataSource.

LivyDataSource crea connessioni fisiche indipendenti e non possiede un pool di connessione né un ciclo di vita chiuso separato. Forniscilo direttamente a HikariCP così che HikariCP sia l'unico proprietario della piscina:

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'esempio abilita la modalità HC nell'URL JDBC. HikariCP gestisce le connessioni lato client, mentre la modalità HC gestisce le sessioni Spark condivise in Fabric. Rimuovi hcEnabled e sessionTag se l'applicazione richiede sessioni classiche.

Usare LivyDriver direttamente

Configura l'URL JDBC e la classe driver in modo che HikariCP crei connessioni fisiche tramite LivyDriver. Impostare esplicitamente la classe del driver non è obbligatorio quando è disponibile il rilevamento automatico del provider di servizi JDBC. Questa configurazione non istanzia LivyDataSource, quindi il relativo pool integrato non viene utilizzato:

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

Usa le seguenti indicazioni quando configuri il pool:

Per l'elenco completo delle opzioni di configurazione HikariCP, vedi:

• Riferimento API HikariConfig
• Opzioni di configurazione HikariCP

Mapping dei tipi di dati

Il driver esegue il mapping dei tipi di dati Spark SQL ai tipi SQL JDBC e ai tipi Java:

Contenuti correlati

• Runtime di Apache Spark in Fabric
• Runtime Fabric 1.3
• Informazioni sull'API Livy per ingegneria dei dati