创建DataFrame的两个途径

Posted 2020-09-15 AI晓

转自：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：从数据库表中加载 DataFrame
• sqlContext.jsonFile：从 JSON 文件中加载 DataFrame
• sqlContext.jsonRDD：从包含 JSON 对象的 RDD 中加载 DataFrame
• sqlContext.parquetFile：从 parquet 文件中加载 DataFrame

需要注意的是，在 Spark 1.4 及之后的版本中，加载数据源的方法为：

// 默认格式parquet文件的加载方法，需要给出文件的路径
sqlContext.read.load("/home/shiyanlou/data.parquet")

// 加载其他格式的文件，需要在format方法中指明格式
sqlContext.read.format("json").load("/home/shiyanlou/data.json")