package cn.spark.study.sy5

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.sql.types.{StringType, StructField, StructType}


object Testsql {
  def main(args: Array[String]) {
       val conf = new SparkConf()
       conf.setMaster("local")
           .setAppName("Testsql")
       val sc = new SparkContext(conf)
       val sqlContext = new SQLContext(sc)
       //hdfs://192.168.6.134:9000/nlc/1.txt
       //H:\文件\数据集
       val studentRDD = sc.textFile("D:\\myDevelopTools\\Intellij IDEA\\workplace\\spark-study-scala\\src\\main\\java\\cn\\spark\\study\\sy5\\employee.txt", 1)
      .map { line => Row(line.split(",")(0), line.split(",")(1), line.split(",")(2)) }
        // 第二步，编程方式动态构造元数据
      val structType = StructType(Array(
          StructField("id", StringType, true),
          StructField("name", StringType, true),
          StructField("age", StringType, true)))
      // 第三步，进行RDD到DataFrame的转换
      val studentDF = sqlContext.createDataFrame(studentRDD, structType)
      // 继续正常使用
      studentDF.registerTempTable("employee")
//      val teenagerDF = sqlContext.sql("select usrid,count(usrid) from students group by usrid order by usrid")
      val teenagerDF = sqlContext.sql("select id,name,age from employee")
      val teenagerRDD = teenagerDF.rdd.collect().foreach { row => println("id:"+row(0)+",name:"+row(1)+",age:"+row(2)) }
    }
}

3. 编程实现利用 DataFrame 读写 MySQL 的数据
（1）在 MySQL 数据库中新建数据库 sparktest，再创建表 employee，包含如表 6-2 所示的
两行数据。
表 6-2 employee 表原有数据
id name gender Age
1 Alice F 22
2 John M 25
（2）配置 Spark 通过 JDBC 连接数据库 MySQL，编程实现利用 DataFrame 插入如表 6-3 所
示的两行数据到 MySQL 中，最后打印出 age 的最大值和 age 的总和。
表 6-3 employee 表新增数据
id name gender age
3 Mary F 　　26
4 Tom M 　　23

package cn.spark.study.sy5
import java.util.Properties

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
/**
  * Created by Lenovo on 2019/3/27.
  */
object TestMySQL {
  def main(args: Array[String]) {
    val conf = new SparkConf()
    conf.setMaster("local")
      .setAppName("Testsql")
    val sc = new SparkContext(conf)
    val sqlContext = new SQLContext(sc)

    val employeeRDD = sqlContext.sparkContext.parallelize(Array("3 Mary F 26","4 Tom M 23")).map(_.split(" "))
    val schema = StructType(List(StructField("id", IntegerType,
      true),StructField("name", StringType, true),StructField("gender", StringType,
      true),StructField("age", IntegerType, true)))
    val rowRDD = employeeRDD.map(p => Row(p(0).toInt,p(1),
      p(2),p(3).toInt))
    val employeeDF = sqlContext.createDataFrame(rowRDD, schema)
    val prop = new Properties()
    prop.put("user", "root")
    prop.put("password", "123123")
    prop.put("driver","com.mysql.jdbc.Driver")
    employeeDF.write.mode("append").jdbc("jdbc:mysql://localhost:3306/sparktest", "sparktest.employee", prop)
      val jdbcDF = sqlContext.read.format("jdbc").option("url",
      "jdbc:mysql://localhost:3306/sparktest").option("driver","com.mysql.jdbc.Driver").option("dbtable","employee").option("user","root").option("password", "123123").load()
      jdbcDF.agg("age" -> "max", "age" -> "sum")
      }

}

赞赏

长按二维码向我转账

受苹果公司新规定影响，微信 iOS 版的赞赏功能被关闭，可通过二维码转账支持公众号。