spark MySQL数据库配置

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