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 数据库了。
