第4章 Spark SQL结构化数据文件处理

本章目的在于如何使用Spark SQL模块来处理结构化数据，结构化数据即以关系型数据库表形式管理的数据。

首先，本章讲解在Linux系统下对结构化数据的处理。Spark SQL模块让用户可以通过SQL、DataFrame API 和 **DataSet API **三种方式来实现对结构化数据的处理。
在Spark-Shell操作下，RDD很容易就能转换成为DataFrame。那么在Windows系统下呢？又该如何使RDD转换成为DataFrame？

启动Spark-Shell的命令如下：$ spark-shell --master local[2]；如不能正常运行，可跳转到spark安装目录，使用命令 $ bin/spark-shell 启动Spark-shell。

文件：no4-createDataFrame.txt，该文件用于本章例子

# 将要用到的文件no4-createDataFrame.txt
[root@hadoop01 ~]# hdfs dfs -cat /sparktest/data/no4-createDataFrame.txt
zhangsan 20
lisi 29
wangwu 25
zhaoliu 30
tianqi 35
jerry 40

DataFrame

DataFrame的创建

创建DataFrame的方式是从一个已经存在的RDD调用toDF()方法进行转换，得到DataFrame；或者通过Spark读取数据源直接创建。
这里提供两种成员函数printSchema()和show()，作用分别为打印当前对象的Schema元数据信息和结果数据。

创建DataFrame的两种方式及部分成员函数

# 通过文件直接创建DataFrame。除了读取text(.txt)文件，还可以读取csv、json、parquet等 | 创建方式1
scala> val personDF=spark.read.text("/sparktest/data/no4-createDataFrame.txt")
personDF: org.apache.spark.sql.DataFrame = [value: string]

# 打印当前对象的Schema元数据信息：String数据类型，且可为空。
scala> personDF.printSchema()
root
 |-- value: string (nullable = true)

# 打印当前DataFrame的结果数据
scala> personDF.show()
+-----------+       
|      value    |
+-----------+
|zhangsan 20|
|    lisi 29|
|  wangwu 25|
| zhaoliu 30|
|  tianqi 35|
|   jerry 40|
+-----------+


# 从已经存在的RDD进行转换得到DataFrame，首先获取数据
# 第一步
scala> case class Person(name:String,age:Int)
defined class Person
# 以上三步可直接写成：
scala> val pDF=sc.textFile("/sparktest/data/no4-createDataFrame.txt").map(_.split(" ")).map(x => Person(x(0),x(1).toInt)).toDF()
scala> pDF.show
+--------+---+      
|    name|age|
+--------+---+
|zhangsan| 20|
|    lisi| 29|
|  wangwu| 25|
| zhaoliu| 30|
|  tianqi| 35|
|   jerry| 40|
+--------+---+

DataFrame的常用操作

DataFrame提供了两种语法风格，分别为DSL风格操作和SQL风格操作。对应的就是DataFrame API 、SQL两种方式。
DSL风格

#查看name 字段的数据，PersonDF会随变量名的变化而变化。
#以下五种方式可同义替代。但第四种方法不推荐！原因是因为某些代码可能不支持！
pDF.select(pDF.col("name"),pDF.col("age")).show
pDF.select(pDF("name"),pDF("age")).show
pDF.select(col("name"),col("age")).show
pDF.select("name","age").show
# pDF.select("name","age"+1).show 不可运行！
# 但其他四个可运行！例如“$"age"+1”、“col("age")+1”
# 请注意可以通过as来重命名列名
pDF.select($"name",($"age"+1).as("new_age")).show

#条件过滤（对应Where语句）
pDF.filter($"age" >30).show

#分组（对应Group By语句）
pDF.groupBy("age").count().show

#排序（对应Order By语句）
pDF.sort($"age".desc).show

SQL风格操作

将DataFrame注册成一个临时表就可以进行SQL风格操作。

1 2	scala> pDF.registerTempTable("t_koinl") scala> spark.sql("select name, age + 1 from t_koinl").show

Dataset

RDD转换为DataFrame

在上文中，有说明在Spark-shell中如何将RDD转换为DataFrame，在本小节中，来说明如何在Windows系统下开发Scala代码。

一般情况下，可以使用两种方法来实现。第一种方法是利用反射机制来推断包含特定类型对象的Schema。当case类不能提前定义，即未知数据结构时，应通过编程接口构造一个Schema，并将其应用在已知的RDD数据中。

Spark SQL操作数据源

操作MySQL

读取MySQL种数据

import java.util.Properties
import org.apache.spark.sql.{DataFrame, SaveMode,SparkSession}
 
object sparkSqlMysql {
  def main(args: Array[String]): Unit = {
    //创建sparkSession对象
    val spark: SparkSession = SparkSession.builder()
      .appName("sparkSqlMysql")
      .master("local[2]")
      .getOrCreate()
    //创建Properties对象，配置连接mysql的用户名和密码
    val prop: prop =new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","123456")
    //从数据库里读取数据
    val mysqlDF: DataFrame = spark.read.jdbc("jdbc:mysql://127.0.0.1:3306/spark", "person", prop)
    //显示Mysql中表数据
    mysqlDF.show()
    spark.stop()
  }
}

写入数据到MySQL

import java.util.Properties
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, SaveMode,SparkSession}
 case class Person(id:Int,name:String,age:Int)

object sparkSqlMysql {
  def main(args: Array[String]): Unit = {
    //创建sparkSession对象
    val spark: SparkSession = SparkSession.builder()
      .appName("sparkSqlMysql")
      .master("local[2]")
      .getOrCreate()
    val sc = spark.sparkContext.parallelize(Array("3,wangwu,22","4,zhaoliu,26"))
    //切分读取数据
    val data: RDD[Array[String]] = sc.map(_.split(","))
    //RDD关联Person
    val personRdd: RDD[Person] = data.map(x => Person(x(0)。toInt,x(1), x(2).toLong))
    //导入隐式转换
    import spark.implicits._
    //将RDD转换成DataFrame
    val personDF: DataFrame = personRdd.toDF()
    //创建Properties对象，配置连接mysql的用户名和密码
    val prop =new Properties()
    prop.setProperty("user","root")
    prop.setProperty("password","123456")
    //将personDF写入MySQL
 
    personDF.write.mode(SaveMode.Append).jdbc("jdbc:mysql://127.0.0.1:3306/spark?useUnicode=true&characterEncoding=utf8","person",prop)
    personDF.show()
    spark.stop()
  }
}

站点信息