Microsoft JDBC-Treiber für Microsoft Fabric Data Engineering

JDBC (Java Database Connectivity) ist ein weit verbreiteter Standard, der es Clientanwendungen ermöglicht, eine Verbindung zu Datenbanken und Big-Data-Plattformen herzustellen und mit diesen zu arbeiten.

Der Microsoft JDBC Driver for Fabric Data Engineering ermöglicht es Ihnen, Spark-Workloads in Fabric mit der Zuverlässigkeit und Einfachheit des JDBC-Standards zu verbinden, abzufragen und zu verwalten. Basierend auf den Livy-APIs von Fabric bietet der Treiber eine sichere und flexible Spark-SQL-Konnektivität zu Ihren Java-Anwendungen und BI-Tools. Diese Integration ermöglicht es Ihnen, Spark-Code direkt einzureichen und auszuführen, ohne separate Notebook- oder Spark-Job-Definitionselemente erstellen zu müssen. Der Treiber ist mit beliebten JDBC-Clients wie DbVisualizer und DBeaver kompatibel sowie mit BI-Tools, die JDBC-Konnektivität unterstützen, einschließlich Tableau.

Wichtige Merkmale

• JDBC 4.2 APIs: Implementiert JDBC 4.2 APIs für die Spark SQL-Konnektivität, mit Spark- und Livy-Einschränkungen wie keine JDBC-Transaktionen und nur vorwärtsgerichteten, lesbaren Ergebnissets
• Microsoft Entra ID-Authentifizierung: Mehrere Authentifizierungsflüsse, einschließlich interaktiver, Clientanmeldeinformationen und zertifikatbasierter Authentifizierung
• Sitzungen mit hoher Parallelität: optionale Freigabe von Fabric-Sitzungen für gleichzeitige Workloads
• Explizite Verbindungspooling: Wähle HikariCP, einen Standard-Poolmanager, oder LivyBuiltInPooledDataSource; Direktzugangspunkte poolen niemals
• Spark SQL native Abfrageunterstützung: Direkte Ausführung von Spark SQL-Anweisungen ohne Übersetzung
• Umfassende Unterstützung für Datentypen: Unterstützung für alle Spark SQL-Datentypen einschließlich komplexer Typen (ARRAY, MAP, STRUCT)
• Asynchrones Ergebnismengen-Prefetching: Hintergrunddatenladen für verbesserte Leistung
• Circuit Breaker Pattern: Schutz vor kaskadierenden Ausfällen mit automatischem Retry
• Automatische Wiederverbindung: Transparente Sitzungswiederherstellung bei Verbindungsfehlern
• Erweiterte Wiederholungslogik: Wiederholen mit exponentieller Backoff- und Sitzungswiederherstellung für verbesserte Resilienz
• Proxyunterstützung: HTTP- und SOCKS-Proxykonfiguration für Unternehmensumgebungen

Voraussetzungen

Bevor Sie den Microsoft JDBC Driver für Fabric Data Engineering verwenden, stellen Sie sicher, dass Sie Folgendes haben:

• Java Development Kit (JDK): Java 11, 17 und 21 (Java 21 empfohlen)
• Fabric Access: Zugang zu einem Fabric-Arbeitsbereich
• Microsoft Entra ID-Anmeldeinformationen: Geeignete Anmeldeinformationen für die Authentifizierung
• Arbeitsbereichs- und Lakehouse-IDs: GUID-IDs für den Fabric-Arbeitsbereich und das Lakehouse

Herunterladen und Installieren

Der Microsoft JDBC Driver for Fabric Data Engineering Version 2.0.1 unterstützt Java 11, 17 und 21. Verwenden Sie die neueste verfügbare Version.

• Laden Sie den Microsoft JDBC-Treiber für Fabric Data Engineering herunter (zip)
• Laden Sie den Microsoft JDBC-Treiber für Fabric Data Engineering (tar) herunter

1. Laden Sie entweder die ZIP- oder Tar-Datei aus den obigen Links herunter.
2. Extrahieren Sie die heruntergeladene Datei, um auf die Treiber-JAR-Dateien zuzugreifen.
3. Wählen Sie die JAR-Datei aus, die Ihrer JRE-Version entspricht:
   • Für Java 11: ms-sparksql-jdbc-2.0.1.jre11.jar
   • Für Java 17: ms-sparksql-jdbc-2.0.1.jre17.jar
   • Für Java 21: ms-sparksql-jdbc-2.0.1.jre21.jar
4. Fügen Sie die ausgewählte JAR-Datei zum Klassenpfad Ihrer Anwendung hinzu.
5. Konfigurieren Sie für JDBC-Clients die JDBC-Treiberklasse: com.microsoft.spark.livy.jdbc.LivyDriver

Schnellstartbeispiel

Dieses Beispiel zeigt, wie man sich mit Fabric verbindet und eine Abfrage mit dem Microsoft JDBC Driver for Fabric Data Engineering ausführt. Bevor Sie diesen Code ausführen, stellen Sie sicher, dass Sie die Voraussetzungen erfüllen und den Treiber installieren. Durch die Verwendung von AuthFlow=2 verwendet der Treiber DefaultAzureCredential. Um die Azure CLI als Quelle für die Anmeldeinformationen zu verwenden, installieren Sie die Azure CLI und führen Sie az login aus.

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

Verbindungszeichenfolgenformat

Basisverbindungszeichenfolge

Der Microsoft JDBC Driver for Fabric Data Engineering verwendet folgendes Verbindungszeichenfolge-Format:

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

Verbindungskettenkomponenten

Beispielverbindungszeichenfolgen

Standardverbindung (interaktive browserbasierte Authentifizierung)

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

Mit Spark-Ressourcenkonfiguration

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

Mit Spark Session-Eigenschaften

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

Mit Modus für hohe Nebenläufigkeit

Der High-Concurrency (HC)-Modus ist für Microsoft Fabric-Verbindungen ab Treiberversion 1.1.0 und später verfügbar. Füge hinzu hcEnabled=true , um eine HC-Session statt einer klassischen Livy-Session zu erhalten:

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

Sie können die Arbeitslast optional mit einem Session-Tag identifizieren und eine Fabric-Umgebung auswählen:

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

Der HC-Modus ist opt-in. Wenn hcEnabled ausgelassen oder auf falsegesetzt ist, verwendet der Treiber den klassischen Livy-Session-Pfad.

Siehe den Abschnitt zum Modus „High-Concurrency (HC)“ für vollständige Konfigurationshinweise.

Authentifizierung

Der Microsoft JDBC Driver for Fabric Data Engineering unterstützt mehrere Authentifizierungsmethoden über Microsoft Entra ID (früher Azure Active Directory). Die Authentifizierung wird mithilfe des AuthFlow Parameters in der Verbindungszeichenfolge konfiguriert.

Authentifizierungsabläufe

Interaktive Browserauthentifizierung

Am besten geeignet für: Entwicklung und interaktive Anwendungen

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

Connection conn = DriverManager.getConnection(url);

Parameter:

• AuthFlow=1: Gibt die interaktive Browserauthentifizierung an.

Benehmen:

• Öffnet ein Browserfenster für die Benutzerauthentifizierung.
• Speichert Zugangsdaten für nachfolgende Verbindungen, bis sie ablaufen
• Geeignet für Einzelbenutzeranwendungen

Client-Anmeldeinformationen oder Serviceprinzipal-Authentifizierung

Am besten geeignet für: Automatisierte Dienste und Hintergrundaufträge

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

Erforderliche Parameter:

• AuthFlow=3: Gibt die Authentifizierung der Clientanmeldeinformationen an.
• AuthClientID: Anwendungs-ID (Client-ID) von Microsoft Entra ID
• AuthClientSecret: Geheimer Clientschlüssel von Microsoft Entra ID
• AuthTenantID: Microsoft Entra Mandanten-ID

Bewährte Methoden:

• Sicheres Speichern geheimer Schlüssel (Azure Key Vault, Umgebungsvariablen)
• Verwenden von verwalteten Identitäten nach Möglichkeit
• Regelmäßiges Erneuern von Geheimnissen

zertifikatbasierte Authentifizierung

Am besten geeignet für: Unternehmensanwendungen, die zertifikatbasierte Authentifizierung erfordern

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

Erforderliche Parameter:

• AuthFlow=4: Gibt die zertifikatbasierte Authentifizierung an.
• AuthClientID: Anwendungs-ID (Client-ID)
• AuthCertificatePath: Pfad zu einer PEM-Zertifikatsdatei
• AuthCertificatePassword: Zertifikatkennwort
• AuthTenantID: Microsoft Entra Mandanten-ID

Zugriffstokenauthentifizierung

Am besten geeignet für: Benutzerdefinierte Authentifizierungsszenarien

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

Authentifizierungs-Caching

Der Treiber speichert Authentifizierungstoken und aktualisiert sie basierend auf Ablauf des Tokens. Die geparsten Eigenschaften AuthEnableCaching und AuthCacheTTLMS steuern dieses Verhalten in Version 2.0.0 nicht.

Konfigurationsparameter

Erforderliche Parameter

Diese Parameter müssen in jeder Verbindungszeichenfolge vorhanden sein:

Optionale Parameter

API-Versionskonfiguration

Umgebungskonfiguration

Konfiguration für eine hohe Anzahl gleichzeitiger Sitzungen

Stellen Sie hcEnabled=true auf den HC-Modus ein. Die übrigen HC-Eigenschaften werden nur verwendet, wenn der HC-Modus aktiviert ist.

Example:

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

• FabricEnvironmentID gilt sowohl für HC- als auch für klassische Sitzungen
• EnvironmentID wird auch als Cross-Driver-Alias akzeptiert. Eigenschaftsnamen sind nicht empfindlich auf Groß-/Kleinschreibung.

Siehe den Abschnitt zum Modus „High-Concurrency (HC)“ für vollständige Konfigurationshinweise.

Spark-Konfiguration

Sitzungsressourcenkonfiguration

Konfigurieren sie Spark-Sitzungsressourcen für eine optimale Leistung:

Example:

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

Benutzerdefinierte Spark-Sitzungseigenschaften

Jeder Parameter mit dem Präfix spark. wird automatisch auf die Spark-Sitzung angewendet:

Beispiel-Spark-Konfigurationen:

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

Natives Ausführungsmodul (NEE):

spark.nee.enabled=true

Vollständiges Beispiel:

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

HTTP-Client-Verbindungseinstellungen

Konfigurieren Sie die HTTP-Transportverbindungen des Treibers für optimale Netzwerkleistung. Diese Einstellungen konfigurieren oder verwalten das JDBC-Verbindungspooling nicht:

Example:

HttpMaxTotalConnections=200;HttpMaxConnectionsPerRoute=100;HttpConnectionTimeoutInSeconds=60

Proxykonfiguration

Konfigurieren von HTTP- und SOCKEN-Proxyeinstellungen für Unternehmensumgebungen:

HTTP-Proxybeispiel:

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

SOCKEN-Proxybeispiel:

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

Protokollierungskonfiguration

Example:

LogLevel=DEBUG

Standardprotokollspeicherort:

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

Benutzerdefinierte Protokollkonfiguration: Verwenden Sie eine benutzerdefinierte log4j2.xml- oder logback.xml-Datei in Ihrem Klassenpfad.

Hoch-Nebenläufigkeit (HC) Modus

High-Concurrency (HC)-Modus ist für JDBC-Anwendungen konzipiert, die mehrere Verbindungen zum selben Fabric-Arbeitsbereich und Lakehouse öffnen. Anstatt für jede Verbindung eine separate klassische Spark-Session bereitzustellen, bedient Fabric diese Verbindungen mit gemeinsamer, serververwalteter Spark-Kapazität.

Der HC-Modus kann die Startzeit der Verbindung verkürzen, wiederholte Spark-Sitzungsbereitstellung vermeiden und die verfügbare Kapazität für gleichzeitige Arbeitslasten effizienter nutzen. Jede JDBC-Verbindung führt SQL-Anweisungen über ihren zugewiesenen Kontext innerhalb der HC-Sitzung weiter aus.

Der HC-Modus ist opt-in und ergänzt das JDBC-Verbindungspooling. Der HC-Modus verwaltet Fabric Spark-Sitzungen auf dem Server, während JDBC Connection Pooling die Verbindungen in der Client-Anwendung verwaltet. Wenn du den HC-Modus nicht aktivierst, erstellt der Fahrer eine klassische Livy-Sitzung.

Wähle zwischen HC und Classic-Modus

Benutze den HC-Modus wenn:

• Ein Anwendungsdienst verarbeitet gleichzeitige Anfragen, die Spark SQL über mehrere JDBC-Verbindungen ausführen.
• Ein BI- oder Reporting-Tool öffnet mehrere Verbindungen zum selben Arbeitsbereich und zum selben Lakehouse.
• Parallele ETL-Aufgaben oder geplante Jobs erzeugen Bursts von JDBC-Verbindungen.
• Verbindungen verwenden kompatible Spark-Konfigurationen und können sich die serververwaltete Spark-Kapazität teilen.
• Die Verkürzung der Verbindungsstartzeit und der wiederholten Spark-Session-Bereitstellung ist wichtig.

Nutze den klassischen Modus, wenn:

• Deine Anwendung nutzt nur eine oder wenige langjährige JDBC-Verbindungen.
• Jede Verbindung benötigt eine dedizierte Spark-Session für strikte Workload- oder Ressourcenisolation.
• Verbindungen erfordern wesentlich unterschiedliche Spark-Konfigurationen oder Fabric-Umgebungen.
• Der Fabric Livy-Endpunkt für den Zielarbeitsbereich und das Lakehouse unterstützt den HC-Modus nicht.

Aktivieren Sie den Modus für hohe Nebenläufigkeit

Der HC-Modus ist optional. Wenn Sie hcEnabled weglassen oder auf „false“ setzen, verwendet der Treiber den klassischen Livy-Sitzungspfad. Setzen Sie hcEnabled auf true, um eine HC-Session zu erhalten:

hcEnabled=true;

Erforderliche Parameter im HC-Modus

Die HC-Aktivierung erfordert folgende gültige Kennungen:

Beziehen Sie SessionTag ein, weil es die serverseitige Zuordnung für Warm-Sessions beeinflusst.

Sitzungsabgleich

Der HC-Modus garantiert nicht, dass jede Verbindung eine bestehende Sitzung wiederverwendet. Für jede Anfrage zur Sitzungserfassung sendet der Treiber:

• Der SessionTag-Wert.
• Die effektive Spark-Konfiguration, einschließlich conf.* Einstellungen und EnvironmentId.

Der Fabric Livy Service bewertet diese Werte und entscheidet, ob die Verbindung mit einer warmen Sitzung verbunden oder eine neue Sitzung erstellt wird. Um die Wiederverwendung der Sitzungen zu verbessern, verwenden Sie konsistente Werte über Verbindungen hinweg, die für die gemeinsame Server-verwaltete Spark-Kapazität gedacht sind.

HC-Modus deaktivieren

Sie können neue HC-Sessions-Erwerbungen deaktivieren, ohne den Treiber neu zu deployen, indem Sie den HC-Verbindungszeichenfolge auf false setzen oder die HC-Parameter entfernen:

hcEnabled=false;

Notfall-Rollbacks: Für den Notfall-Rollback starten Sie die JVM mit -Dcom.microsoft.fabric.jdbc.hc.disable=true. Dieser JVM-weite Schalter erzwingt den klassischen Modus, selbst wenn eine Verbindungszeichenfolge hcEnabled=true enthält.

Verstehen Sie das Laufzeitverhalten von HC

• Die Anforderung einer HC-Session blockiert, bis die Session bereit ist oder hcAcquireTimeoutSeconds abläuft.
• Das Schließen einer Verbindung während einer laufenden Datenerfassung unterbricht die Datenerfassung und löst eine HC-Sitzungsbereinigung nach bestem Bemühen aus.
• Authentifizierung, Abfrageausführung, Abbruch, Prepared Statements, Verarbeitung von Ergebnissen und FabricEnvironmentID werden in beiden Modi über die Standard-JDBC-Schnittstellen abgewickelt.

Verwendungsbeispiele

Grundlegende Verbindung

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

Verbindung im Hochparallelitätsmodus

Das folgende Beispiel öffnet eine JDBC-Verbindung im HC-Modus, führt eine Abfrage aus und löst die Verbindung automatisch:

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

Ausführen von Abfragen

Einfache Abfrage

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

Abfrage mit Filter

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

Abfrage mit Limit

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

Arbeiten mit Resultsets

Traverse-Ergebnismengen

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

Die Ergebnissätze sind nur vorwärts durchlaufbar und schreibgeschützt. Der Treiber unterstützt keine Methoden wie first(), last(), und absolute().

Verarbeiten großer Ergebnismengen

Der Treiber paginiert die Ergebnisse nach Antwortgröße. Stellen Sie LivyStatementPageSize in der JDBC-URL ein, um die angeforderte Seitengröße in Bytes festzulegen. Die Methode Statement.setFetchSize() steuert weder den Abruf noch den Speicherverbrauch.

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
}

Verwendung vorbereiteter Anweisungen

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

Batchvorgänge

Der Treiber unterstützt keine JDBC-Transaktionen. Die Batch-Ausführung ist nicht-transaktional und kann teilweise erfolgreich sein. Ein späterer Fehlschlag macht frühere erfolgreiche Anweisungen nicht rückgängig.

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

Wählen Sie einen Verbindungsbesitzer

Wählen Sie eine Komponente, die den Verbindungslebenszyklus besitzt. Platzieren Sie keinen Verbindungspool innerhalb eines anderen.

ConnectionPoolEnabled ist sowohl im Treibercode als auch in den mitgelieferten Eigenschaften standardmäßig auf false gesetzt. Es ist kein opt-in-Schalter für die aktuellen Einstiegspunkte. Verwenden Sie LivyBuiltInPooledDataSource, wenn der Treiber den Pool verwalten soll, oder konfigurieren Sie einen externen Pool wie HikariCP. Die anderen Einstiegspunkte erstellen direkte Verbindungen, die keinen Pool besitzen.

Verbindungspooling mit HikariCP

HikariCP kann Verbindungen entweder aus LivyDataSource oder aus der JDBC-URL über LivyDriver erstellen. Erstellen Sie eine HikariCP-Instanz, verwenden Sie sie für die Anwendungslebensdauer und verwenden Sie in jeder Konfiguration nur einen Verbindungspool.

Maven-Abhängigkeit

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

LivyDataSource verwenden

LivyDataSource Schafft unabhängige physische Verbindungen und besitzt keinen Verbindungspool oder einen separaten, engen Lebenszyklus. Geben Sie es direkt an HikariCP weiter, damit HikariCP der einzige Poolbesitzer ist:

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

Das Beispiel aktiviert den HC-Modus in der JDBC-URL. HikariCP verwaltet clientseitige Verbindungen, während der HC-Modus gemeinsame Spark-Sitzungen in Fabric verwaltet. Entferne hcEnabled und sessionTag wenn die Anwendung klassische Sitzungen erfordert.

Verwenden Sie LivyDriver direkt

Konfigurieren Sie die JDBC-URL und die Treiberklasse so, dass HikariCP physische Verbindungen über LivyDrivererstellt. Das explizite Festlegen der Treiberklasse ist optional, wenn die automatische Erkennung des JDBC-Dienstanbieters verfügbar ist. Diese Konfiguration instanziiert LivyDataSource nicht, daher ist ihr integrierter Pool nicht relevant:

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

Verwenden Sie folgende Anleitung bei der Konfiguration des Pools:

Die vollständige Liste der HikariCP-Konfigurationsoptionen finden Sie hier:

• HikariConfig API-Referenz
• HikariCP-Konfigurationsoptionen

Datentypzuordnung

Der Treiber ordnet Spark SQL-Datentypen JDBC-SQL-Typen und Java-Typen zu.

Verwandte Inhalte

• Apache Spark-Runtimes in Fabric
• Fabric Runtime 1.3
• Was ist die Livy-API für Data Engineering