本頁包含範例,展示如何使用 Databricks JDBC 驅動程式(版本 3 及以上)執行查詢。
備註
Databricks JDBC 驅動程式的參數化陳述式參數限制為 256。
範例:執行查詢
下列範例示範如何使用 Databricks JDBC 驅動程式 ,使用 Azure Databricks 計算資源來執行 Databricks SQL 查詢。
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
import java.util.Properties;
public class DatabricksJDBCExample {
public static void main(String[] args) {
Class.forName("com.databricks.client.jdbc.Driver");
// Set JDBC URL properties
String jdbcUrl = "jdbc:databricks://dbc-a1b2345c-d6e7.cloud.databricks.com:443";
Properties connectionProperties = new Properties();
connectionProperties.put("httpPath", "sql/protocolv1/o/123456780012345/0123-123450-z000pi22");
connectionProperties.put("ssl", "1");
// Set authentication properties (personal access token)
connectionProperties.put("AuthMech", "3");
connectionProperties.put("user", "token");
connectionProperties.put("password", "12345678901234667890abcdabcd");
// Set logging properties
connectionProperties.put("logPath", "logs/myapplication.log");
// Establish connection and execute query
try (Connection connection = DriverManager.getConnection(jdbcUrl, connectionProperties);
Statement statement = connection.createStatement();
ResultSet resultSet = statement.executeQuery("SELECT * FROM samples.nyctaxi.trips")) {
// Get metadata and column names
ResultSetMetaData metaData = resultSet.getMetaData();
String[] columns = new String[metaData.getColumnCount()];
for (int i = 0; i < columns.length; i++) {
columns[i] = metaData.getColumnName(i + 1);
}
// Process and print the result set
while (resultSet.next()) {
System.out.print("Row " + resultSet.getRow() + "=[");
for (int i = 0; i < columns.length; i++) {
if (i != 0) {
System.out.print(", ");
}
System.out.print(columns[i] + "='" + resultSet.getObject(i + 1) + "'");
}
System.out.println("]");
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
範例:以異步方式執行查詢
下列範例示範如何使用 Databricks JDBC 驅動程式 來執行和處理非同步 Databricks SQL 查詢。
如需 API 參考,請參閱 Databricks JDBC 驅動程式的 Java API 參考。
啟動語句的非同步執行:
Statement statement = conn.createStatement();
IDatabricksStatement dbStatement = statement.unwrap(IDatabricksStatement.class);
ResultSet result = dbStatement.executeAsync(sql);
IDatabricksResultSet asyncResult = result.unwrap(IDatabricksResultSet.class);
IExecutionStatus asyncStatus = asyncResult.getExecutionStatus();
long startTime = System.currentTimeMillis();
while ((asyncStatus.getExecutionState() == ExecutionState.RUNNING | asyncStatus.getExecutionState() == ExecutionState.PENDING) || (startTime + timeout < System.currentTimeMillis())) {
Thread.sleep(1000); // Sleep for 1000 ms
asyncResult = dbStatement.getExecutionResult().unwrap(IDatabricksResultSet.class);
asyncStatus = asyncResult.getExecutionStatus();
}
if (asyncStatus.getExecutionStatus() == ExecutionState.RUNNING | ExecutionState.PENDING) {
dbStatement.cancel();
}
if (asyncStatus.getExecutionStatus() == ExecutionState.SUCCEEDED) {
// process result set
}
if (asyncStatus.getExecutionStatus() == ExecutionState.FAILED) {
String sqlState = asyncStatus.getSqlState();
String errorMessage = asyncStatus.getErrorMessage();
// log error code and message
}
在單獨的線程中處理陳述:
Statement statement = conn1.createStatement();
IDatabricksStatement dbStatement = statement.unwrap(IDatabricksStatement.class);
ResultSet asyncResult = dbStatement.executeAsync(sql);
IDatabricksResultSet drs = asyncResult.unwrap(IDatabricksResultSet.class);
String statementId = drs.getStatementId();
ExecutionState state = drs.getExecutionStatus().getExecutionState();
while (state != ExecutionState.SUCCEEDED) {
Thread.sleep(sleepInterval);
asyncResult = dbStatement.getExecutionResult();
state = asyncResult.unwrap(IDatabricksResultSet.class).getExecutionStatus().getExecutionState();
}
// In another thread
IDatabricksConnection dbConn2 = conn2.unwrap(IDatabricksConnection.class);
IDatabricksStatement asyncStatementHandle = dbConn2.getStatement(statementId).unwrap(IDatabricksStatement.class);
IDatabricksResultSet asyncResultHandle = asyncStatementHandle.getExecutionResult().unwrap(IDatabricksResultSet.class);
// Cancel if needed
if (asyncResultHandle.getExecutionStatus().getExecutionState() == ExecutionState.PENDING | asyncResultHandle.getExecutionStatus().getExecutionState() == ExecutionState.RUNNING) {
asyncStatementHandle.cancel();
}
使用連線 ID 關閉連線:
// Get connection-Id from existing connection
String connectionId = conn.unwrap(IDatabricksConnection.class).getConnectionId();
// Close the connection from other thread using same JDBC Url and connection properties and connection-Id retrieved from above
com.databricks.client.jdbc.Driver.getInstance().closeConnection(jdbcUrl, properties, connectionId);
範例:查詢地理空間資料
以下範例展示了如何使用 Databricks JDBC 驅動程式查詢與檢索地理空間資料型態。 若要以結構化 Java 物件形式取得地理空間資料,請啟用 和 EnableComplexDatatypeSupportEnableGeoSpatialSupport connection 屬性。
欲了解更多關於地理空間資料類型與功能的資訊,請參見 ST 地理空間函數。
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.Statement;
import java.util.Properties;
import com.databricks.jdbc.api.IGeometry;
import com.databricks.jdbc.api.IGeography;
public class GeospatialExample {
public static void main(String[] args) {
// Set JDBC URL properties
String jdbcUrl = "jdbc:databricks://dbc-a1b2345c-d6e7.cloud.databricks.com:443";
Properties connectionProperties = new Properties();
connectionProperties.put("httpPath", "sql/protocolv1/o/123456780012345/0123-123450-z000pi22");
connectionProperties.put("ssl", "1");
// Set authentication properties (personal access token)
connectionProperties.put("AuthMech", "3");
connectionProperties.put("user", "token");
connectionProperties.put("password", "12345678901234667890abcdabcd");
// Enable geospatial support
connectionProperties.put("EnableComplexDatatypeSupport", "1");
connectionProperties.put("EnableGeoSpatialSupport", "1");
// Establish connection and execute geospatial query
try (Connection connection = DriverManager.getConnection(jdbcUrl, connectionProperties);
Statement statement = connection.createStatement();
ResultSet rs = statement.executeQuery(
"SELECT ST_Point(1.0, 2.0) as point, " +
"ST_GeogFromText('POINT(-122.4194 37.7749)') as location")) {
while (rs.next()) {
// Retrieve GEOMETRY object
IGeometry point = (IGeometry) rs.getObject("point");
System.out.println("Point WKT: " + point.getWKT());
System.out.println("Point SRID: " + point.getSRID());
// Retrieve GEOGRAPHY object
IGeography location = (IGeography) rs.getObject("location");
System.out.println("Location WKT: " + location.getWKT());
System.out.println("Location SRID: " + location.getSRID());
// Metadata
ResultSetMetaData meta = rs.getMetaData();
System.out.println("Column 1 type: " + meta.getColumnTypeName(1)); // GEOMETRY
System.out.println("Column 1 class: " + meta.getColumnClassName(1)); // com.databricks.jdbc.api.IGeometry
System.out.println("Column 2 type: " + meta.getColumnTypeName(2)); // GEOGRAPHY
System.out.println("Column 2 class: " + meta.getColumnClassName(2)); // com.databricks.jdbc.api.IGeography
}
} catch (Exception e) {
e.printStackTrace();
}
}
}