- 浏览: 230701 次
- 性别:
- 来自: 上海
文章分类
最新评论
-
lwb314:
你的这个是创建的临时的hive表,数据也是通过文件录入进去的, ...
Spark SQL操作Hive数据库 -
yixiaoqi2010:
你好 我的提交上去 总是报错,找不到hive表,可能是哪里 ...
Spark SQL操作Hive数据库 -
bo_hai:
target jvm版本也要选择正确。不能选择太高。2.10对 ...
eclipse开发spark程序配置本地运行
简单说明
使用Spark SQL中的内置函数对数据进行分析,Spark SQL API不同的是,DataFrame中的内置函数操作的结果返回一个Column对象,而DataFrame天生就是“A distributed collection of data organized into named columns”,这就为数据的复杂分析建立了坚实的基础并提供了极大的方便性,例如说,我们在操作DataFrame的方法中可以随时调用内置函数进行业务需要的处理,这之于我们构建附件的业务逻辑而言是可以极大的减少不必须的时间消耗(基本上就是实际模型的映射),让我们聚焦在数据分析上,这对于提高工程师的生产力而言是非常有价值的;
Spark 1.5.x开始提供了大量的内置函数,例如agg:
还有max,mean,min,sum,ave,explode,size,sort_array,day,to_date,abs,acros,asin,atan等
总体而言,内置函数包含了五大基本类型:
1.聚合函数,例如countDistinct,sumDistinct等;
2.集合函数,例如sort_array,explode等;
3.日期,时间函数,例如hour,quarter,next_day等;
4.数学函数,例如asin,atan,sqrt,tan,round等;
5.开窗函数,例如rowNumber等;
6.字符串函数,例如concat,format_number,rexexp_extract(正则)等;
7.其他函数,isNaN,sha,randn,callUDF等;
执行结果
使用Spark SQL中的内置函数对数据进行分析,Spark SQL API不同的是,DataFrame中的内置函数操作的结果返回一个Column对象,而DataFrame天生就是“A distributed collection of data organized into named columns”,这就为数据的复杂分析建立了坚实的基础并提供了极大的方便性,例如说,我们在操作DataFrame的方法中可以随时调用内置函数进行业务需要的处理,这之于我们构建附件的业务逻辑而言是可以极大的减少不必须的时间消耗(基本上就是实际模型的映射),让我们聚焦在数据分析上,这对于提高工程师的生产力而言是非常有价值的;
Spark 1.5.x开始提供了大量的内置函数,例如agg:
def agg(aggExpr: (String, String), aggExprs: (String, String)*): DataFrame = { groupBy().agg(aggExpr, aggExprs : _*) }
还有max,mean,min,sum,ave,explode,size,sort_array,day,to_date,abs,acros,asin,atan等
总体而言,内置函数包含了五大基本类型:
1.聚合函数,例如countDistinct,sumDistinct等;
2.集合函数,例如sort_array,explode等;
3.日期,时间函数,例如hour,quarter,next_day等;
4.数学函数,例如asin,atan,sqrt,tan,round等;
5.开窗函数,例如rowNumber等;
6.字符串函数,例如concat,format_number,rexexp_extract(正则)等;
7.其他函数,isNaN,sha,randn,callUDF等;
package com.imf.spark.sql import org.apache.spark.sql.{SQLContext, Row} import org.apache.spark.sql.types.{StructType, IntegerType, StringType, StructField} import org.apache.spark.{SparkConf, SparkContext} import org.apache.spark.sql.functions._ /** * Description:使用Spark SQL中的内置函数对数据进行分析 * Author:lujinyong168 * Date:2016/4/14 21:09 */ object SparkSQLAgg { def main(args: Array[String]): Unit = { val conf = new SparkConf() conf.setAppName("SparkSQLAgg for scala") conf.setMaster("local") val sc = new SparkContext(conf) // val sqlContext = new HiveContext(sc)//构建上下文 val sqlContext = new SQLContext(sc)//构建上下文 //要使用Spark SQL 的内置函数,就一定要导入SQLContext下的隐式转换 import sqlContext.implicits._ /** * 简单模拟电商访问的数据 */ val userData = Array( "2016-04-15,1001,http://spark.apache.org,1000", "2016-04-15,1001,http://hadoop.apache.org,1001", "2016-04-15,1002,http://fink.apache.org,1002", "2016-04-16,1003,http://kafka.apache.org,1020", "2016-04-16,1004,http://spark.apache.org,1010", "2016-04-16,1002,http://hive.apache.org,1200", "2016-04-16,1001,http://parquet.apache.org,1500", "2016-04-16,1001,http://spark.apache.org,1800" ) val userDataRDD = sc.parallelize(userData)//生成RDD分布式集合对象 /** * 根据业务需要对数据进行预处理生成DataFrame,要想把RDD转换成DataFrame,需要先把RDD中的元素类型变成Row类型 * 与此同时要提供DataFrame中的Columns的元素信息描述 */ val userDataRDDRow = userDataRDD.map(row=> {val splited = row.split(",");Row(splited(0),splited(1).toInt,splited(2),splited(3).toInt)}) val structTypes = StructType(Array( StructField("date",StringType,true), StructField("id",IntegerType,true), StructField("url",StringType,true), StructField("amount",IntegerType,true) )) val userDataDF = sqlContext.createDataFrame(userDataRDDRow,structTypes) /** * 使用Spark SQL提供的内置函数对DataFrame进行操作,特别注意:内置函数生成的Column对象且自动进行CG(code generation) * 所有的内置函数操作结果都会返回具体的列 * DataFrame中的列可以动态增长 */ //userDataDF.groupBy("date").agg('date,countDistinct('id)).show() // userDataDF.groupBy("date").agg('date,countDistinct('id)).map(row => Row(row(1),row(2))).collect.foreach(println)//数据量比较大时不能用collect userDataDF.groupBy("date").agg('date,sum('amount)).show()//对销售额统计 } }
执行结果
userDataDF.groupBy("date").agg('date,countDistinct('id)).show() 返回的结果: +----------+----------+---------+ | date| date|count(id)| +----------+----------+---------+ |2016-04-15|2016-04-15| 2| |2016-04-16|2016-04-16| 4| +----------+----------+---------+ userDataDF.groupBy("date").agg('date,countDistinct('id)).map(row => Row(row(1),row(2))).collect.foreach(println)//数据量比较大时不能用collect 返回的结果: [2016-04-15,2] [2016-04-16,4] userDataDF.groupBy("date").agg('date,sum('amount)).show()//对销售额统计 返回的结果: +----------+----------+-----------+ | date| date|sum(amount)| +----------+----------+-----------+ |2016-04-15|2016-04-15| 3003| |2016-04-16|2016-04-16| 6530| +----------+----------+-----------+
发表评论
-
SparkStreaming pull data from Flume
2016-06-19 17:29 1195Spark Streaming + Flume Integra ... -
Flume push数据到SparkStreaming
2016-06-19 15:16 1894上节http://kevin12.iteye.com/blog ... -
Spark Streaming 统计单词的例
2016-06-19 14:55 3测试Spark Streaming 统计单词的例子 1.准 ... -
Spark Streaming 统计单词的例子
2016-06-19 12:29 3633测试Spark Streaming 统计单词的例子 1.准备 ... -
Spark SQL窗口函数
2016-04-22 07:18 2508窗口函数又叫着窗口分析函数,Spark 1.4版本SparkS ... -
Spark SQL操作Hive数据库
2016-04-13 22:37 17542本次例子通过scala编程实现Spark SQL操作Hive数 ... -
Spark SQL on hive配置和实战
2016-03-26 18:40 5491spark sql 官网:http://spark ... -
Hive分析搜索引擎的数据(搜狗实验室数据)
2016-03-13 12:48 1679搜狗实验室:http://www.sogou.com/labs ... -
Hive使用默认数据库derby报错及解决方法
2016-03-05 21:19 2984在使用derby数据库时,遇到了很多错误,报错信息和解决方法如 ... -
Hive1.2.1安装和使用(基于Hadoop2.6.0)
2016-03-05 20:22 8744安装hive,这里使用mysql作为hive的metastor ... -
Spark RDD弹性表现和来源
2016-02-09 20:12 3818hadoop 的MapReduce是基于数 ... -
Spark内核架构
2016-02-07 12:24 9761.在将spark内核架构前,先了解一下Hadoop的MR,H ... -
spark集群HA搭建
2016-01-31 08:50 4444spark集群的HA图: 搭建spark的HA需要安装z ... -
Spark集群中WordCount运行原理
2016-01-31 07:05 2477以数据流动的视角解释一下wordcount运行的原理 pa ... -
eclipse开发spark程序配置在集群上运行
2016-01-27 08:08 9292这篇bolg讲一下,IDE开发的spark程序如何提交到集群上 ... -
eclipse开发spark程序配置本地运行
2016-01-27 07:58 12311今天简单讲一下在local模式下用eclipse开发一个简单的 ... -
spark1.6.0搭建(基于hadoop2.6.0分布式)
2016-01-24 10:11 5924本文是基于hadoop2.6.0的分布式环境搭建spark1. ... -
hive第一个字段为null
2014-03-20 23:17 1332在hive中创建表后,将准备好的数据导入到该表中,如果第一个字 ...
相关推荐
主要给大家介绍了关于Spark SQL操作JSON字段的小技巧,文中通过示例代码介绍的非常详细,对大家学习或者使用spark sql具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧。
Learning Spark SQL 英文epub 本资源转载自网络,如有侵权,请联系上传者或csdn删除 本资源转载自网络,如有侵权,请联系上传者或csdn删除
Spark SQL的DataFrame接口支持多种数据源的操作。一个DataFrame可以进行RDDs方式的操作,也可以被注册为临时表。把DataFrame注册为临时表之后,就可以对该DataFrame执行SQL查询。 Spark SQL的默认数据源为Parquet...
Databrciks工程师,Spark Committer,Spark SQL主要开发者之一的连城详细解读了“Spark SQL结构化数据分析”。他介绍了Spark1.3版本中的很多新特性。重点介绍了DataFrame。其从SchemaRDD演变而来,提供了更加高层...
表达式计算在Spark SQL中随处可见,本演讲将简介表达式、UDF、UDAF、UDTF的概念,主要的API,以及如何扩展Spark SQL函数库。本演讲还将提及Catalyst在计划阶段和Project Tungsten在执行层做的优化,以及未来性能提升...
参考Spark官网以及一些文献,制作的Spark SQL教学幻灯片,适合进行Spark入门介绍与教学!所有的Spark教学系列都在我的资源内!
Spark SQL 详细介绍 实验介绍 有需要的尽快下载吧
实训指导书_使用Spark SQL进行法律服务网站数据分析.zip
Learning Spark SQL - Aurobindo Sarka
使用spark读取hbase中的数据,并插入到mysql中
Spark SQL操作大全.zip
Spark SQL源码概览.zip Spark SQL源码概览.zip Spark SQL源码概览.zip Spark SQL源码概览.zipSpark SQL源码概览.zip
List<row> list= spark.sql(sql).collectAsList(),获或者其他方法将数据存在List里面,然后就list转为 Dataset分批入库
spark-sql解析-源码分析
Spark SQL Core 封装 Catalyst,向应用程序提供 SparkSession、Dataset、 DataFrame 等 API(DataFrame 是 Dataset[Row]的别名);Spark SQL Hive 则提供操作 Hive 的 接口。本文主要关注查询执行过程,不涉及 ...
我们把基于成本的优化器框架贡献给社区版本Spark 2.2。...这个基于成本的优化器框架对Spark SQL查询的性能有很好的提升 。在这次演讲中,我们将展示Spark SQL的新的基于成本的优化器框架及其对TPC-DS查询的性能影响。
spark官文文档翻译有关spark SQL编程指南,spark SQL是以测试组件,有关api的介绍