河南省住房城乡建设厅官方网站wordpress新主题去版权
以下是使用 Spark/SparkSQL 连接 MySQL 数据库、添加数据和读取数据的完整示例（需提前准备 MySQL 驱动包）：
一、环境准备
1. 下载 MySQL 驱动
- 下载 mysql-connector-java-8.0.33.jar （或对应版本），放入 Spark 的 jars 目录，或提交任务时用 --jars 指定路径。
2. 启动 SparkSession
scala
import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder()
.appName("Spark MySQL Demo")
.master("local[*]") // 本地模式，集群环境需调整
.getOrCreate()
二、连接 MySQL 并读取数据
方式 1：使用 spark.read.jdbc 读取
scala
// 配置连接参数
val jdbcUrl = "jdbc:mysql://localhost:3306/test_db"
val tableName = "users"
val connectionProperties = new java.util.Properties()
connectionProperties.put("user", "root")
connectionProperties.put("password", "your_password")
connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver")
// 读取表数据
val df = spark.read.jdbc(jdbcUrl, tableName, connectionProperties)
// 显示数据
df.show()
方式 2：使用 SparkSQL 执行查询
scala
// 直接执行 SQL 查询（需指定完整表名）
val queryDF = spark.read.jdbc(
jdbcUrl,
s"(SELECT id, name FROM $tableName WHERE age > 18) AS temp", // 子查询防注入
connectionProperties
)
queryDF.show()
三、向 MySQL 插入数据
场景 1：写入新表（自动创建）
scala
// 创建示例数据
val data = Seq(
(3, "Alice", 25, java.sql.Timestamp.valueOf("2023-01-01 00:00:00")),
(4, "Bob", 30, java.sql.Timestamp.valueOf("2023-02-02 00:00:00"))
)
val schema = StructType(
Seq(
StructField("id", IntegerType, false),
StructField("name", StringType, true),
StructField("age", IntegerType, true),
StructField("create_time", TimestampType, true)
)
)
val insertDF = spark.createDataFrame(data, schema)
// 写入 MySQL（若表不存在则自动创建，需确保库存在）
insertDF.write.format("jdbc")
.option("url", jdbcUrl)
.option("dbtable", "users") // 表名
.option("user", "root")
.option("password", "your_password")
.option("driver", "com.mysql.cj.jdbc.Driver")
.mode("append") // 追加模式（可选：overwrite/ignore/replace）
.save()
场景 2：向现有表追加数据
scala
// 读取现有数据并新增记录
val existingDF = spark.read.jdbc(jdbcUrl, tableName, connectionProperties)
val newData = Seq((5, "Charlie", 35, java.sql.Timestamp.valueOf("2023-03-03 00:00:00")))
val newDF = spark.createDataFrame(newData, schema)
// 合并后写入（追加模式）
val combinedDF = existingDF.union(newDF)
combinedDF.write.jdbc(
jdbcUrl,
tableName,
connectionProperties,
"append" // 模式可通过第四个参数指定
)
四、关键参数说明
mode：
写入模式： append （追加）、 overwrite （覆盖）、 ignore （忽略冲突）
dbtable ：
目标表名（支持库名.表名格式，如 test_db.users ）
partitionColumn ：
分区列（大数据量时用于并行读取，需配合 lowerBound / upperBound ）
fetchSize ：
每次从数据库拉取的行数（优化性能，默认 1000）
createTableOptions ：
建表时的额外参数（如 ENGINE=InnoDB CHARSET=utf8 ）
五、注意事项
1. 驱动版本匹配
- MySQL 8.0+ 需使用 mysql-connector-java-8.0+ ，低版本数据库用 5.1.x 驱动。
2. 权限问题
- 确保 MySQL 用户有 INSERT / SELECT 权限：
sql
GRANT INSERT, SELECT ON test_db.* TO 'user'@'localhost';
3. 大数据量优化
- 并行写入：通过 numPartitions 和 partitionColumn 分区（需指定主键或索引列）。
- 批量提交：设置 batchSize=1000 减少连接开销：
scala
.option("batchSize", "1000")
4. 类型映射
- Spark 与 MySQL 类型需匹配（如 StringType → VARCHAR ， TimestampType → DATETIME ）。
六、完整示例（Scala 版）
scala
import org.apache.spark.sql.{SparkSession, Row}
import org.apache.spark.sql.types.{StructType, StructField, IntegerType, StringType, TimestampType}
import java.sql.Timestamp
// 1. 初始化 SparkSession
val spark = SparkSession.builder()
.appName("MySQL Demo")
.master("local[*]")
.getOrCreate()
// 2. 定义连接参数
val jdbcUrl = "jdbc:mysql://localhost:3306/test_db"
val tableName = "users"
val props = new java.util.Properties()
props.setProperty("user", "root")
props.setProperty("password", "your_password")
props.setProperty("driver", "com.mysql.cj.jdbc.Driver")
// 3. 读取数据
val df = spark.read.jdbc(jdbcUrl, tableName, props)
println("读取的数据：")
df.show()
// 4. 准备插入数据
val newData = Seq(
Row(6, "David", 28, new Timestamp(System.currentTimeMillis()))
)
val schema = new StructType(
Array(
StructField("id", IntegerType, nullable = false),
StructField("name", StringType, nullable = true),
StructField("age", IntegerType, nullable = true),
StructField("create_time", TimestampType, nullable = true)
)
)
val insertDF = spark.createDataFrame(spark.sparkContext.parallelize(newData), schema)
// 5. 插入数据（追加模式）
insertDF.write.jdbc(jdbcUrl, tableName, "append", props)
println("数据插入完成！")
// 6. 验证插入结果
val updatedDF = spark.read.jdbc(jdbcUrl, tableName, props)
updatedDF.show()
// 7. 停止 SparkSession
spark.stop()
执行后可通过 MySQL 客户端验证数据是否正确写入。