当前位置: 首页 > >

理解 Spark DataFrame

发布时间:

DataFrame它不是spark sql提出来的,而是早期在R、Pandas语言就已经有了的
DataSet: A DataSet is a distributed collection of data. (分布式的数据集)
DataFrame:A DataFrame is a DataSet organized into named columns.
以列(列名,列类型,列值)的形式构成的分布式的数据集,按照列赋予不同的名称


?


student
id:int
name:string
city:string

概念

1)分布式的数据集,并且以列的方式组合的。相当于具有schema的RDD
2)相当于关系型数据库中的表,但是底层有优化
3)提供了一些抽象的操作,如select、filter、aggregation、plot
4)它是由于R语言或者Pandas语言处理小数据集的经验应用到处理分布式大数据集上
5)在1.3版本之前,叫SchemaRDD


DataFrame对比RDD

1)DataFrame不仅仅分布式数据集,并且有schema描述(按列存储,有对应列的描述)



?


2)运行环境


RDD:
java/scala ==> JVM
python ==> python runtime
DataFrame
java/scala/python ==> Logic Plan (最终性能一样)


3)三者的特点如下:
rdd的优点:


1.强大,内置很多函数操作,group,map,filter等,方便处理结构化或非结构化数据
2.面向对象编程,直接存储的java对象,类型转化也安全
rdd的缺点:
1.由于它基本和hadoop一样万能的,因此没有针对特殊场景的优化,比如对于结构化数据处理相对于sql来比非常麻烦
2.默认采用的是java序列号方式,序列化结果比较大,而且数据存储在java堆内存中,导致gc比较频繁


dataframe的优点:


1.结构化数据处理非常方便,支持Avro, CSV, elastic search, and Cassandra等kv数据,也支持HIVE tables, MySQL等传统数据表
2.有针对性的优化,由于数据结构元信息spark已经保存,序列化时不需要带上元信息,大大的减少了序列化大小,而且数据保存在堆外内存中,减少了gc次数。
3.hive兼容,支持hql,udf等
dataframe的缺点:
1.编译时不能类型转化安全检查,运行时才能确定是否有问题
2.对于对象支持不友好,rdd内部数据直接以java对象存储,dataframe内存存储的是row对象而不能是自定义对象


dataset的优点:


1.dataset整合了rdd和dataframe的优点,支持结构化和非结构化数据
2.和rdd一样,支持自定义对象存储
3.和dataframe一样,支持结构化数据的sql查询
4.采用堆外内存存储,gc友好
5.类型转化安全,代码友好
6.官方建议使用dataset


?


4)实例

首先我们在本地放置了一个JSON文件,文件内容如下:


{"name":"Michael"}
{"name":"Andy", "age":30}
{"name":"Justin", "age":19}

然后我们进入spark-shell,控制台的提示说明Spark为我们创建了一个叫sqlContext的上下文,注意,它是DataFrame的起点。
接下来我们希望把本地的JSON文件转化为DataFrame


scala> val df = sqlContext.jsonFile("/path/to/your/jsonfile")
df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]

从控制台的提示可以得知,我们成功创建了一个DataFrame的对象,包含agename两个字段。
DataFrame自带的玩法就多了:


// 输出表结构
df.printSchema()

// 选择所有年龄大于21岁的人,只保留name字段
df.filter(df("age") > 21).select("name").show()

// 选择name,并把age字段自增
df.select("name", df("age") + 1).show()

// 按年龄分组计数
df.groupBy("age").count().show()

// 左联表(注意是3个等号!)
df.join(df2, df("name") === df2("name"), "left").show()

此外,我们也可以把DataFrame对象转化为一个虚拟的表,然后用SQL语句查询,比如下面的命令就等同于df.groupBy("age").count().show()


df.registerTempTable("people")
sqlContext.sql("select age, count(*) from people group by age").show()

当然,Python有同样丰富的API(由于最终都是转化为JVM bytecode执行,Python和Scala的效率是一样的),而且Python还提供了类Pandas的操作语法。关于Python的API,可以参考Spark新年福音:一个用于大规模数据科学的API??DataFrame。


MySQL

除了JSON之外,DataFrame现在已经能支持MySQL、Hive、HDFS、PostgreSQL等外部数据源,而对关系数据库的读取,是通过jdbc实现的。


对于不同的关系数据库,必须在SPARK_CLASSPATH变量中加入对应connector的jar包,比如希望连接MySQL的话应该这么启动spark-shell


SPARK_CLASSPATH=mysql-connector-java-x.x.x-bin.jar spark-shell

下面要将一个MySQL表转化为DataFrame对象:


val jdbcDF = sqlContext.load("jdbc", Map("url" -> "jdbc:mysql://localhost:3306/your_database?user=your_user&password=your_password", "dbtable" -> "your_table"))

然后十八般武艺又可以派上用场了。


Hive

Spark提供了一个HiveContext的上下文,其实是SQLContext的一个子类,但从作用上来说,sqlContext也支持Hive数据源。只要在部署Spark的时候加入Hive选项,并把已有的hive-site.xml文件挪到$SPARK_HOME/conf路径下,我们就可以直接用Spark查询包含已有元数据的Hive表了:


?


sqlContext.sql("select count(*) from hive_people").show()



友情链接: