Spark 連接 MySQL 數據庫的配置
要讓 Spark 與 MySQL 數據庫實現連接，需要進行以下配置步驟。下面為你提供詳細的操作指南和示例代碼：
1. 添加 MySQL JDBC 驅動依賴
你得把 MySQL 的 JDBC 驅動添加到 Spark 的類路徑中。可以通過以下兩種方式來完成：
- 方法一：將 MySQL JDBC JAR 文件（mysql-connector-java-*.jar）復制到 Spark 的
jars目錄下。 - 方法二：在提交 Spark 作業時，使用
--jars參數指定 JDBC 驅動路徑。 - 方法三（針對 PySpark）：在代碼里設置
spark.jars.packages屬性。
2. 配置連接參數
連接 MySQL 數據庫時，需要配置以下參數：
- JDBC URL，格式為：
jdbc:mysql://<hostname>:<port>/<database> - 數據庫用戶名
- 數據庫密碼
- JDBC 驅動類名：
com.mysql.cj.jdbc.Driver
3. PySpark 連接示例
下面是使用 PySpark 連接 MySQL 數據庫并讀取數據的示例代碼：
python
運行
from pyspark.sql import SparkSession# 創建SparkSession
spark = SparkSession.builder \.appName("MySQL Connection Example") \.config("spark.jars.packages", "mysql:mysql-connector-java:8.0.26") \.getOrCreate()# 配置數據庫連接參數
jdbc_url = "jdbc:mysql://localhost:3306/your_database"
connection_properties = {"user": "your_username","password": "your_password","driver": "com.mysql.cj.jdbc.Driver"
}# 讀取數據
df = spark.read.jdbc(url=jdbc_url, table="your_table", properties=connection_properties)# 顯示數據
df.show()# 寫入數據到MySQL
data = [("Alice", 25), ("Bob", 30), ("Charlie", 35)]
columns = ["name", "age"]
df_to_write = spark.createDataFrame(data, columns)df_to_write.write.jdbc(url=jdbc_url,table="new_table",mode="overwrite",properties=connection_properties
)# 停止SparkSession
spark.stop()
4. Scala 連接示例
若使用 Scala 連接 MySQL 數據庫，可參考以下代碼：
scala
import org.apache.spark.sql.SparkSessionobject MySQLExample {def main(args: Array[String]): Unit = {// 創建SparkSessionval spark = SparkSession.builder.appName("MySQL Connection Example").config("spark.jars.packages", "mysql:mysql-connector-java:8.0.26").getOrCreate()// 配置數據庫連接參數val jdbcUrl = "jdbc:mysql://localhost:3306/your_database"val connectionProperties = new java.util.Properties()connectionProperties.setProperty("user", "your_username")connectionProperties.setProperty("password", "your_password")connectionProperties.setProperty("driver", "com.mysql.cj.jdbc.Driver")// 讀取數據val df = spark.read.jdbc(jdbcUrl, "your_table", connectionProperties)// 顯示數據df.show()// 寫入數據到MySQLval data = Seq(("Alice", 25), ("Bob", 30), ("Charlie", 35))val columns = Seq("name", "age")import spark.implicits._val dfToWrite = data.toDF(columns: _*)dfToWrite.write.jdbc(jdbcUrl, "new_table", connectionProperties)// 停止SparkSessionspark.stop()}
}
5. 常見問題解決辦法
- 驅動版本不兼容：要保證使用的 MySQL JDBC 驅動版本和你的 MySQL 服務器版本相匹配。
- 網絡連接問題：確認 MySQL 服務器正在運行，并且可以從 Spark 集群訪問。
- 權限問題：確保數據庫用戶擁有讀取或寫入指定表的權限。
按照上述步驟操作，你就能成功在 Spark 中配置并連接 MySQL 數據庫了。
 解析)
