使用Spark连接MySQL数据库后，可以通过读取数据、执行查询、写入数据等方式进行操作。

Spark连接MySQL数据库后的使用

准备工作

1、安装并配置好Spark和MySQL数据库。

2、下载MySQL的JDBC驱动，并将其添加到Spark的classpath中。

创建SparkSession对象

1、导入必要的包：

import org.apache.spark.sql.SparkSession

2、创建SparkSession对象：

val spark = SparkSession.builder()
  .appName("Spark连接MySQL")
  .config("spark.driver.extraClassPath", "mysqlconnectorjavax.x.xx.jar") // 替换为实际的JDBC驱动路径
  .getOrCreate()

3、设置SparkSession的连接信息：

spark.conf.set("spark.jdbc.url", "jdbc:mysql://localhost:3306/database_name") // 替换为实际的数据库URL
spark.conf.set("spark.jdbc.driver", "com.mysql.jdbc.Driver") // 替换为实际的JDBC驱动类名
spark.conf.set("spark.jdbc.user", "username") // 替换为实际的用户名
spark.conf.set("spark.jdbc.password", "password") // 替换为实际的密码

4、读取MySQL数据库中的表数据：

val df = spark.read
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/database_name") // 替换为实际的数据库URL
  .option("driver", "com.mysql.jdbc.Driver") // 替换为实际的JDBC驱动类名
  .option("user", "username") // 替换为实际的用户名
  .option("password", "password") // 替换为实际的密码
  .option("dbtable", "table_name") // 替换为实际的表名
  .load()

5、对DataFrame进行操作：

df.show() // 显示前10行数据
df.printSchema() // 打印表结构
df.select("column1", "column2").filter($"column1" > 10).count() // 根据条件筛选并计算满足条件的记录数

保存DataFrame到MySQL数据库中

1、将DataFrame保存到MySQL表中：

df.write
  .mode("overwrite") // or "append" to save data to existing table without overwriting it
  .format("jdbc")
  .option("url", "jdbc:mysql://localhost:3306/database_name") // 替换为实际的数据库URL
  .option("driver", "com.mysql.jdbc.Driver") // 替换为实际的JDBC驱动类名
  .option("user", "username") // 替换为实际的用户名
  .option("password", "password") // 替换为实际的密码
  .option("dbtable", "table_name") // 替换为实际的表名
  .save()

问题与解答栏目

问题1：在创建SparkSession对象时，如何指定使用的JDBC驱动版本？

答案：在spark.driver.extraClassPath中指定JDBC驱动的路径时，可以根据实际情况修改驱动的版本号，如果使用MySQL Connector/J版本8，则可以将路径设置为"mysqlconnectorjava8.x.xx.jar"。

问题2：如何从MySQL数据库中读取多个表的数据？

答案：可以使用union或unionAll方法将多个DataFrame合并成一个DataFrame，分别读取每个表的数据，然后使用union或unionAll方法将它们合并起来。

val df1 = spark.read... // read from table1 in database_name database
val df2 = spark.read... // read from table2 in database_name database
val combinedDf = df1.union(df2) // combine the two tables into one using union method (you can also use unionAll)
combinedDf.show() // display the combined dataframe's content