SparkSQL操作Mysql
Spark SQL 提供了与 MySQL 数据库交互的强大功能，可以方便地读取和写入 MySQL 数据。以下是完整的操作指南：
1. 准备工作
添加依赖
确保你的项目中包含 MySQL JDBC 驱动依赖：
<!-- Maven 依赖 -->
<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.28</version> <!-- 使用适合你的MySQL版本的驱动 -->
</dependency>创建 SparkSession
import org.apache.spark.sql.SparkSessionval spark = SparkSession.builder().appName("MySQL with Spark SQL").master("local[*]") // 生产环境应使用集群配置.getOrCreate()2. 从 MySQL 读取数据
基本读取方式
val jdbcDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/your_database").option("dbtable", "your_table").option("user", "your_username").option("password", "your_password").load()jdbcDF.show()使用 SQL 查询读取
val queryDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/your_database").option("query", "SELECT id, name FROM your_table WHERE age > 30").option("user", "your_username").option("password", "your_password").load()配置连接参数
val connectionProperties = new java.util.Properties()
connectionProperties.put("user", "your_username")
connectionProperties.put("password", "your_password")
connectionProperties.put("driver", "com.mysql.jdbc.Driver")val jdbcDF2 = spark.read.jdbc("jdbc:mysql://localhost:3306/your_database", "your_table", connectionProperties)3. 写入数据到 MySQL
基本写入方式
val data = Seq(("Alice", 25), ("Bob", 30), ("Charlie", 35))
val df = spark.createDataFrame(data).toDF("name", "age")df.write.format("jdbc").option("url", "jdbc:mysql://localhost:3306/your_database").option("dbtable", "new_table").option("user", "your_username").option("password", "your_password").mode("append") // 可以是 "overwrite", "append", "ignore" 或 "error".save()使用批量写入提高性能
df.write.option("batchsize", "10000") // 批量大小.option("isolationLevel", "READ_COMMITTED") // 事务隔离级别.jdbc("jdbc:mysql://localhost:3306/your_database", "new_table", connectionProperties)4. 高级配置
分区读取大数据表
val partitionedDF = spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/your_database").option("dbtable", "large_table").option("user", "your_username").option("password", "your_password").option("partitionColumn", "id") // 用于分区的列.option("lowerBound", "1") // 最小值.option("upperBound", "1000000") // 最大值.option("numPartitions", "10") // 分区数.load()连接池配置
connectionProperties.put("connectionPool", "HikariCP") // 使用连接池
connectionProperties.put("maximumPoolSize", "10") // 最大连接数5. 使用 Spark SQL 直接查询 MySQL
// 注册MySQL表为临时视图
spark.read.format("jdbc").option("url", "jdbc:mysql://localhost:3306/your_database").option("dbtable", "(SELECT * FROM your_table WHERE age > 30) as tmp").option("user", "your_username").option("password", "your_password").load().createOrReplaceTempView("mysql_data")// 使用Spark SQL查询
val result = spark.sql("SELECT name, age FROM mysql_data WHERE age < 40")
result.show()6. 性能优化建议
1. 分区读取：对大表使用分区读取选项
2. 批量写入：设置合适的 batchsize 参数
3. 列裁剪：只读取需要的列
4. 谓词下推：在查询中使用 WHERE 条件
5. 连接池：使用连接池管理数据库连接
6. 合理设置fetchsize：控制每次从数据库获取的行数
7. 常见问题解决
.option("sessionInitStatement", "SET time_zone='+08:00'") // 设置时区SSL 连接
.option("useSSL", "true")
.option("requireSSL", "true")编码问题
.option("characterEncoding", "UTF-8")通过以上方法，你可以高效地在 Spark 中操作 MySQL 数据库，实现大规模数据的处理和分析。