学习记录15

　　本次学习学习了将dataframe里吗有结构的数据加载到mysql以及进行读

这里采用独立应用程序的方式读取MySQL数据库内容。创建一个代码文件SparkReadMySQL.scala，其内容如下：

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
object SparkReadMySQL {
  def main(args: Array[String]): Unit ={
    Logger.getLogger("org").setLevel(Level.ERROR)
    val spark = SparkSession.builder().appName("SparkReadMySQL").getOrCreate()
    val df = spark.read
      .format("jdbc")
      .option("url", "jdbc:mysql://localhost:3306/spark")
      .option("driver", "com.mysql.jdbc.Driver")
      .option("dbtable", "student")
      .option("user", "root")
      .option("password", "123456")
      .load()
df.show()
spark.stop()
  }
}

对代码进行编译打包，然后执行如下命令运行程序：

$ /usr/local/spark/bin/spark-submit \
> --jars  \
> /usr/local/spark/jars/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar \
> --class "SparkReadMySQL" \
> /home/hadoop/sparkapp/target/scala-2.12/simple-project_2.12-1.0.jar

写入MySQL

这里采用独立应用程序的方式把数据写入MySQL数据库。创建一个代码文件SparkWriteMySQL.scala，其内容如下：

import java.util.Properties
import org.apache.spark.sql.types._
import org.apache.spark.sql.Row
import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession

object SparkWriteMySQL {
  def main(args: Array[String]): Unit ={
    Logger.getLogger("org").setLevel(Level.ERROR)
    val spark = SparkSession.builder().appName("SparkWriteMySQL").getOrCreate()
    //下面我们设置两条数据表示两个学生信息
    val studentRDD = spark.sparkContext.parallelize(Array("3 Rongcheng M 26","4 Guanhua M 27")).map(_.split(" "))
//下面要设置模式信息
    val schema = StructType(List(StructField("id", IntegerType, true),StructField("name", StringType, true),StructField("gender", StringType, true),StructField("age", IntegerType, true)))
 
    //下面创建Row对象，每个Row对象都是rowRDD中的一行
    val rowRDD = studentRDD.map(p => Row(p(0).toInt, p(1).trim, p(2).trim, p(3).toInt))
 
    //建立起Row对象和模式之间的对应关系，也就是把数据和模式对应起来
    val studentDF = spark.createDataFrame(rowRDD, schema)
 
    //下面创建一个prop变量用来保存JDBC连接参数
    val prop = new Properties()
    prop.put("user", "root") //表示用户名是root
    prop.put("password", "123456") //表示密码是123456
    prop.put("driver","com.mysql.jdbc.Driver") //表示驱动程序是com.mysql.jdbc.Driver

    //下面就可以连接数据库，采用append模式，表示追加记录到数据库spark的student表中
    studentDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/spark", "spark.student", prop)
    spark.stop()
  }
}

对代码进行编译打包，然后执行如下命令运行程序：

$ /usr/local/spark/bin/spark-submit \
> --jars  \
> /usr/local/spark/jars/mysql-connector-java-5.1.40/mysql-connector-java-5.1.40-bin.jar \
> --class "SparkWriteMySQL" \
> /home/hadoop/sparkapp/target/scala-2.12/simple-project_2.12-1.0.jar