使用spark rdd连接mysql数据源

小鱼儿 2023-06-11 09:29 6阅读 0赞

**Spark 支持通过Java JDBC访问关系型数据库。需要通过JdbcRDD进行，示例如下:**

#### 1. 在pom.xml中添加mysql的依赖 ####

<dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.27</version>
        </dependency>

#### 2. Mysql读取过程： ####

##### 2.1 定义连接mysql的参数 #####

val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://localhost:3306/tmall_ssm"
    val userName = "root"
    val passWd = "admin111"

##### 2.2 创建jdbcrdd访问数据库 #####

查看`jdbcrdd`源码，其中`lowerBlowound`和`upperBound`代表数据范围

class JdbcRDD[T: ClassTag](
        sc: SparkContext,
        getConnection: () => Connection,
        sql: String,
        lowerBlowound: Long,
        upperBound: Long,
        numPartitions: Int,
        mapRow: (ResultSet) => T = JdbcRDD.resultSetToObjectArray _)
      extends RDD[T](sc, Nil) with Logging

代码

val rdd = new JdbcRDD(sc, () => { 
     //获取数据库连接对象
          Class.forName(driver)
          DriverManager.getConnection(url, userName, passWd)
        },
          "select * from `user` where `id`>=? and id <=?;",
          1,
          10,
          1,
          //从查询结果拿数据
          r => (r.getInt(1), r.getString(2))
        )

##### 2.3 打印结果 #####

rdd.foreach(println)

这是我的`user`表结构  
![在这里插入图片描述][20191030103706161.png]  
**控制台打印：**

(1,lss)
    (2,admin)
    (3,zh)
    (4,zgw)

#### 3. Mysql插入过程： ####

//插入数据
        val value: RDD[(String, String)] = sc.makeRDD(List(("zhangsan","zhangsan"),("zhangsan1","zhangsan1")))
        value.foreach{ 
          case (name,password)=>{ 
            Class.forName(driver)
            val connection: Connection = DriverManager.getConnection(url, userName, passWd)
            var sql = "insert into user (name,password) values (?,?) "
            val prepareStatement: PreparedStatement = connection.prepareStatement(sql)
            prepareStatement.setString(1,name)
            prepareStatement.setString(2,password)
            prepareStatement.executeUpdate()
            prepareStatement.close()
    
          }
        }

查看`mysql`结果，后面两列请忽视…  
![在这里插入图片描述][20191030104235992.png]  
**改进：**

val value: RDD[(String, String)] = sc.makeRDD(List(("zhangsan","zhangsan"),("zhangsan1","zhangsan1")),2)
        /** * 在executor里面执行，优化连接方式 */
       value.foreachPartition(dates=>{ 
         Class.forName(driver)
         val connection: Connection = DriverManager.getConnection(url, userName, passWd)
         dates.foreach{ 
           case (name,password)=>{ 
             var sql = "insert into user (name,password) values (?,?) "
             val prepareStatement: PreparedStatement = connection.prepareStatement(sql)
             prepareStatement.setString(1,name)
             prepareStatement.setString(2,password)
             prepareStatement.executeUpdate()
             prepareStatement.close()
           }
         }
         connection.close()
    
       })

优化了连接方式，节省了connection资源，以分区做循环，比较占用内存空间，可能导致oom

完整代码托管于[https://github.com/daizikaikou/learningSpark][https_github.com_daizikaikou_learningSpark]

[20191030103706161.png]: /images/20230531/3863a5135fe24a16be275e79287fb87f.png
[20191030104235992.png]: /images/20230531/32c89afb5ef84369a840e356220fa8b8.png
[https_github.com_daizikaikou_learningSpark]: https://github.com/daizikaikou/learningSpark