1. 读取 mysql

  1. val jdbcDF = spark.read
  2.   .format("jdbc")
  3.   .option("url", "jdbc:postgresql:dbserver")
  4.   .option("dbtable", "schema.tablename")
  5.   .option("user", "username")
  6.   .option("password", "password")
  7.   .option("driver", "com.mysql.jdbc.Driver")
  8.   .load()

2. 写入 mysql

  1. jdbcDF.write
  2.   .format("jdbc")
  3.   .option("url", "jdbc:postgresql:dbserver")
  4.   .option("dbtable", "schema.tablename")
  5.   .option("user", "username")
  6.   .option("password", "password")
  7.   .option("driver", "com.mysql.jdbc.Driver")
  8.   .save()
  1. trait Base {
  3.   val appName: String
  5.   lazy val spark = SparkSession
  6.     .builder()
  7.     .appName(appName)
  8.     .enableHiveSupport()
  9.     .getOrCreate()
  11.   def run(): Unit
  13.   def fetchFromMysql(dbtable: String): DataFrame = {
  14.     spark.read
  15.       .format("jdbc")
  16.       .option("url", MysqlUtil.url)
  17.       .option("dbtable", dbtable)
  18.       .option("user", MysqlUtil.user)
  19.       .option("password", MysqlUtil.password)
  20.       .option("driver", "com.mysql.jdbc.Driver")
  21.       .load()
  22.   }
  24.   def loadIntoMysql(df: DataFrame, dbtable: String): Unit = {
  25.     //overwrite模式会删除整个表!!!
  26.     df.write
  27.       .format("jdbc")
  28.       .mode("append")
  29.       .option("url", MysqlUtil.url)
  30.       .option("dbtable", dbtable)
  31.       .option("user", MysqlUtil.user)
  32.       .option("password", MysqlUtil.password)
  33.       .option("driver", "com.mysql.jdbc.Driver")
  34.       .save()
  35.   }
  37.   def insertIntoMysql(df: DataFrame, dao: DaoBase): Unit = {
  38.     //使用原始jdbc连接msql,防止重复插入
  39.     df.coalesce(1).foreachPartition { partitionOfRecords =>
  40.       val ps = MysqlUtil.getConn.prepareStatement(dao.sqlText)
  41.       partitionOfRecords.foreach { row =>
  42.         dao.fun(row, ps)
  43.         ps.executeUpdate()
  44.       }
  45.     }
  46.   }
  48.   def writeToHive(df: DataFrame, tableName: String): Unit = {
  49.     // 分区表和非分区表都可以写入
  50.     // df的字段顺序要求和tableName的顺序完全一致,包括分区字段
  51.     spark.sql("set hive.exec.dynamic.partition=true")
  52.     spark.sql("set hive.exec.dynamic.partition.mode=nonstrict")
  53.     df.coalesce(1)
  54.       .write
  55.       .mode("overwrite")
  56.       .format("hive")
  57.       .insertInto(tableName)
  58.   }
  59. }
  62. trait DaoBase {
  63.   /**
  64.    * insert … select … where not exist
  65.    * 防止重复插入
  66.    */
  67.   val sqlText: String
  69.   /**
  70.    * 从df的row中提取插入mysql表实体信息,填充ps
  71.    *
  72.    * @param row
  73.    * @param ps
  74.    */
  75.   def fun(row: Row, ps: PreparedStatement): Unit
  76. }
  80. object carInfoDao extends DaoBase {
  82.   val sqlText =
  83.     """
  84.       |INSERT INTO car_info (name
  85.       |      ) SELECT
  86.       |    ?
  87.       |FROM
  88.       |    DUAL
  89.       |WHERE
  90.       |    NOT EXISTS (
  91.       |        SELECT
  92.       |            id
  93.       |        FROM
  94.       |            car_info
  95.       |        WHERE
  96.       |            id = ?
  97.       |    )
  98.       |""".stripMargin
  100.   def fun(row: Row, ps: PreparedStatement): Unit = {
  101.     val entity = rowToEntity(row)
  102.     ps.setString(1, entity.name)
  104.   }
  106.   def rowToEntity(row: Row): Entity = {
  107.     Entity("ok")
  108.   }
  110. }

3. spark 官网资料

JDBC To Other Databases