spark MySQL数据库配置

news2025/5/18 7:42:50

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.SparkSession

object MySQLExample {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val 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()

    // 写入数据到MySQL
    val 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)

    // 停止SparkSession
    spark.stop()
  }
}