如何使用apache spark在mysql数据库中创建表

Question

我正在尝试创建一个对 创建、读取、写入和更新 MySQL 数据。那么，有没有办法使用 Spark 创建 MySQL 表呢？

下面我有一个在 MySQL 中创建表的 Scala-JDBC 代码 数据库。如何通过 Spark 做到这一点？

package SparkMysqlJdbcConnectivity

import org.apache.spark.sql.SparkSession
import java.util.Properties
import java.lang.Class
import java.sql.Connection
import java.sql.DriverManager

object MysqlSparkJdbcProgram {

  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("MysqlJDBC Connections")
      .master("local[*]")
      .getOrCreate()

    val driver = "com.mysql.jdbc.Driver"
    val url = "jdbc:mysql://localhost:3306/world"
    val operationtype = "create table"
    val tablename = "country"
    val tablename2 = "state"

    val connectionProperties = new Properties()

    connectionProperties.put("user", "root")
    connectionProperties.put("password", "root")

    val jdbcDf = spark.read.jdbc(url, s"${tablename}", connectionProperties)

    operationtype.trim() match {
      case "create table" => {
       // Class.forName(driver)
        try{
          val con:Connection = DriverManager.getConnection(url,connectionProperties)
          val result = con.prepareStatement(s"create table ${tablename2} (name varchar(255), country varchar(255))").execute()
          println(result)
          if(result) println("table creation is unsucessful") else println("table creation is unsucessful")
        }
      }

      case "read table" => {

        val jdbcDf = spark.read.jdbc("jdbc:mysql://localhost:3306/world", s"${tablename}", connectionProperties)
        jdbcDf.show()
      }

      case "write table" => {}

      case "drop table"  => {}

    }

  }

}

Answer 1

当您编写 jdbcDf 数据帧时，这些表将自动创建。

jdbcDf
 .write
 .jdbc("jdbc:mysql://localhost:3306/world", s"${tablename}", connectionProperties)

如果你想指定表模式，

jdbcDf
 .write
 .option("createTableColumnTypes", "name VARCHAR(500), col1 VARCHAR(1024), col3 int")
 .jdbc("jdbc:mysql://localhost:3306/world", s"${tablename}", connectionProperties)