spark直接连接postgre数据库

Myth丶恋晨 2022-04-23 15:54 199阅读 0赞

注意点：

数据存入Mysql或是postgre注意事项  
A. 尽量先设置好存储模式  
默认为SaveMode.ErrorIfExists模式，该模式下，如果数据库中已经存在该表，则会直接报异常，导致数据不能存入数据库.另外三种模式如下：  
SaveMode.Append 如果表已经存在，则追加在该表中；若该表不存在，则会先创建表，再插入数据；  
SaveMode.Overwrite 重写模式，其实质是先将已有的表及其数据全都删除，再重新创建该表，最后插入新的数据；  
SaveMode.Ignore 若表不存在，则创建表，并存入数据；在表存在的情况下，直接跳过数据的存储，不会报错。  
B. 设置存储模式的步骤为:  
org.apache.spark.sql.SaveMode  
......  
df.write.mode(SaveMode.Append)  
C. 若提前在数据库中手动创建表，需要注意列名称和数据类型，  
**需要保证Spark SQL中schema中的field name与Mysql中的列名称一致！**

**特别注意: Scala中的String类型，在MySQL中对应的是Text类型（经过亲自测试所知）**

**上面是本人在Spark SQL 读取与写入Mysql方面的遇到的一些坑，特在此备忘。**

package com.emg.rec_match
    
    import java.util.Properties
    
    import com.emg.real.RealDataDAO.{WayAllDAO, WayRealDAO}
    import com.emg.rec_match.model._
    import com.emg.rec_match.pool.CreatePGSqlPoolForBigdata
    import org.apache.log4j.Logger
    import org.apache.spark.SparkConf
    import org.apache.spark.rdd.RDD
    import org.apache.spark.sql.{SaveMode, SparkSession}
    
    import scala.collection.mutable.ListBuffer
    
    
    object RecMatch {
    
      private val logger = Logger.getLogger(RecMatch.getClass)
    
      def main(args: Array[String]): Unit = {
    
        //创建sparkSession
        val conf = new SparkConf().setAppName("matcher2") //.setMaster("local[*]")
          .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
          //.set("spark.kryoserializer.buffer.max", "512m")
          .registerKryoClasses(Array[Class[_]](AvgSpeedBuffer.getClass, Point.getClass, UsefulFields.getClass,
          DistAndDegree.getClass, Uuid2oriData.getClass, Uuid2midData.getClass, Uuid2extData.getClass, RoadPart.getClass, Uuid.getClass))
        val spark = SparkSession.builder().config(conf).getOrCreate()
        val sc = spark.sparkContext
    
        CreateTableDAODay.createTable(utils.CurrentTimeUtil.getFrontHour())
    
        import spark.implicits._
        //接收数据
        val dataRDD = sc.textFile(args(0))
    
        val value = dataRDD.map(
          rdd => {
            val line = rdd.split(",", rdd.length)
            val uuid = line(0).toLong
            val lat = line(3).toDouble
            val lon = line(4).toDouble
            val dir = line(7).toDouble
            val speed = line(8).toDouble
            val utc = line(9).toLong
            UsefulFields(uuid, lat, lon, dir, speed, utc)
          }
        )
    
    
        val len = sc.longAccumulator //累加器，计数使用
    
        val rdds: RDD[Uuid2extData] = value.mapPartitions(partition => {
          val datas: List[Uuid2oriData] = WayAllDAO.getRoadIdBatch(partition.toList)
          val result: List[Uuid2extData] = WayRealDAO.getMatchRoadID(datas)
          //len.add(result.length)
          /*      result.map(uuid => {
                  Uuid(uuid.uuid, uuid.point.lat, uuid.point.lon, uuid.dir, uuid.speed, uuid.utc)
                }).iterator*/
          result.iterator
        })
    
        val pro = new Properties
        //必须增加数据库的用户名(user)密码(password),必须指定postgresql驱动(driver)
        pro.put("user", "postgres")
        pro.put("password", "password")
        pro.put("driver", "org.postgresql.Driver")
        rdds.toDF()
          .select("uuid", "point.lat", "point.lon", "dir", "speed", "utc")
          .write.mode("append")
          .jdbc("jdbc:postgresql://192.168.00.00:38888/bigdata", s"cover_${utils.CurrentTimeUtil.getFrontHour().toString}", pro)
        sc.stop()
        spark.stop()
      }
    }
     /*
            val unit = rdd.foreachPartition(partition => {
              val poolForBigdata = CreatePGSqlPoolForBigdata()
              val clientForBigdata = poolForBigdata.borrowObject()
              partition.foreach(line => {
                val sql = s"INSERT INTO cover_${utils.CurrentTimeUtil.getFrontHour().toString} VALUES(?,?,?,?,?,?)" //uuid,roadid,roadpart,point,dir,speed,utc
                clientForBigdata.executeUpdate(sql, Array(line.uuid, line.point.lat, line.point.lon, line.dir, line.speed, line.utc)) //
              })
              //使用完成后将对象返回给对象池
              poolForBigdata.returnObject(clientForBigdata)
            })*/

参考：[https://blog.csdn.net/dream\_an/article/details/54962464][https_blog.csdn.net_dream_an_article_details_54962464]

https://blog.csdn.net/dai451954706/article/details/52840011

[https_blog.csdn.net_dream_an_article_details_54962464]: https://blog.csdn.net/dream_an/article/details/54962464