Spark DataFrame小试牛刀

三月中旬,Spark发布了最新的1.3.0版本,其中最重要的变化,便是DataFrame这个API的推出。DataFrame让Spark具备了处理大规模结构化数据的能力,在比原有的RDD转化方式易用的前提下,计算性能更还快了两倍。这一个小小的API,隐含着Spark希望大一统「大数据江湖」的野心和决心。DataFrame像是一条联结所有主流数据源并自动转化为可并行处理格式的水渠,通过它Spark能取悦大数据生态链上的所有玩家,无论是善用R的数据科学家,惯用SQL的商业分析师,还是在意效率和实时性的统计工程师。

以一个常见的场景 -- 日志解析为例,有时我们需要用到一些额外的结构化数据(比如做IP和地址的映射),通常这样的数据会存在MySQL,而访问的方式有两种:一是每个worker远程去检索数据库,弊端是耗费额外的网络I/O资源;二是使用JdbcRDD的API转化为RDD格式,然后编写繁复的函数去实现检索,显然要写更多的代码。而现在,Spark提供了一种新的选择,一行代码就能实现从MySQL到DataFrame的转化,并且支持SQL查询。

实例

首先我们在本地放置了一个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()

结语

Spark的目标在于成为一个跨环境、跨语言、跨工具的大数据处理和分析平台。DataFrame的推出很好诠释了这一目标,从初步的使用来看确实很容易上手。随着性能和稳定性的持续优化,我相信某一天所有玩数据的人,都可以使用Spark作为惟一的平台入口。

转载自: http://segmentfault.com/a/1190000002614456点击打开链接

时间: 2024-10-07 14:34:57

Spark DataFrame小试牛刀的相关文章

spark dataframe操作集锦(提取前几行,合并,入库等)

Spark dataframe派生于RDD类,但是提供了非常强大的数据操作功能.当然主要对类SQL的支持. 在实际工作中会遇到这样的情况,主要是会进行两个数据集的筛选.合并,重新入库. 首先加载数据集,然后在提取数据集的前几行过程中,才找到limit的函数. 而合并就用到union函数,重新入库,就是registerTemple注册成表,再进行写入到Hive中. 不得不赞叹dataframe的强大. 具体示例:为了得到样本均衡的训练集,需要对两个数据集中各取相同的训练样本数目来组成,因此用到了这

spark dataframe unionall

今天本来想写一个spark dataframe unionall的demo,由于粗心报下面错误: Exception in thread "main" org.apache.spark.sql.AnalysisException: Union can only be performed on tables with the same number of columns, but the left table has 3 columns and the right has 4; at o

spark DataFrame 常见操作

spark dataframe派生于RDD类,但是提供了非常强大的数据操作功能.当然主要对类SQL的支持. 在实际工作中会遇到这样的情况,主要是会进行两个数据集的筛选.合并,重新入库. 首先加载数据集,然后在提取数据集的前几行过程中,才找到limit的函数. 而合并就用到union函数,重新入库,就是registerTemple注册成表,再进行写入到HIVE中. 不得不赞叹dataframe的强大. 具体示例:为了得到样本均衡的训练集,需要对两个数据集中各取相同的训练样本数目来组成,因此用到了这

Spark DataFrame ETL教程

前言 ETL是 Extract-Transform-Load的缩写,也就是抽取-转换-加载,在数据工作中是非常重要的部分.实际上,ETL就是一个对数据进行批处理的过程,一个ETL程序就是一个批处理脚本,执行时能将一堆数据转化成我们需要的形式. 每个接触过数据批处理的工程师,都走过ETL的流程,只是没有意识到而已.按照ETL过程的框架来重新认识数据批处理,有利于我们更清晰地编写批处理脚本. 在单机范围内的数据量下,使用python的pandas包就可以非常方便地完成数据批处理工作.但当数据量达到1

Spark DataFrame 数据框空值判断和处理

scala> val data1 = data.toDF("affairs", "gender", "age", "yearsmarried", "children", "religiousness", "education", "occupation", "rating") data1: org.apache.spark

spark&dataframe

1.今天,我们来介绍spark以及dataframe的相关的知识点,但是在此之前先说一下对以前的hadoop的一些理解 当我启动hadoop的时候,上面有hdfs的存储结构,由于这个是分布式存储,所以当一个节点挂了之后,此后由于 还有别的机器上存储这些block块(这里面你肯定要问了,我们怎么知道它挂了,其实我前面关于akaka的时候rpc 通信的机制,心跳机制),所以这个是我们选择它的理由之一,还有一个原因我们可以进行无限扩容,是因为当我们 使用zookeeper进行管理这些datanode的

spark dataframe 类型转换

读一张表,对其进行二值化特征转换.可以二值化要求输入类型必须double类型,类型怎么转换呢? 直接利用spark column 就可以进行转换: DataFrame dataset = hive.sql("select age,sex,race from hive_race_sex_bucktizer "); /** * 类型转换 */ dataset = dataset.select(dataset.col("age").cast(DoubleType).as(

Spark DataFrame写入HBase的常用方式

Spark是目前最流行的分布式计算框架,而HBase则是在HDFS之上的列式分布式存储引擎,基于Spark做离线或者实时计算,数据结果保存在HBase中是目前很流行的做法.例如用户画像.单品画像.推荐系统等都可以用HBase作为存储媒介,供客户端使用. 因此Spark如何向HBase中写数据就成为很重要的一个环节了.本文将会介绍三种写入的方式,其中一种还在期待中,暂且官网即可... 代码在spark 2.2.0版本亲测 1. 基于HBase API批量写入 第一种是最简单的使用方式了,就是基于R

spark dataFrame withColumn

说明:withColumn用于在原有DF新增一列 1. 初始化sqlContext val sqlContext = new org.apache.spark.sql.SQLContext(sc) 2.导入sqlContext隐式转换 import sqlContext.implicits._ 3.  创建DataFrames val df = sqlContext.read.json("file:///usr/local/spark-2.3.0/examples/src/main/resour