转自：https://www.shiyanlou.com/courses/543/labs/1835/document
https://www.shiyanlou.com/courses/536/labs/1818/document
一、从 RDD 创建 DataFrame：
方法一 由反射机制推断出模式：
Step 1：引用必要的类。
import org.apache.spark.sql._
import sqlContext.implicits._ //idea中此处导入应在sqlContext 创建之后，否则报错，不知道为什么。。？？
// 在使用Spark Shell时，下面这句不是必需的。
// Spark Shell已默认为你启用了SQL context，在需要时可直接使用sqlContext。
val sqlContext = new org.apache.spark.sql.SQLContext(sc)
Step 2：创建RDD。
//导入CSV文件并处理逗号分隔的数据
val sfpdRDD = sc.textFile("/home/shiyanlou/SFPD.csv").map(inc => inc.split(","))
Step 3：定义 case class 。
case class Incidents(incidentnum:String, category:String, description:String, dayofweek:String, date:String, time:String, pddistrict:String, resolution:String, address:String, x:String, y:String, location:String, pdid:String)
Step 4：将 RDD 转换为含有 case 对象的 RDD 。
val sfpdCase = sfpdRDD.map(inc => Incidents(inc(0), inc(1), inc(2), inc(3), inc(4), inc(5), inc(6), inc(7), inc(8), inc(9), inc(10), inc(11), inc(12)))
Step 5：隐式转换会将含有 case 对象的 RDD 转换为 DataFrame ，将 DataFrame 的一些操作和函数应用于这个 DataFrame 中。
val sfpdDF = sfpdCase.toDF()
方法二 通过编程方式构建模式：
这种方式适用于列和类型在运行时不可知的情况，我们就需要手动地去构建 DataFrame 的模式。通常 DataFrame 的模式在动态变化时才会使用这种方式。
注意：该方式在 case class 在不能被提前定义时，或者使用 Scala 语言的项目中 case class 超过22个字段时，才会用到。
Step 1：引入必要的类。
import sqlContext.implicits._
import org.apache.spark.sql._
import org.apache.spark.sql.types._
Step 2：由原始 RDD 创建一个 Row RDD 。
val rowRDD = sc.textFile("/home/shiyanlou/data.txt").map(x => x.split(" ")).map( p => Row(p(0), p(2), p(4)))
Step 3：使用 StructType 和 StructField 分别创建模式。其中， StructType 对应于 table （表），StructField 对应于 field （字段）。
val testSchema = StructType(Array(StructField("IncNum", StringType, true), StructField("Date", StringType, true), StructField("District", StringType, true)))
Step 4：使用 SQLContext 提供的方法，将模式应用于 Row RDD 上，以创建 DataFrame。
val testDF = sqlContext.createDataFrame(rowRDD, testSchema)
// 将DataFrame注册为表
testDF.registerTempTable("test")
val incs = sql("SELECT * FROM test")
二、从数据源创建 DataFrame：
现有的大数据应用通常需要搜集和分析来自不同的数据源的数据。而 DataFrame 支持 JSON 文件、 Parquet 文件、 Hive 表等数据格式。它能从本地文件系统、分布式文件系统（HDFS）、云存储（Amazon S3）和外部的关系数据库系统（通过JDBC，在Spark 1.4版本起开始支持）等地方读取数据。另外，通过 Spark SQL 的外部数据源 API ，DataFrame 能够被扩展，以支持第三方的数据格式或数据源。
csv：
主要是 com.databricks_spark-csv_2.11-1.1.0 这个库，用于支持 CSV 格式文件的读取和操作。
step 1：
在终端中输入命令： wget http://labfile.oss.aliyuncs.com/courses/610/spark_csv.tar.gz 下载相关的 jar 包。
将该压缩文件解压至 /home/shiyanlou/.ivy2/jars/ 目录中，确保该目录含有如图所示的以下三个 jar 包。
step 2 导入包：
spark-shell --packages com.databricks:spark-csv_2.11:1.1.0
step 3 直接将 CSV 文件读入为 DataFrame ：
val df = sqlContext.read.format("com.databricks.spark.csv").option("header", "true").load("/home/shiyanlou/1987.csv")
// 此处的文件路径请根据实际情况修改
step 4 根据需要修改字段类型：
def convertColumn(df: org.apache.spark.sql.DataFrame, name:String, newType:String) = {
val df_1 = df.withColumnRenamed(name, "swap")
df_1.withColumn(name, df_1.col("swap").cast(newType)).drop("swap")
}
//例如
val df_3 = convertColumn(df_2, "ArrDelay", "int")
val df_4 = convertColumn(df_2, "DepDelay", "int")
json：
sqlContext.read.json(filePath)
扩展阅读：
CSV：逗号分隔值（ Comma-Separated Values ），其文件以纯文本形式存储表格数据。请参考：http://commons.apache.org/proper/commons-csv/ 。Apache Avro： 一个数据序列化的系统，相当于基于二进制数据传输高性能的中间件。请参考：https://avro.apache.org/docs/current/ 。Elasticsearch：一个基于 Lucene 的搜索服务器。提供了一个基于 RESTful web 接口的分布式全文搜索引擎。请参考：https://www.elastic.co/products/elasticsearch/ 。Apache Cassandra：一套开源分布式NoSQL数据库系统。请参考：http://cassandra.apache.org/ 。
由于数据格式和数据源众多，这里暂不一一展开讲解。在实际应用中，如果需要使用某种格式的数据或者某个数据源，应查询其官方文档。通常官方文档（特别是 API 手册）都提供了详细的集成方法和指导。
在 Spark 中，默认的数据源被设定为 Parquet ，所以通用的加载方式为：
sqlContext.load("/home/shiyanlou/data.parquet")
如果是其他格式，则需要手动地指定格式：
sqlContext.load("/home/shiyanlou/data", "json")
下面给出了其他的加载指定数据源的方法：
sqlContext.jdbc：从数据库表中加载 DataFramesqlContext.jsonFile：从 JSON 文件中加载 DataFramesqlContext.jsonRDD：从包含 JSON 对象的 RDD 中加载 DataFramesqlContext.parquetFile：从 parquet 文件中加载 DataFrame
需要注意的是，在 Spark 1.4 及之后的版本中，加载数据源的方法为：
// 默认格式parquet文件的加载方法，需要给出文件的路径
sqlContext.read.load("/home/shiyanlou/data.parquet")
// 加载其他格式的文件，需要在format方法中指明格式
sqlContext.read.format("json").load("/home/shiyanlou/data.json")