spark dataframe使用jdbc

### 使用 Spark DataFrame 与 JDBC 进行数据库操作 #### 准备工作 为了使 Spark 能够通过 JDBC 访问外部数据库，如 MySQL 或 Oracle 数据库，需要准备相应的 JDBC 驱动程序。对于 MySQL 来说，这通常意味着下载并配置 `mysql-connector-java` 的 JAR 文件[^1]。 #### 创建 Maven 项目 当使用 IntelliJ IDEA 开发环境来构建基于 Spark 和 JDBC 的应用时，建议从创建一个新的 Maven 工程开始。Maven 是一种强大的工具链管理器，它可以帮助简化依赖管理和编译过程[^3]。 #### 添加必要的依赖项到 pom.xml 中 确保在项目的 `pom.xml` 文件里包含了如下所示的相关依赖声明： ```xml <dependencies> <!-- Spark SQL --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.12</artifactId> <version>3.4.0</version> </dependency> <!-- MySQL Connector Java --> <dependency> <groupId>mysql</groupId> <artifactId>mysql-connector-java</artifactId> <version>8.0.30</version> </dependency> </dependencies> ``` 这段 XML 片段定义了两个主要组件：一个是 Apache Spark SQL 库本身；另一个则是用于连接 MySQL 数据库所需的驱动程序。 #### 编写 Scala/Java 代码实现读取和保存数据至 MySQL 表格 一旦完成了上述准备工作之后，则可以着手编写实际的应用逻辑部分。这里给出一段简单的例子展示如何利用 Spark Session API 加载来自 MySQL 的表格作为 DataFrame 对象以及将本地集合转换成新的记录集再存入目标表内。 ```scala import org.apache.spark.sql.SparkSession val spark = SparkSession.builder() .appName("JDBC Example") .config("spark.master", "local[*]") .getOrCreate() // 定义连接属性 val connectionProperties = new java.util.Properties() connectionProperties.put("driver", "com.mysql.cj.jdbc.Driver") // 构建 URL 字符串 val jdbcUrl = s"jdbc:mysql://localhost:3306/testdb?useSSL=false" // 读取现有表中的全部内容 val dfRead = spark.read.jdbc(jdbcUrl, "employees", connectionProperties) dfRead.show() // 显示前几条记录 // 将新员工信息追加到 employees 表中 val newData = Seq( (null.asInstanceOf[java.lang.Integer], "John Doe"), (null.asInstanceOf[java.lang.Integer], "Jane Smith")) val employeeDF = spark.createDataFrame(newData).toDF("id", "name") employeeDF.write.mode("append").jdbc(jdbcUrl, "employees", connectionProperties) ``` 此脚本首先建立了通往测试数据库 testdb 下面名为 employees 的关系型表之间的桥梁，并从中提取所有字段构成 DataFrames 结构供后续处理分析之用。接着构造了一组模拟的新雇员姓名列表并通过调用 write 方法将其持久化回原位置完成整个流程闭环。