标签:product href 否则 tab 官方 服务 dataframe 修改 sql
https://www.shiyanlou.com/courses/536/labs/1818/document
方法一 由反射机制推断出模式:
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 支持 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")
标签:product href 否则 tab 官方 服务 dataframe 修改 sql
原文地址:http://www.cnblogs.com/libin2015/p/6782667.html