โปรแกรมควบคุม Microsoft JDBC สําหรับ Microsoft Fabric Data Engineering (พรีวิว)

JDBC (Java Database Connectivity) เป็นมาตรฐานที่นํามาใช้กันอย่างแพร่หลายซึ่งช่วยให้แอปพลิเคชันไคลเอ็นต์สามารถเชื่อมต่อและทํางานกับข้อมูลจากฐานข้อมูลและแพลตฟอร์มข้อมูลขนาดใหญ่ได้

โปรแกรมควบคุม Microsoft JDBC สําหรับวิศวกรรมข้อมูล Fabric ช่วยให้คุณเชื่อมต่อ คิวรี และจัดการปริมาณงาน Spark ใน Microsoft Fabric ด้วยความน่าเชื่อถือและความเรียบง่ายของมาตรฐาน JDBC ไดรเวอร์ที่สร้างขึ้นบน Livy API ของ Microsoft Fabric ให้การเชื่อมต่อ Spark SQL ที่ปลอดภัยและยืดหยุ่นกับแอปพลิเคชัน Java และเครื่องมือ BI ของคุณ การผสานรวมนี้ช่วยให้คุณสามารถส่งและเรียกใช้โค้ด Spark ได้โดยตรงโดยไม่จําเป็นต้องสร้างสิ่งประดิษฐ์ Notebook หรือ Spark Job Definition แยกต่างหาก

คุณสมบัติหลัก

• สอดคล้องกับ JDBC 4.2: การใช้งานข้อกําหนด JDBC 4.2 อย่างเต็มรูปแบบ
• การรับรองความถูกต้องของ Microsoft Entra ID: โฟลว์การรับรองความถูกต้องหลายขั้นตอน รวมถึงการโต้ตอบ ข้อมูลประจําตัวไคลเอ็นต์ และการรับรองความถูกต้องตามใบรับรอง
• Enterprise Connection Pooling: การรวมการเชื่อมต่อในตัวพร้อมการตรวจสอบความสมบูรณ์และการกู้คืนอัตโนมัติ
• รองรับการสืบค้นแบบเนทีฟของ Spark SQL: การดําเนินการโดยตรงของคําสั่ง Spark SQL โดยไม่ต้องแปล
• รองรับประเภทข้อมูลที่ครอบคลุม: รองรับประเภทข้อมูล Spark SQL ทั้งหมดรวมถึงประเภทที่ซับซ้อน (ARRAY, MAP, STRUCT)
• การดึงข้อมูลล่วงหน้าของชุดผลลัพธ์แบบอะซิงโครนัส: การโหลดข้อมูลพื้นหลังเพื่อประสิทธิภาพที่ดีขึ้น
• รูปแบบเซอร์กิตเบรกเกอร์: ป้องกันความล้มเหลวแบบเรียงซ้อนด้วยการลองใหม่อัตโนมัติ
• การเชื่อมต่อใหม่อัตโนมัติ: การกู้คืนเซสชันแบบโปร่งใสเมื่อการเชื่อมต่อล้มเหลว
• การสนับสนุนพร็อกซี: การกําหนดค่าพร็อกซี HTTP และ SOCKS สําหรับสภาพแวดล้อมขององค์กร

ข้อกําหนดเบื้องต้น

ก่อนใช้ไดรเวอร์ Microsoft JDBC สําหรับ Microsoft Fabric Data Engineering ตรวจสอบให้แน่ใจว่าคุณมี:

• Java Development Kit (JDK): เวอร์ชัน 11 หรือสูงกว่า (แนะนําให้ใช้ Java 21)
• การเข้าถึง Microsoft Fabric: การเข้าถึงพื้นที่ทํางาน Microsoft Fabric
• ข้อมูลประจําตัว Azure Entra ID: ข้อมูลประจําตัวที่เหมาะสมสําหรับการรับรองความถูกต้อง
• รหัสพื้นที่ทํางานและเลคเฮาส์: ตัวระบุ GUID สําหรับพื้นที่ทํางาน Fabric และเลคเฮาส์ของคุณ

ดาวน์โหลดและติดตั้ง

ไดรเวอร์ Microsoft JDBC สําหรับ Microsoft Fabric Data Engineering เวอร์ชัน 1.0.0 เป็นเวอร์ชันตัวอย่างสาธารณะและรองรับ Java 11, 17 และ 21 เรากําลังปรับปรุงการสนับสนุนการเชื่อมต่อ Java อย่างต่อเนื่อง และขอแนะนําให้คุณทํางานกับโปรแกรมควบคุม Microsoft JDBC เวอร์ชันล่าสุด

• ดาวน์โหลดโปรแกรมควบคุม Microsoft JDBC สําหรับวิศวกรรมข้อมูล Microsoft Fabric (zip)
• ดาวน์โหลดโปรแกรมควบคุม Microsoft JDBC สําหรับ Microsoft Fabric Data Engineering (tar)

1. ดาวน์โหลดไฟล์ zip หรือ tar จากลิงก์ด้านบน
2. แตกไฟล์ที่ดาวน์โหลดเพื่อเข้าถึงไฟล์ JAR ของไดรเวอร์
3. เลือกไฟล์ JAR ที่ตรงกับเวอร์ชัน JRE ของคุณ:
   • สําหรับ Java 11: ms-sparksql-jdbc-1.0.0.jre11.jar
   • สําหรับ Java 17: ms-sparksql-jdbc-1.0.0.jre17.jar
   • สําหรับ Java 21: ms-sparksql-jdbc-1.0.0.jre21.jar
4. เพิ่มไฟล์ JAR ที่เลือกลงในคลาสพาธของแอปพลิเคชัน
5. สําหรับไคลเอ็นต์ JDBC ให้กําหนดค่าคลาสไดรเวอร์ JDBC: com.microsoft.spark.livy.jdbc.LivyDriver

ตัวอย่างการเริ่มต้นอย่างรวดเร็ว

ตัวอย่างนี้สาธิตวิธีการเชื่อมต่อกับ Microsoft Fabric และดําเนินการแบบสอบถามโดยใช้โปรแกรมควบคุม Microsoft JDBC สําหรับวิศวกรรมข้อมูล Microsoft Fabric ก่อนเรียกใช้รหัสนี้ ตรวจสอบให้แน่ใจว่าคุณได้ทําตามข้อกําหนดเบื้องต้นและติดตั้งไดรเวอร์แล้ว

import java.sql.*;

public class QuickStartExample {
    public static void main(String[] args) {
        // Connection string with required parameters
        String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
                     "FabricWorkspaceID=<workspace-id>;" +
                     "FabricLakehouseID=<lakehouse-id>;" +
                     "AuthFlow=2;" +  // Azure CLI based authentication
                     "LogLevel=INFO";
        
        try (Connection conn = DriverManager.getConnection(url)) {
            // Execute a simple query
            try (Statement stmt = conn.createStatement();
                 ResultSet rs = stmt.executeQuery("SELECT 'Hello from Fabric!' as message")) {
                
                if (rs.next()) {
                    System.out.println(rs.getString("message"));
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }
}

รูปแบบสตริงการเชื่อมต่อ

สตริงการเชื่อมต่อพื้นฐาน

โปรแกรมควบคุม Microsoft JDBC สําหรับวิศวกรรมข้อมูล Microsoft Fabric ใช้รูปแบบสายอักขระการเชื่อมต่อต่อไปนี้:

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

ส่วนประกอบสตริงการเชื่อมต่อ

ตัวอย่างสตริงการเชื่อมต่อ

การเชื่อมต่อพื้นฐาน (การรับรองความถูกต้องตามเบราว์เซอร์แบบโต้ตอบ)

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

ด้วยการกําหนดค่าทรัพยากร 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

ด้วยคุณสมบัติเซสชัน 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

การรับรองความถูกต้อง

โปรแกรมควบคุม Microsoft JDBC สําหรับ Microsoft Fabric Data Engineering สนับสนุนวิธีการรับรองความถูกต้องหลายวิธีผ่าน Microsoft Entra ID (เดิมคือ Azure Active Directory) การรับรองความถูกต้องถูกกําหนดค่าโดยใช้ AuthFlow พารามิเตอร์ในสตริงการเชื่อมต่อ

โฟลว์การรับรองความถูกต้อง

การรับรองความถูกต้องของเบราว์เซอร์แบบโต้ตอบ

ดีที่สุดสําหรับ: การพัฒนาและแอปพลิเคชันแบบโต้ตอบ

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

Connection conn = DriverManager.getConnection(url);

พารามิเตอร์:

• AuthFlow=1: ระบุการรับรองความถูกต้องของเบราว์เซอร์แบบโต้ตอบ
• AuthTenantID (ไม่บังคับ): รหัสผู้เช่า Azure
• AuthClientID (ไม่บังคับ): รหัสแอปพลิเคชัน (ไคลเอ็นต์)

พฤติกรรม:

• เปิดหน้าต่างเบราว์เซอร์สําหรับการตรวจสอบผู้ใช้
• ข้อมูลประจําตัวจะถูกแคชสําหรับการเชื่อมต่อครั้งต่อไปจนกว่าจะหมดอายุ
• เหมาะสําหรับการใช้งานแบบผู้ใช้คนเดียว

ข้อมูลประจําตัวไคลเอ็นต์หรือการรับรองความถูกต้องของบริการหลัก

ดีที่สุดสําหรับ: บริการอัตโนมัติและงานเบื้องหลัง

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

พารามิเตอร์ที่จําเป็น:

• AuthFlow=3: ระบุการรับรองความถูกต้องของข้อมูลประจําตัวไคลเอ็นต์
• AuthClientID: รหัสแอปพลิเคชัน (ไคลเอ็นต์) จาก Microsoft Entra ID
• AuthClientSecret: ข้อมูลลับของไคลเอ็นต์จาก Microsoft Entra ID
• AuthTenantID: รหัสผู้เช่า Azure

แนวทางปฏิบัติที่ดีที่สุด:

• จัดเก็บข้อมูลลับอย่างปลอดภัย (Azure Key Vault, ตัวแปรสภาพแวดล้อม)
• ใช้ข้อมูลประจําตัวที่มีการจัดการเมื่อเป็นไปได้
• หมุนเวียนความลับอย่างสม่ําเสมอ

การรับรองความถูกต้องของ Certificate-Based

ดีที่สุดสําหรับ: แอปพลิเคชันระดับองค์กรที่ต้องการการรับรองความถูกต้องตามใบรับรอง

String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=4;" +  
             "AuthClientID=<client-id>;" +
             "AuthCertificatePath=/path/to/certificate.pfx;" +
             "AuthCertificatePassword=<certificate-password>;" +
             "AuthTenantID=<tenant-id>;" +
             "LogLevel=INFO";

Connection conn = DriverManager.getConnection(url);

พารามิเตอร์ที่จําเป็น:

• AuthFlow=4: ระบุการรับรองความถูกต้องตามใบรับรอง
• AuthClientID: รหัสแอปพลิเคชัน (ไคลเอ็นต์)
• AuthCertificatePath: เส้นทางไปยังไฟล์ใบรับรอง PFX/PKCS12
• AuthCertificatePassword: รหัสผ่านใบรับรอง
• AuthTenantID: รหัสผู้เช่า Azure

การรับรองความถูกต้องของโทเค็นการเข้าถึง

ดีที่สุดสําหรับ: สถานการณ์การรับรองความถูกต้องแบบกําหนดเอง

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

การแคชการรับรองความถูกต้อง

โปรแกรมควบคุมจะแคชโทเค็นการรับรองความถูกต้องโดยอัตโนมัติเพื่อปรับปรุงประสิทธิภาพ:

// Enable/disable caching (enabled by default)
String url = "jdbc:fabricspark://api.fabric.microsoft.com;" +
             "FabricWorkspaceID=<workspace-id>;" +
             "FabricLakehouseID=<lakehouse-id>;" +
             "AuthFlow=2;" +
             "AuthEnableCaching=true;" +  // Enable token caching
             "AuthCacheTTLMS=3600000";    // Cache TTL: 1 hour

Connection conn = DriverManager.getConnection(url);

พารามิเตอร์การกําหนดค่า

พารามิเตอร์ที่จำเป็น

พารามิเตอร์เหล่านี้ต้องมีอยู่ในทุกสตริงการเชื่อมต่อ:

พารามิเตอร์เผื่อเลือก

การกําหนดค่าเวอร์ชัน API

การกําหนดค่าสภาพแวดล้อม

การกําหนดค่า Spark

การกําหนดค่าทรัพยากรเซสชัน

กําหนดค่าทรัพยากรเซสชัน Spark เพื่อประสิทธิภาพสูงสุด:

ตัวอย่าง:

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

คุณสมบัติเซสชัน Spark แบบกําหนดเอง

พารามิเตอร์ใดๆ ที่มีคํานํา spark. หน้าจะถูกนําไปใช้กับเซสชัน Spark โดยอัตโนมัติ:

ตัวอย่างการกําหนดค่า 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

เอ็นจิ้นการดําเนินการดั้งเดิม (NEE):

spark.nee.enabled=true

ตัวอย่างที่สมบูรณ์:

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

กําหนดค่าการรวมการเชื่อมต่อ HTTP เพื่อประสิทธิภาพเครือข่ายที่ดีที่สุด:

ตัวอย่าง:

HttpMaxTotalConnections=200;HttpMaxConnectionsPerRoute=100;HttpConnectionTimeoutInSeconds=60

การกําหนดค่าพร็อกซี

กําหนดการตั้งค่าพร็อกซี HTTP และ SOCKS สําหรับสภาพแวดล้อมขององค์กร:

ตัวอย่างพร็อกซี HTTP:

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

ตัวอย่างพร็อกซี SOCKS:

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

การกําหนดค่าการบันทึก

ตัวอย่าง:

LogLevel=DEBUG

ตําแหน่งบันทึกเริ่มต้น:

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

การกําหนดค่าบันทึกที่กําหนดเอง: ใช้แบบกําหนดเอง log4j2.xml หรือ logback.xml ไฟล์ในเส้นทางชั้นเรียนของคุณ

ตัวอย่างการใช้งาน

การเชื่อมต่อพื้นฐาน

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

การดําเนินการสืบค้น

แบบสอบถามอย่างง่าย

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

คิวรีที่มีตัวกรอง

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

คิวรีที่มีขีดจํากัด

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

การทํางานกับชุดผลลัพธ์

การนําทางชุดผลลัพธ์

public void navigateResultSet(Connection conn) throws SQLException {
    String sql = "SELECT id, name, amount FROM orders";
    
    try (Statement stmt = conn.createStatement(
            ResultSet.TYPE_SCROLL_INSENSITIVE,
            ResultSet.CONCUR_READ_ONLY);
         ResultSet rs = stmt.executeQuery(sql)) {
        
        // Move to first row
        if (rs.first()) {
            System.out.println("First row: " + rs.getString("name"));
        }
        
        // Move to last row
        if (rs.last()) {
            System.out.println("Last row: " + rs.getString("name"));
            System.out.println("Total rows: " + rs.getRow());
        }
        
        // Move to specific row
        if (rs.absolute(5)) {
            System.out.println("Row 5: " + rs.getString("name"));
        }
    }
}

การประมวลผลชุดผลลัพธ์ขนาดใหญ่

public void processLargeResultSet(Connection conn) throws SQLException {
    String sql = "SELECT * FROM large_table";
    
    try (Statement stmt = conn.createStatement()) {
        // Set fetch size for efficient memory usage
        stmt.setFetchSize(1000);
        
        try (ResultSet rs = stmt.executeQuery(sql)) {
            int rowCount = 0;
            while (rs.next()) {
                // Process row
                processRow(rs);
                rowCount++;
                
                if (rowCount % 10000 == 0) {
                    System.out.println("Processed " + rowCount + " rows");
                }
            }
            System.out.println("Total rows processed: " + rowCount);
        }
    }
}

private void processRow(ResultSet rs) throws SQLException {
    // Process individual row
}

การใช้ข้อความที่เตรียมไว้

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

การดําเนินการแบทช์

public void executeBatchInsert(Connection conn) throws SQLException {
    String sql = "INSERT INTO logs (timestamp, level, message) VALUES (?, ?, ?)";
    
    try (PreparedStatement pstmt = conn.prepareStatement(sql)) {
        conn.setAutoCommit(false);  // Disable auto-commit for batch
        
        // Add multiple statements to batch
        for (int i = 0; i < 1000; i++) {
            pstmt.setTimestamp(1, new Timestamp(System.currentTimeMillis()));
            pstmt.setString(2, "INFO");
            pstmt.setString(3, "Log message " + i);
            pstmt.addBatch();
            
            // Execute batch every 100 statements
            if (i % 100 == 0) {
                pstmt.executeBatch();
                pstmt.clearBatch();
            }
        }
        
        // Execute remaining statements
        pstmt.executeBatch();
        conn.commit();
        
        System.out.println("Batch insert completed successfully");
    } catch (SQLException e) {
        conn.rollback();
        throw e;
    } finally {
        conn.setAutoCommit(true);
    }
}

การแม็ปชนิดข้อมูล

ไดรเวอร์แมปประเภทข้อมูล Spark SQL กับประเภท JDBC SQL และประเภท Java:

เนื้อหาที่เกี่ยวข้อง

• รันไทม์ Apache Spark ใน Fabric
• รันไทม์ของผ้า 1.3
• Livy API สําหรับวิศวกรรมข้อมูลคืออะไร