码迷,mamicode.com
首页 > 其他好文 > 详细

创建DataFrame的两个途径

时间:2017-05-02 12:02:19      阅读:811      评论:0      收藏:0      [点我收藏+]

标签:product   href   否则   tab   官方   服务   dataframe   修改   sql   

转自: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)

 

扩展阅读:

由于数据格式和数据源众多,这里暂不一一展开讲解。在实际应用中,如果需要使用某种格式的数据或者某个数据源,应查询其官方文档。通常官方文档(特别是 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")

 

创建DataFrame的两个途径

标签:product   href   否则   tab   官方   服务   dataframe   修改   sql   

原文地址:http://www.cnblogs.com/libin2015/p/6782667.html

(0)
(0)
   
举报
评论 一句话评论(0
登录后才能评论!
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码!