PySpark 通过 PySpark 连接 MySQL 数据库

在本文中，我们将介绍如何使用 PySpark 连接 MySQL 数据库，并进行数据的读取和写入操作。

阅读更多：PySpark 教程

准备工作

在开始使用 PySpark 连接 MySQL 数据库之前，需要确保已经完成以下准备工作：

1. 安装 Java Development Kit (JDK)，并设置 JAVA_HOME 环境变量。
2. 安装 Apache Spark，并设置 SPARK_HOME 环境变量。
3. 安装 PySpark 包，可以通过 pip 或者 conda 进行安装。

连接 MySQL 数据库

在 PySpark 中连接 MySQL 数据库主要有两种方式：使用 JDBC 连接和使用 PyMySQL 连接。下面将分别介绍这两种方式的具体操作。

使用 JDBC 连接

使用 JDBC 连接需要将 MySQL 的 JDBC 驱动程序添加到 Spark 的类路径中。首先，下载 MySQL 的 JDBC 驱动程序 (如 mysql-connector-java-8.0.26.jar) 并将其复制到 $SPARK_HOME/jars 目录下。

接下来，使用以下代码创建一个 SparkSession 对象，并指定 JDBC 连接参数：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("MySQL Connection") \
    .config("spark.jars", "$SPARK_HOME/jars/mysql-connector-java-8.0.26.jar") \
    .getOrCreate()

url = "jdbc:mysql://localhost:3306/database_name"
properties = {
    "user": "username",
    "password": "password"
}

df = spark.read.jdbc(url=url, table="table_name", properties=properties)

上述代码中，需要替换以下参数：

• database_name：MySQL 数据库名称
• username：MySQL 用户名
• password：MySQL 密码
• table_name：要读取的表名称

通过以上代码，我们可以将 MySQL 数据库中的数据读取到一个 PySpark DataFrame 中进行进一步的处理和分析。

使用 PyMySQL 连接

PyMySQL 是 Python 中的一个 MySQL 连接库，可以用于直接连接 MySQL 数据库。我们可以使用 PyMySQL 和 PySpark 结合起来，实现直接从 MySQL 数据库读取数据，如下所示：

import pymysql
from pyspark.sql import SparkSession

# 建立 MySQL 连接
connection = pymysql.connect(
    host='localhost',
    user='username',
    password='password',
    db='database_name',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

# 使用 SparkSession 创建 DataFrame
spark = SparkSession.builder.getOrCreate()
df = spark.createDataFrame(connection.cursor().fetchall())

# 关闭连接
connection.close()

上述代码中，需要替换以下参数：

• username：MySQL 用户名
• password：MySQL 密码
• database_name：MySQL 数据库名称

通过以上代码，我们可以直接使用 PySpark 从 MySQL 数据库中读取数据，并将其转换为 DataFrame 进行数据分析。

数据读写操作

在获得连接之后，我们可以对数据进行读取和写入的操作。

读取数据

为了从 MySQL 数据库中读取数据，我们可以使用 spark.read.jdbc() 方法。下面是一个从 employees 表中读取数据的示例：

df = spark.read.jdbc(url=url, table="employees", properties=properties)

我们也可以使用 SQL 查询语句来读取特定的数据，如下所示：

query = "(SELECT * FROM employees WHERE salary > 5000) tmp"
df = spark.read.jdbc(url=url, table=query, properties=properties)

在读取数据时，可以通过指定分区数来并行读取数据，以提高读取速度。例如，我们可以将数据分为 4 个分区进行并行读取：

df = spark.read.jdbc(url=url, table="employees", properties=properties, numPartitions=4)

写入数据

使用 PySpark 写入数据到 MySQL 数据库，我们可以使用 df.write.jdbc() 方法。下面是一个将 DataFrame 中的数据写入到 employees 表中的示例：

df.write.jdbc(url=url, table="employees", mode="append", properties=properties)

在写入数据时，可以通过指定 mode 参数来指定写入模式。常用的写入模式包括：

• append：追加模式，将数据追加到表的末尾。
• overwrite：覆盖模式，删除原有数据并写入新数据。
• ignore：忽略模式，如果表已存在，则不写入数据。
• error：错误模式，如果表已存在，则抛出异常。

总结

本文介绍了在 PySpark 中连接 MySQL 数据库的两种方式：使用 JDBC 连接和使用 PyMySQL 连接。我们还学习了如何进行数据的读取和写入操作，以及常用的读写模式。使用 PySpark 连接 MySQL 数据库可以方便地进行大规模数据分析和处理，为我们带来了更多处理数据的可能性。希望本文对你在 PySpark 中连接和操作 MySQL 数据库有所帮助！