sterownik Microsoft JDBC na potrzeby inżynierii danych Microsoft Fabric

JDBC (łączność z bazą danych Java) to powszechnie przyjęty standard, który umożliwia aplikacjom klienckim łączenie się z danymi z baz danych i platform danych big data oraz pracę z nimi.

Microsoft JDBC Driver for Fabric Data Engineering pozwala łączyć, zapytywać i zarządzać obciążeniami Spark w Fabric z niezawodnością i prostotą standardu JDBC. Oparty na API Livy firmy Fabric, sterownik zapewnia bezpieczną i elastyczną łączność Spark SQL z aplikacjami Java i narzędziami BI. Ta integracja umożliwia bezpośrednie przesyłanie i wykonywanie kodu Spark bez konieczności tworzenia oddzielnych elementów notesu lub elementów definicji zadania Spark. Sterownik jest zgodny z popularnymi klientami JDBC, takimi jak DbVisualizer i DBeaver, a także narzędziami analizy biznesowej obsługującymi łączność JDBC, w tym Tableau.

Najważniejsze funkcje    

• API JDBC 4.2: Implementuje API JDBC 4.2 dla łączności Spark SQL, z ograniczeniami Spark i Livy, takimi jak brak transakcji JDBC oraz zestawy wyników tylko do odczytu i przekierowania
• Uwierzytelnianie Microsoft Entra ID: wiele przepływów uwierzytelniania, w tym interakcyjne, poświadczenia klienta i uwierzytelnianie oparte na certyfikatach
• Sesje o wysokiej współbieżności: Opcjonalne udostępnianie sesji Fabric dla współbieżnych obciążeń
• Jawne zarządzanie pulą połączeń: Wybierz HikariCP, standardowy menedżer puli połączeń, lub LivyBuiltInPooledDataSource; bezpośrednie punkty wejścia nigdy nie korzystają z puli połączeń
• Natywne wsparcie zapytań SQL w Spark: Bezpośrednie wykonywanie instrukcji SQL Spark bez tłumaczenia
• Kompleksowa obsługa typów danych: obsługa wszystkich typów danych Spark SQL, w tym typów złożonych (ARRAY, MAP, STRUCT)
• Asynchroniczne pobieranie zestawu wyników: ładowanie danych w tle w celu zwiększenia wydajności
• Wzorzec obwodu przerywającego: Ochrona przed awariami kaskadowymi z automatycznym ponawianiem
• Automatyczne ponowne nawiązywanie połączenia: Przezroczyste odzyskiwanie sesji w przypadku niepowodzeń połączenia
• Zaawansowana logika ponawiania prób: ponawianie prób z wykładniczym opóźnieniem i odzyskiwanie sesji w celu zwiększenia odporności.
• Obsługa serwera proxy: konfiguracja serwera proxy HTTP i SOCKS dla środowisk przedsiębiorstwa

Wymagania wstępne

Przed użyciem sterownika Microsoft JDBC dla Fabric Data Engineering, upewnij się, że masz:

• Java Development Kit (JDK): Java 11, 17 i 21 (zalecana Java 21)
• Fabric Access: Dostęp do przestrzeni roboczej Fabric
• Microsoft Entra ID poświadczenia: Odpowiednie poświadczenia do uwierzytelniania
• Identyfikatory obszaru roboczego i lakehouse: identyfikatory GUID dla obszaru roboczego Fabric i lakehouse

Pobieranie i instalacja

Microsoft JDBC Driver for Fabric Data Engineering w wersji 2.0.1 obsługuje Java 11, 17 i 21. Użyj najnowszej dostępnej wersji.

• Pobierz sterownik Microsoft JDBC dla Fabric Data Engineering (zip)
• Pobierz sterownik Microsoft JDBC dla Fabric Data Engineering (tar)

1. Pobierz plik zip lub tar z powyższych linków.
2. Wyodrębnij pobrany plik, aby uzyskać dostęp do plików JAR sterownika.
3. Wybierz plik JAR zgodny z wersją środowiska JRE:
   • W przypadku języka Java 11: ms-sparksql-jdbc-2.0.1.jre11.jar
   • Dla środowiska Java 17: ms-sparksql-jdbc-2.0.1.jre17.jar
   • Dla środowiska Java 21: ms-sparksql-jdbc-2.0.1.jre21.jar
4. Dodaj wybrany plik JAR do ścieżki klasy aplikacji.
5. W przypadku klientów JDBC skonfiguruj klasę sterowników JDBC: com.microsoft.spark.livy.jdbc.LivyDriver

Przykład szybkiego startu

Ten przykład pokazuje, jak połączyć się z Fabric i wykonać zapytanie za pomocą sterownika Microsoft JDBC Driver for Fabric Data Engineering. Przed uruchomieniem tego kodu upewnij się, że spełniłeś wymagania wstępne i zainstalowałeś sterownik. Używając AuthFlow=2, sterownik używa DefaultAzureCredential. Aby użyć Azure CLI jako źródła poświadczenia, zainstaluj Azure CLI i uruchom 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 ciągu połączenia

Podstawowy ciąg połączenia

Microsoft JDBC Driver for Fabric Data Engineering używa następującego formatu parametrów połączenia:

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

Składniki ciągu połączenia

Przykładowe ciągi połączenia

Połączenie podstawowe (uwierzytelnianie oparte na przeglądarce interakcyjnej)

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

Za pomocą konfiguracji zasobów platformy 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

Za pomocą właściwości sesji 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

W trybie wysokiej współbieżności

Tryb wysokiej współbieżności (HC) jest dostępny dla połączeń Microsoft Fabric w wersji sterownika 1.1.0 i nowszych. Dodaj hcEnabled=true, aby uzyskać sesję HC zamiast klasycznej sesji Livy:

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

Możesz opcjonalnie zidentyfikować obciążenie za pomocą tagu sesji i wybrać środowisko Fabric:

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

Tryb HC wymaga włączenia. Jeśli hcEnabled zostanie pominięty lub ustawiony na false, sterownik korzysta z klasycznej ścieżki sesji Livy.

Zobacz sekcję High-Concurrency (HC) Mode, aby uzyskać pełne wskazówki dotyczące konfiguracji.

Authentication

Sterownik Microsoft JDBC dla Fabric Data Engineering obsługuje wiele metod uwierzytelniania poprzez Microsoft Entra ID (dawniej Azure Active Directory). Uwierzytelnianie jest konfigurowane przy użyciu parametru AuthFlow w parametrach połączenia.

Przepływy uwierzytelniania

Uwierzytelnianie za pomocą przeglądarki interakcyjnej

Najlepsze do: Rozwoju i interaktywnych aplikacji

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

Connection conn = DriverManager.getConnection(url);

Parametry:

• AuthFlow=1: Określa uwierzytelnianie za pomocą przeglądarki interakcyjnej

Zachowanie:

• Otwiera okno przeglądarki na potrzeby uwierzytelniania użytkownika
• Zapisuje dane uwierzytelniające do kolejnych połączeń aż do ich wygaśnięcia
• Odpowiednie dla aplikacji z jednym użytkownikiem

Poświadczenia klienta lub uwierzytelnianie jednostki usługi

Najlepsze dla: Zautomatyzowane usługi i zadania w tle

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

Wymagane parametry:

• AuthFlow=3: Określa uwierzytelnianie poświadczeń klienta
• AuthClientID: Identyfikator aplikacji (klienta) z Microsoft Entra ID
• AuthClientSecret: Tajny klucz klienta od Microsoft Entra ID
• AuthTenantID: identyfikator dzierżawy Microsoft Entra

Najlepsze rozwiązania:

• Bezpieczne przechowywanie wpisów tajnych (azure Key Vault, zmienne środowiskowe)
• Używanie tożsamości zarządzanych, jeśli jest to możliwe
• Regularne zmienianie sekretów

Uwierzytelnianie certyfikatem

Najlepsze rozwiązanie: Aplikacje dla przedsiębiorstw wymagające uwierzytelniania opartego na certyfikatach

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

Wymagane parametry:

• AuthFlow=4: Określa uwierzytelnianie oparte na certyfikatach
• AuthClientID: Identyfikator aplikacji (klienta)
• AuthCertificatePath: Ścieżka do pliku certyfikatu PEM
• AuthCertificatePassword: Hasło certyfikatu
• AuthTenantID: identyfikator dzierżawy Microsoft Entra

Uwierzytelnianie tokenu dostępu

Najlepszy wybór dla: niestandardowe scenariusze uwierzytelniania

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

Buforowanie uwierzytelniania

Sterownik buforuje tokeny uwierzytelniające i odświeża je w zależności od ich wygaśnięcia. Przeanalizowane właściwości AuthEnableCaching i AuthCacheTTLMS nie wpływają na to zachowanie w wersji 2.0.0.

Parametry konfiguracji

Wymagane parametry

Te parametry muszą być obecne w każdym ciągu połączenia:

Parametry opcjonalne

Konfiguracja wersji interfejsu API

Konfiguracja środowiska

Konfiguracja sesji o wysokiej współbieżności

Ustaw hcEnabled=true tryb HC. Pozostałe właściwości HC są używane tylko wtedy, gdy tryb HC jest włączony.

Example:

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

• FabricEnvironmentID dotyczy zarówno sesji HC, jak i sesji klasycznych
• EnvironmentID jest również akceptowany jako pseudonim cross-driver. Nazwy właściwości są niewrażliwe na wielkość liter.

Zobacz sekcję High-Concurrency (HC) Mode, aby uzyskać pełne wskazówki dotyczące konfiguracji.

Konfiguracja platformy Spark

Konfiguracja zasobu sesji

Skonfiguruj zasoby sesji platformy Spark w celu uzyskania optymalnej wydajności:

Example:

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

Niestandardowe właściwości sesji platformy Spark

Dowolny parametr z prefiksem spark. jest automatycznie stosowany do sesji platformy Spark:

Przykładowe konfiguracje platformy 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

Natywny silnik wykonywania (NEE):

spark.nee.enabled=true

Kompletny przykład:

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

Ustawienia połączenia klienta HTTP

Konfiguruj połączenia transportowe HTTP sterownika dla optymalnej wydajności sieci. Te ustawienia nie konfigurują ani nie zarządzają pulą połączeń JDBC:

Example:

HttpMaxTotalConnections=200;HttpMaxConnectionsPerRoute=100;HttpConnectionTimeoutInSeconds=60

Konfiguracja serwera proxy

Skonfiguruj ustawienia serwera proxy HTTP i SOCKS dla środowisk przedsiębiorstwa:

Przykład serwera proxy HTTP:

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

Przykład serwera proxy SOCKS:

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

Konfiguracja rejestrowania

Example:

LogLevel=DEBUG

Domyślna lokalizacja dziennika:

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

Niestandardowa konfiguracja dziennika: Użyj niestandardowego pliku lub log4j2.xmllogback.xml na ścieżce klas.

Tryb wysokiej współbieżności (HC)

Tryb wysokiej współbieżności (HC) jest zaprojektowany dla aplikacji JDBC, które otwierają wiele połączeń do tej samej przestrzeni roboczej Fabric i Lakehouse. Zamiast przydzielać osobną klasyczną sesję Spark dla każdego połączenia, Fabric obsługuje te połączenia z wspólną, serwerowo zarządzaną pojemnością Sparka.

Tryb HC może skrócić czas uruchomienia połączenia, uniknąć powtarzającego się udostępniania sesji Spark oraz efektywniej wykorzystać dostępną pojemność dla równoczesnych obciążeń. Każde połączenie JDBC nadal wykonuje polecenia w ramach przypisanego kontekstu w sesji HC.

Tryb HC jest opcjonalny i uzupełnia pulowanie połączeń JDBC. Tryb HC zarządza sesjami Fabric Spark na serwerze, natomiast pula połączeń JDBC zarządza połączeniami w aplikacji klienckiej. Jeśli nie włączysz trybu HC, sterownik tworzy klasyczną sesję Livy.

Wybierz między trybem HC a klasycznym

Używaj trybu HC, gdy:

• Usługa aplikacyjna obsługuje jednoczesne żądania uruchamiające Spark SQL przez wiele połączeń JDBC.
• BI lub narzędzie raportujące otwiera kilka połączeń z tym samym miejscem pracy i domem nad jeziorem.
• Równoległe zadania ETL lub zaplanowane zadania generują fale połączeń JDBC.
• Połączenia korzystają z kompatybilnych konfiguracji Spark i mogą współdzielić przepustowość Spark zarządzaną przez serwer.
• Skrócenie czasu uruchamiania połączenia i powtarzającego się udostępniania sesji w Spark jest ważne.

Użyj trybu klasycznego, gdy:

• Twoja aplikacja korzysta tylko z jednego lub kilku długo utrzymywanych połączeń JDBC.
• Każde połączenie wymaga dedykowanej sesji Spark do ścisłej izolacji obciążenia lub zasobów.
• Połączenia wymagają znacznie innych konfiguracji Spark lub środowisk Fabric.
• Punkt końcowy usługi Fabric Livy dla docelowego obszaru roboczego i lakehouse nie obsługuje trybu HC.

Włącz tryb wysokiej współbieżności

Tryb HC wymaga włączenia. Jeśli pominiesz hcEnabled lub ustawisz false, sterownik używa klasycznej ścieżki sesji Livy. Ustaw hcEnabled na true, aby uzyskać sesję HC:

hcEnabled=true;

Wymagane parametry w trybie HC

Aktywacja HC wymaga następujących ważnych identyfikatorów:

Uwzględnij, SessionTag ponieważ wpływa to na dopasowanie po stronie serwera dla sesji ciepłych.

Dopasowywanie sesji

Tryb HC nie gwarantuje, że każde połączenie ponownie wykorzystuje istniejącą sesję. Dla każdego żądania pozyskania sesji sterownik wysyła:

• Wartość SessionTag.
• Rzeczywista konfiguracja platformy Spark, w tym ustawienia conf.* i EnvironmentId.

Usługa Fabric Livy ocenia te wartości i decyduje, czy przypisać połączenie do aktywnej sesji, czy utworzyć nową sesję. Aby poprawić ponowne wykorzystanie sesji, używaj spójnych wartości między połączeniami, które mają współdzielić przepustowość Sparka zarządzanego przez serwer.

Wyłącz tryb HC

Możesz wyłączyć nawiązywanie nowych sesji HC bez ponownego wdrażania sterownika, ustawiając parametry połączenia HC na false lub usuwając parametry HC:

hcEnabled=false;

Awaryjne wycofanie: Aby przeprowadzić awaryjne wycofanie, uruchom maszynę JVM za pomocą -Dcom.microsoft.fabric.jdbc.hc.disable=true. Ten przełącznik obejmujący całą JVM wymusza tryb klasyczny nawet jeśli parametry połączenia zawierają hcEnabled=true.

Zrozumienie zachowania środowiska wykonawczego HC

• Pozyskiwanie sesji HC blokuje się do momentu jej gotowości lub hcAcquireTimeoutSeconds wygaśnięcia.
• Zamknięcie połączenia w trakcie trwającej akwizycji przerywa akwizycję i wywołuje oczyszczanie sesji HC w miarę możliwości.
• Uwierzytelnianie, wykonywanie zapytań, anulowanie, instrukcje przygotowane, obsługa wyników oraz FabricEnvironmentID odbywają się za pośrednictwem standardowych interfejsów JDBC w obu trybach.

Przykłady użycia

Połączenie podstawowe

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

Połączenie w trybie wysokiej współbieżności

Poniższy przykład otwiera połączenie JDBC w trybie HC, uruchamia zapytanie i automatycznie zwalnia połączenie:

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

Wykonywanie zapytań

Proste zapytanie

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

Zapytanie z filtrem

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

Wykonywanie zapytań z limitem

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

Praca z zestawami wyników

Przeglądaj zbiory wyników

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

Zestawy wyników są tylko do odczytu i umożliwiają przewijanie tylko do przodu. Sterownik nie obsługuje metod takich jak first(), last(), i absolute().

Przetwarzanie dużych zestawów wyników

Sterownik paginuje wyniki według wielkości odpowiedzi. Skonfiguruj LivyStatementPageSize w adresie URL JDBC, aby ustawić żądany rozmiar strony w bajtach. Metoda ta Statement.setFetchSize() nie kontroluje pobierania ani zużycia pamięci.

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
}

Korzystanie z przygotowanych zapytań

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

Operacje wsadowe

Sterownik nie obsługuje transakcji JDBC. Wykonywanie wsadowe jest nietransakcyjne i może częściowo się powiódć. Późniejsza porażka nie cofa wcześniejszych udanych zdań.

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

Wybierz właściciela połączenia

Wybierz jeden komponent, który będzie zarządzał cyklem życia połączenia. Nie umieszczaj jednej puli połączeń w innej.

ConnectionPoolEnabled ma domyślną wartość false zarówno w kodzie sterownika, jak i w dostarczonych plikach właściwości. Nie jest to przełącznik opt-in dla obecnie istniejących punktów wejścia. Użyj LivyBuiltInPooledDataSource, gdy sterownik powinien zarządzać pulą, lub skonfiguruj zewnętrzną pulę połączeń, na przykład HikariCP. Pozostałe punkty wejścia tworzą bezpośrednie połączenia, które nie należą do puli.

Pulowanie połączeń z HikariCP

HikariCP może tworzyć połączenia zarówno z LivyDataSource, jak i z adresu JDBC URL za pomocą LivyDriver. Stwórz jedną instancję HikariCP, wykorzystaj ją ponownie przez cały okres życia aplikacji i używaj tylko jednej puli połączeń w obu konfiguracjach.

Zależność Maven

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

Użyj LivyDataSource

LivyDataSource tworzy niezależne fizyczne połączenia i nie posiada puli połączeń ani osobnego, bliskiego cyklu życia. Dostarczaj ją bezpośrednio do HikariCP, aby HikariCP był jedynym właścicielem puli:

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

Przykład ten umożliwia tryb HC w adresie URL JDBC. HikariCP zarządza połączeniami po stronie klienta, natomiast tryb HC zarządza współdzielonymi sesjami Spark w Fabric. Usuń hcEnabled i sessionTag jeśli aplikacja wymaga sesji klasycznych.

Użyj LivyDriver bezpośrednio

Skonfiguruj adres URL JDBC i klasę sterownika tak, aby HikariCP tworzył fizyczne połączenia przez LivyDriver. Ustawienie klasy sterownika jest opcjonalne, gdy dostępne jest automatyczne wykrywanie przez dostawcę usług JDBC. Ta konfiguracja nie tworzy instancji LivyDataSource, więc jej wbudowana pula nie jest używana:

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

Skorzystaj z następujących wskazówek przy konfigurowaniu puli:

Pełną listę opcji konfiguracji HikariCP można znaleźć tutaj:

• HikariConfig API reference
• Opcje konfiguracyjne HikariCP

Mapowanie typów danych

Sterownik mapuje typy danych Spark SQL na typy SQL JDBC i typy języka Java:

Treści powiązane

• Środowiska uruchomieniowe platformy Apache Spark w Fabric
• Środowisko uruchomieniowe Fabric 1.3
• Co to jest interfejs API Livy do inżynierii danych