43、内置函数及每日uv、销售额统计案例
一、spark1.5内置函数
在Spark 1.5.x版本,增加了一系列内置函数到DataFrame API中,并且实现了code-generation的优化。与普通的函数不同,DataFrame的函数并不会执行后立即返回一个结果值,
而是返回一个Column对象,用于在并行作业中进行求值。Column可以用在DataFrame的操作之中,比如select,filter,groupBy等。函数的输入值,也可以是Column。
种类 |
函数 |
聚合函数 |
approxCountDistinct, avg, count, countDistinct, first, last, max, mean, min, sum, sumDistinct |
集合函数 |
array_contains, explode, size, sort_array |
日期/时间函数 |
日期时间转换 |
日期/时间函数 |
日期/时间计算 |
数学函数 |
abs, acros, asin, atan, atan2, bin, cbrt, ceil, conv, cos, sosh, exp, expm1, factorial, floor, hex, hypot, log, log10, log1p, log2, pmod, pow, rint, round, shiftLeft, shiftRight, shiftRightUnsigned, signum, sin, sinh, sqrt, tan, tanh, toDegrees, toRadians, unhex |
混合函数 |
array, bitwiseNOT, callUDF, coalesce, crc32, greatest, if, inputFileName, isNaN, isnotnull, isnull, least, lit, md5, monotonicallyIncreasingId, nanvl, negate, not, rand, randn, sha, sha1, sparkPartitionId, struct, when |
字符串函数 |
ascii, base64, concat, concat_ws, decode, encode, format_number, format_string, get_json_object, initcap, instr, length, levenshtein, locate, lower, lpad, ltrim, printf, regexp_extract, regexp_replace, repeat, reverse, rpad, rtrim, soundex, space, split, substring, substring_index, translate, trim, unbase64, upper |
窗口函数 |
cumeDist, denseRank, lag, lead, ntile, percentRank, rank, rowNumber |
二、案例
案例实战:根据每天的用户访问日志和用户购买日志,统计每日的uv和销售额
1、UV案例scala实现
package cn.spark.study.sql import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.sql.functions._ object DailyUV {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setMaster("local").setAppName("DailyUV")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc) // 要使用Spark SQL的内置函数,就必须在这里导入SQLContext下的隐式转换
import sqlContext.implicits._ // 构造用户访问日志数据,并创建DataFrame // 模拟用户访问日志,日志用逗号隔开,第一列是日期,第二列是用户id
val userAccessLog = Array(
"2019-08-04,1122",
"2019-08-04,1122",
"2019-08-04,1123",
"2019-08-04,1124",
"2019-08-04,1124",
"2019-08-05,1122",
"2019-08-05,1121",
"2019-08-05,1123",
"2019-08-05,1123"); val userAccessLogRDD = sc.parallelize(userAccessLog, 5) // 将模拟出来的用户访问日志RDD,转换为DataFrame
// 首先,将普通的RDD,转换为元素为Row的RDD
// String到Int : toInt
val userAccessLogRowRDD = userAccessLogRDD
.map { log => Row(log.split(",")(0), log.split(",")(1).toInt) } // 构造DataFrame的元数据
// 将一个RDD转换为DataFrame,这一步经常需要生成一个StructType来生成DataFrame的schema
// 通过StructType直接指定每个字段的schema
val structType = StructType(Array(
StructField("date", StringType, true),
StructField("userid", IntegerType, true))) // 使用SQLContext创建DataFrame
val userAccessLogRowDF = sqlContext.createDataFrame(userAccessLogRowRDD, structType) // 这里讲解一下uv的基本含义和业务
// 每天都有很多用户来访问,但是每个用户可能每天都会访问很多次
// 所以,uv,指的是,对用户进行去重以后的访问总数 // 这里,正式开始使用Spark 1.5.x版本提供的最新特性,内置函数,countDistinct
// 讲解一下聚合函数的用法
// 首先,对DataFrame调用groupBy()方法,对某一列进行分组
// 然后,调用agg()方法 ,第一个参数,必须传入之前在groupBy()方法中出现的字段,前面要写一个单引号
// 第二个参数,传入countDistinct、sum、first等,Spark提供的内置函数
// 内置函数中,传入的参数,也是用单引号作为前缀的,其他的字段
userAccessLogRowDF.groupBy("date")
.agg('date, countDistinct('userid))
.map { row => Row(row(1), row(2)) }
.collect()
.foreach(println) }
}
2、销售额案例scala实现
package cn.spark.study.sql import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.apache.spark.sql.types.StructField
import org.apache.spark.sql.types.StringType
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.sql.functions._ /**
* @author Administrator
*/ object DailySale { def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setMaster("local")
.setAppName("DailySale")
val sc = new SparkContext(conf)
val sqlContext = new SQLContext(sc) import sqlContext.implicits._ // 说明一下,业务的特点
// 实际上呢,我们可以做一个,单独统计网站登录用户的销售额的统计
// 有些时候,会出现日志的上报的错误和异常,比如日志里丢了用户的信息,那么这种,我们就一律不统计了 // 模拟数据
val userSaleLog = Array("2019-08-04,55.05,1122",
"2019-08-04,23.15,1133",
"2019-08-04,15.20,",
"2019-08-05,56.05,1144",
"2019-08-05,78.87,1155",
"2019-08-05,113.02,1123")
val userSaleLogRDD = sc.parallelize(userSaleLog, 5) // 进行有效销售日志的过滤
val filteredUserSaleLogRDD = userSaleLogRDD
.filter { log => if (log.split(",").length == 3) true else false } val userSaleLogRowRDD = filteredUserSaleLogRDD
.map { log => Row(log.split(",")(0), log.split(",")(1).toDouble) } val structType = StructType(Array(
StructField("date", StringType, true),
StructField("sale_amount", DoubleType, true))) val userSaleLogDF = sqlContext.createDataFrame(userSaleLogRowRDD, structType) // 开始进行每日销售额的统计
userSaleLogDF.groupBy("date")
.agg('date, sum('sale_amount))
.map { row => Row(row(1), row(2)) }
.collect()
.foreach(println)
} }
最新文章
- position
- dyld: Library not loaded: @rpath/libswiftCore.dylib 解决方法
- ZooKeeper 笔记(4) 实战应用之【消除单点故障】
- Java基础之扩展GUI——显示About对话框(Sketcher 2 displaying an About dialog)
- 自己动手做jQuery插件
- 每天一道LeetCode--58. Length of Last Word
- 转:HTTP请求(GET、POST和soap区别)和响应
- Moduli number system
- 判断文件是否存在(exist)
- IIS下访问网络驱动器(网络位置)
- 查看当前支持的shell,echo -e相关转义符,一个简单shell脚本,dos2unix命令把windows格式转为Linux格式
- JavaScript网页全屏API
- PS抠图简单粗暴2种方法
- Win10系统下安装Gradle-3.5
- 各种OJ网站汇总
- PS_图象调整_太暗/过亮_曝光不足/过度
- Python之traceback错误堆栈信息处理
- 使用XWAF框架(1)——Web项目的代码分层
- js-jQuery对象与dom对象相互转换(转载)
- lintcode-436-最大正方形