本篇介绍Spark-SQL之DataFrame的创建，正如学习永无止境，本篇也将不断的迭代更新。
Spark SQL中的DataFrame类似于一张关系型数据表。在关系型数据库中对单表或进行的查询操作，在DataFrame中都可以通过调用其API接口来实现。可以参考，Scala提供的DataFrame API。本篇的Spark API基于spark 2.1版本。
1、DataFrame对象的生成
DataFrame(以下简称df)的生成方式有很多,我们一一道来，不过在生成之前，我们首先要创建一个SparkSession：
val spark = SparkSession
.builder()
.appName("Spark SQL basic example")
.enableHiveSupport()
//.config("spark.some.config.option", "some-value")
.getOrCreate()
使用toDF方法
通过导入(importing)Spark sql implicits, 就可以将本地序列(seq), 列表或者RDD转为DataFrame。只要这些数据的内容能指定数据类型即可。
比如，下面的代码,将本地序列转化为df：
import spark.implicits._
val df = Seq(
(1, "First Value", java.sql.Date.valueOf("2010-01-01")),
(2, "Second Value", java.sql.Date.valueOf("2010-02-01"))
).toDF("int_column", "string_column", "date_column")
df.show()
输出结果为:
+----------+-------------+-----------+
|int_column|string_column|date_column|
+----------+-------------+-----------+
| 1| First Value| 2010-01-01|
| 2| Second Value| 2010-02-01|
+----------+-------------+-----------+
同样，我们可以将一个RDD转化为df：
val rdd = spark.sparkContext.parallelize(List(1,2,3,4,5))
val df = rdd.map(x=>(x,x^2)).toDF("org","xor")
df.show()
输出为：
+---+---+
|org|xor|
+---+---+
| 1| 3|
| 2| 0|
| 3| 1|
| 4| 6|
| 5| 7|
+---+---+
当然，我们已可以将一个scala的列表转化为df：
val arr = List((1,3),(2,4),(3,5))
val df = arr.toDF("first","second")
df.show()
输出结果为:
+-----+------+
|first|second|
+-----+------+
| 1| 3|
| 2| 4|
| 3| 5|
+-----+------+
使用createDataFrame方法
在SqlContext中使用createDataFrame也可以创建DataFrame。跟toDF一样，这里创建DataFrame的数据形态也可以是本地数组或者RDD。
我们也可以通过row+schema创建DataFrame：
import org.apache.spark.sql.types._
val schema = StructType(List(
StructField("integer_column", IntegerType, nullable = false),
StructField("string_column", StringType, nullable = true),
StructField("date_column", DateType, nullable = true)
))
val rdd = spark.sparkContext.parallelize(Seq(
Row(1, "First Value", java.sql.Date.valueOf("2010-01-01")),
Row(2, "Second Value", java.sql.Date.valueOf("2010-02-01"))
))
val df = spark.createDataFrame(rdd, schema)
df.show()
输出为:
+--------------+-------------+-----------+
|integer_column|string_column|date_column|
+--------------+-------------+-----------+
| 1| First Value| 2010-01-01|
| 2| Second Value| 2010-02-01|
+--------------+-------------+-----------+
通过文件直接创建DataFrame
我们介绍几种常见的通过文件创建DataFrame。
json
json文件内容如下：
val df = spark.read.json("src/main/scala/a/test.json")
df.show()
输出为:
+-----+---------+----------+
|email|firstName| lastName|
+-----+---------+----------+
| aaaa| Brett|McLaughlin|
| bbbb| Jason| Hunter|
| cccc| Elliotte| Harold|
+-----+---------+----------+
csv
dataFrame可以通过csv文件来创建：
val df = spark.read.format("com.databricks.spark.csv.DefaultSource15")
.option("header", true)
.option("delimiter", ",")
.load("src/main/scala/a/t.csv");
mysql
mysql数据库的内容如下：
代码为：
val url = "jdbc:mysql://localhost:3306/test"
val df = spark.read
.format("jdbc")
.option("url", url)
.option("dbtable", "pivot")
.option("user", "root")
.option("password", "admin")
.load()
df.show()
代码输出为:
+---+----+----+
| id|user|type|
+---+----+----+
| 1| 1| 助手1|
| 2| 1|APP1|
| 3| 2| 助手1|
| 4| 2| 助手1|
| 5| 3|APP1|
| 6| 3|APP1|
| 7| 3| 助手2|
| 8| 3|APP2|
| 9| 2|APP2|
| 10| 2| 助手1|
| 11| 1|APP1|
| 12| 1| 助手2|
+---+----+----+
hive
val sqlDF = spark.sql("SELECT * FROM people")
sqlDF.show()
// +----+-------+
// | age| name|
// +----+-------+
// |null|Michael|
// | 30| Andy|
// | 19| Justin|
// +----+-------+
