Databricks 第3篇：pyspark.sql 通過JDBC連接數據庫

Databricks Runtime 包含Azure SQL 數據庫的 JDBC 驅動程序，本文介紹如何使用數據幀 API 連接到使用 JDBC 的 SQL 數據庫，通過 JDBC 接口進行的讀取操作和更新操作。

在Databricks的Notebook中，spark是Databricks內置的一個SparkSession，可以通過該SparkSession來創建DataFrame、引用DataFrameReader和DataFrameWriter等。

一，創建JDBC URL

本文適用Python語言和JDBC驅動程序來連接Azure SQL Database，  

jdbcHostname = "Azure SQL Database"
jdbcDatabase = "db_name"
jdbcPort = 1433
jdbcUsername="user_name"
jdbcPassword="user_password"

jdbcUrl = "jdbc:sqlserver://{0}:{1};database={2}".format(jdbcHostname, jdbcPort, jdbcDatabase)
connectionProperties = {
  "user" : jdbcUsername,
  "password" : jdbcPassword,
  "driver" : "com.microsoft.sqlserver.jdbc.SQLServerDriver"
}

二，把查詢向下推送到數據庫引擎

可把整個查詢或表向下推送到數據庫，且只返回結果。table 參數標識要讀取的 JDBC 表。

spark.read.jdbc(url, table, column=None, lowerBound=None, upperBound=None, numPartitions=None, predicates=None, properties=None)

參數注釋：

• url：JDBC URL
• table：表名或查詢
• column、numPartitions、lowerBound和upperBound：用於指定分區的列名，分區的數量，分區的列的最小值和最大值
• predicates：謂詞，用於對數據進行過濾，類似於Where子句
• properties：JDBC數據庫連接參數的字典

1，向下推送表

如果把table設置為表名，那么表示查詢整個表。

pushdown_query = "table_name"
df = spark.read.jdbc(url=jdbcUrl, table=pushdown_query, properties=connectionProperties)
display(df)

對查詢的結果進行投影和過濾：

spark.read.jdbc(jdbcUrl, table=pushdown_query, connectionProperties).select("carat", "cut", "price").where("cut = 'Good'")

2，向下推送查詢

如果向下推送查詢，那么需要采用下方的格式：(query) data_alias

pushdown_query = "(select * from employees where emp_no < 10008) emp_alias"
df = spark.read.jdbc(url=jdbcUrl, table=pushdown_query, properties=connectionProperties)
display(df)

三，向下推送更新

通過JDBC，把DataFrame的內容存儲到外部數據表中： 

spark.write.jdbc(url, table, mode=None, properties=None)

參數注釋：

• url：JDBC Url
• table：外部數據庫的表名
• mode：數據更新的模式，append、overwrite、ignore、error（默認，如果數據存在，拋出異常）
• properties：JDBC數據庫連接參數的字典

 

 

 

參考文檔：