Spark学习(2) Spark SQL
什么事sparkSQL
Spark SQL是Spark用来处理结构化数据的一个模块，它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用, 它是将Spark SQL转换成RDD，然后提交到集群执行，执行效率非常快 1）易整合 2）统一的数据访问方式 3）兼容Hive 4）标准的数据连接
SparkSQL可以看做是一个转换层，向下对接各种不同的结构化数据源，向上提供不同的数据访问方式
RDD和Dataframe和DataSet
RDD: 劣势是性能限制，它是一个JVM驻内存对象, 这也就决定了存在GC的限制和数据增加时Java序列化成本的升高 , 无法使用sql进行操作 , 需要考虑怎么做
DataFrame: DataFrame更像传统数据库的二维表格，除了数据以外，还记录数据的结构信息，即schema。DataFrame除了提供了比RDD更丰富的算子以外，更重要的特点是提升执行效率、减少数据读取以及执行计划的优化，比如filter下推、裁剪 , 懒执行 定制化内存管理 DataFrame数据以二进制的方式存在于非堆内存 缺少类型安全检查
dataSet : 既具有类型安全检查也具有Dataframe的查询优化特性
DataFrame/Dataset转RDD：
val rdd1=testDF.rdd
val rdd2=testDS.rdd
RDD转DataFrame
import spark.implicits._
val testDF = rdd.map {line=>
(line._1,line._2)
}.toDF("col1","col2")
RDD转Dataset：
import spark.implicits._
case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型
val testDS = rdd.map {line=>
Coltest(line._1,line._2)
}.toDS
Dataset转DataFrame：
import spark.implicits._
val testDF = testDS.toDF
DataFrame转Dataset：
import spark.implicits._
case class Coltest(col1:String,col2:Int)extends Serializable //定义字段名和类型
val testDS = testDF.as[Coltest]
SparkSession
SparkSession是Spark最新的SQL查询起始点，实质上是SQLContext和HiveContext的组合 代码创建sparkSession import org.apache.spark.sql.SparkSession import spark.implicits._ //将DataFrames隐式转换成RDD val spark = SparkSession .builder() .appName("Spark SQL basic example") .config("spark.some.config.option", "some-value") .enableHiveSupport() //hive支持 .getOrCreate()
DataFrame创建
1. RDD转化 var df = sc.makeRDD(1 to 5).map(x=>(x,x)).toDF("id1","id2").show 2. 指定文件类型创建dfs 入var df1 = spark.read.json("/usr/local/people.json").show 3. schema创建 spark.read.format("csv").schema("a String , b String").option("sep","?").load("/usr/local/people.csv").show
4. 从hive中创建 从jdbc中创建
常用方法: df.show df.printSchema 显示schema df.queryExection 显示查询计划 df.explain 显示物理计划
DSL函数
df1.select("name").show() df1.select($"name", $"age").filter($"age">20).show df1.select("age").filter($"age">20).groupBy("age").count.show df.select($"a" 1,$"b").filter($"a">1001).orderBy(-col("a")).show df.select($"a" 1,$"b").filter($"a">1001).orderBy(-col("a")).take(1)
df.join(df3,Seq("a","a")).show Seq()指连接条件 例如两个表中该字段相同 ---- ---- --------- --------- | a| a| b| b| ---- ---- --------- --------- |1001|1001|zhangshan|zhangshan| ---- ---- --------- ---------
SQL语句
如果要想些sql语句，就需要创建一个临时视图 df1.createOrReplaceTempView("people") df1.createGlobalTempView("people") 全局临时视图 可以跨session使用 然后调用sql spark.sql("select age from people where age is not null group by age").show
自定义函数
注册UDF: spark.udf.register("aaa",(x:String)=>"Name:" x) 使用UDF:spark.sql("select aaa(a) from pp").show UDAF: import org.apache.spark.sql.Row import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction} import org.apache.spark.sql.types._ class MyAverage extends UserDefinedAggregateFunction{ // 聚合函数输入参数的数据类型 override def inputSchema: StructType = StructType(StructField("inputCloum",LongType)::Nil) // 聚合缓冲区中值得数据类型 override def bufferSchema: StructType = StructType(StructField("sum",LongType)::StructField("count",LongType)::Nil) // 返回值的数据类型 override def dataType: DataType = DoubleType // 对于相同的输入是否一直返回相同的输出。 // 确保一致性 一般用true,用以标记针对给定的一组输入，UDAF是否总是生成相同的结果。 override def deterministic: Boolean = true // 初始化 override def initialize(buffer: MutableAggregationBuffer): Unit = { // 存工资的总额 buffer(0) = 0L // 存工资的个数 buffer(1) = 0L } // 相同Execute间的数据合并。 override def update(buffer: MutableAggregationBuffer, input: Row): Unit = { if(!input.isNullAt(0)){ //工资累加 buffer(0) = buffer.getLong(0) input.getLong(0) //工资个数累加 buffer(1) = buffer.getLong(1) 1 } } // 不同Execute间的数据合并 override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = { //工资累加 buffer1(0) = buffer1.getLong(0) buffer2.getLong(0) //工资个数相加 buffer1(1) = buffer1.getLong(1) buffer2.getLong(1) } override def evaluate(buffer: Row): Double = { buffer.getLong(0).toDouble/buffer.getLong(1) } }开窗函数 spark.sql("select * ,rank() over(partition by _2 order by _3 desc) _4 from t_score").show
加载保存文件
SparkSession提供的read.load方法用于通用加载数据，使用write和save保存数据
val peopleDF = spark.read.format("json").load("examples/src/main/resources/people.json") peopleDF.write.format("parquet").save("hdfs://hadoop102:9000/namesAndAges.parquet")
scoreRDD.write.format("parquet").save("hdfs://cc1:9000/score.parquet")
DF.write.parquet("hdfs://master01:9000/people.parquet")
spark.read.parquet("hdfs://master01:9000/people.parquet") //parquest是可以高效存储嵌套字记录的列式存储格式 , 基于B 树
内置函数大全
聚合函数 approx_count_distinct count_distinct近似值 avg 平均值 collect_list 聚合指定字段的值到list collect_set 聚合指定字段的值到set corr 计算两列的Pearson相关系数 count 计数 countDistinct 去重计数 SQL中用法 select count(distinct class) covar_pop 总体协方差（population covariance） covar_samp 样本协方差（sample covariance） first 分组第一个元素 last 分组最后一个元素 grouping grouping_id kurtosis 计算峰态(kurtosis)值 skewness 计算偏度(skewness) max 最大值 min 最小值 mean 平均值 stddev 即stddev_samp stddev_samp 样本标准偏差（sample standard deviation） stddev_pop 总体标准偏差（population standard deviation） sum 求和 sumDistinct 非重复值求和 SQL中用法 select sum(distinct class) var_pop 总体方差（population variance） var_samp 样本无偏方差（unbiased variance） variance 即var_samp 集合函数 array_contains(column,value) 检查array类型字段是否包含指定元素 explode 展开array或map为多行 explode_outer 同explode，但当array或map为空或null时，会展开为null。 posexplode 同explode，带位置索引。 posexplode_outer 同explode_outer，带位置索引。 from_json 解析JSON字符串为StructType or ArrayType，有多种参数形式，详见文档。 to_json 转为json字符串，支持StructType, ArrayType of StructTypes, a MapType or ArrayType of MapTypes。 get_json_object(column,path) 获取指定json路径的json对象字符串。 select get_json_object(‘{"a"1,"b":2}‘,‘$.a‘)? [JSON Path介绍](http://blog.csdn.net/koflance/article/details/63262484) json_tuple(column,fields) 获取json中指定字段值。select json_tuple(‘{"a":1,"b":2}‘,‘a‘,‘b‘)? map_keys 返回map的键组成的array map_values 返回map的值组成的array size array or map的长度 sort_array(e: Column, asc: Boolean) 将array中元素排序（自然排序），默认asc。 时间函数 add_months(startDate: Column, numMonths: Int) 指定日期添加n月 date_add(start: Column, days: Int) 指定日期之后n天 e.g. select date_add(‘20180101‘,3) date_sub(start: Column, days: Int) 指定日期之前n天 datediff(end: Column, start: Column) 两日期间隔天数 current_date() 当前日期 current_timestamp() 当前时间戳，TimestampType类型 date_format(dateExpr: Column, format: String) 日期格式化 dayofmonth(e: Column) 日期在一月中的天数，支持 date/timestamp/string dayofyear(e: Column) 日期在一年中的天数， 支持 date/timestamp/string weekofyear(e: Column) 日期在一年中的周数， 支持 date/timestamp/string from_unixtime(ut: Column, f: String) 时间戳转字符串格式 from_utc_timestamp(ts: Column, tz: String) 时间戳转指定时区时间戳 to_utc_timestamp(ts: Column, tz: String) 指定时区时间戳转UTF时间戳 hour(e: Column) 提取小时值 minute(e: Column) 提取分钟值 month(e: Column) 提取月份值 quarter(e: Column) 提取季度 second(e: Column) 提取秒 year(e: Column):提取年 last_day(e: Column) 指定日期的月末日期 months_between(date1: Column, date2: Column) 计算两日期差几个月 next_day(date: Column, dayOfWeek: String) 计算指定日期之后的下一个周一、二...，dayOfWeek区分大小写，只接受 "Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"。 to_date(e: Column) 字段类型转为DateType trunc(date: Column, format: String) 日期截断 unix_timestamp(s: Column, p: String) 指定格式的时间字符串转时间戳 unix_timestamp(s: Column) 同上，默认格式为 yyyyMMdd HH:mm:ss unix_timestamp():当前时间戳(秒),底层实现为unix_timestamp(current_timestamp(), yyyy?MMdd HH:mm:ss) window(timeColumn: Column, windowDuration: String, slideDuration: String, startTime: String) 时间窗口函数，将指定时间(TimestampType)划分到窗口 数学函数 cos,sin,tan 计算角度的余弦，正弦。。。 sinh,tanh,cosh 计算双曲正弦，正切，。。 acos,asin,atan,atan2 计算余弦/正弦值对应的角度 bin 将long类型转为对应二进制数值的字符串For example, bin("12") returns "1100". bround 舍入，使用Decimal的HALF_EVEN模式，v>0.5向上舍入，v< 0.5向下舍入，v0.5向最近的 偶数舍入。 round(e: Column, scale: Int) HALF_UP模式舍入到scale为小数点。v>=0.5向上舍入，v< 0.5向下舍入,即四舍五入。 ceil 向上舍入 floor 向下舍入 cbrt Computes the cuberoot of the given value. conv(num:Column, fromBase: Int, toBase: Int) 转换数值（字符串）的进制 log(base: Double, a: Column):$log_{base}(a)$ log(a: Column):$log_e(a)$ log10(a: Column):$log_{10}(a)$ log2(a: Column):$log_{2}(a)$ log1p(a: Column):$log_{e}(a 1)$ pmod(dividend: Column, divisor: Column):Returns the positive value of dividend mod divisor. pow(l: Double, r: Column):$r^l$ 注意r是列 pow(l: Column, r: Double):$r^l$ 注意l是列 pow(l: Column, r: Column):$r^l$ 注意r,l都是列 radians(e: Column):角度转弧度 rint(e: Column):Returns the double value that is closest in value to the argument and is equal to a mathematical integer. shiftLeft(e: Column, numBits: Int):向左位移 shiftRight(e: Column, numBits: Int):向右位移 shiftRightUnsigned(e: Column, numBits: Int):向右位移（无符号位） signum(e: Column):返回数值正负符号 sqrt(e: Column):平方根 hex(column: Column):转十六进制 unhex(column: Column):逆转十六进制 混杂(misc)函数 crc32(e: Column):计算CRC32,返回bigint hash(cols: Column*):计算 hash code，返回int md5(e: Column):计算MD5摘要，返回32位，16进制字符串 sha1(e: Column):计算SHA1摘要，返回40位，16进制字符串 sha2(e: Column, numBits: Int):计算SHA1摘要，返回numBits位，16进制字符串。numBits 支持224, 256, 384, or 512. 其他非聚合函数 abs(e: Column) 绝对值 array(cols: Column*) 多列合并为array，cols必须为同类型 map(cols: Column*): 将多列组织为map，输入列必须为（key,value)形式，各列的key/value分别为同一类型。 bitwiseNOT(e: Column): Computes bitwise NOT. broadcast[T](df: Dataset[T]): Dataset[T]: 将df变量广播，用于实现broadcast join。如left.join(broadcast(right), "joinKey") coalesce(e: Column*): 返回第一个非空值 col(colName: String): 返回colName对应的Column column(colName: String): col函数的别名 expr(expr: String): 解析expr表达式，将返回值存于Column，并返回这个Column。 greatest(exprs: Column*): 返回多列中的最大值，跳过Null least(exprs: Column*): 返回多列中的最小值，跳过Null input_file_name():返 回当前任务的文件名 ？？ isnan(e: Column): 检查是否NaN（非数值） isnull(e: Column): 检查是否为Null lit(literal: Any): 将字面量(literal)创建一个Column typedLit[T](literal: T)(implicit arg0: scala.reflect.api.JavaUniverse.TypeTag[T]): 将字面量(literal)创建一个Column，literal支持 scala types e.g.: List, Seq and Map. monotonically_increasing_id(): 返回单调递增唯一ID，但不同分区的ID不连续。ID为64位整型。 nanvl(col1: Column, col2: Column): col1为NaN则返回col2 negate(e: Column): 负数，同df.select( df("amount") ) not(e: Column): 取反，同df.filter( !df("isActive") ) rand(): 随机数[0.0, 1.0] rand(seed: Long): 随机数[0.0, 1.0]，使用seed种子 randn(): 随机数，从正态分布取 randn(seed: Long): 同上 spark_partition_id(): 返回partition ID struct(cols: Column*): 多列组合成新的struct column ？？ when(condition: Column, value: Any): 当condition为true返回value，如 people.select(when(people("gender") === "male", 0) .when(people("gender") === "female", 1) .otherwise(2)) 如果没有otherwise且condition全部没命中，则返回null. 排序函数 asc(columnName: String):正序 asc_nulls_first(columnName: String):正序，null排最前 asc_nulls_last(columnName: String):正序，null排最后 e.g. df.sort(asc("dept"), desc("age")) 对应有desc函数 desc,desc_nulls_first,desc_nulls_last 字符串函数 ascii(e: Column): 计算第一个字符的ascii码 base64(e: Column): base64转码 unbase64(e: Column): base64解码 concat(exprs: Column*):连接多列字符串 concat_ws(sep: String, exprs: Column*):使用sep作为分隔符连接多列字符串 decode(value: Column, charset: String): 解码 encode(value: Column, charset: String): 转码，charset支持 ‘USASCII‘, ‘ISO88591‘, ‘UTF8‘, ‘UTF16BE‘, ‘UTF16LE‘, ‘UTF16‘。 format_number(x: Column, d: Int):格式化‘#,###,###.##‘形式的字符串 format_string(format: String, arguments: Column*): 将arguments按format格式化，格式为 printfstyle。 initcap(e: Column): 单词首字母大写 lower(e: Column): 转小写 upper(e: Column): 转大写 instr(str: Column, substring: String): substring在str中第一次出现的位置 length(e: Column): 字符串长度 levenshtein(l: Column, r: Column): 计算两个字符串之间的编辑距离（Levenshtein distance） locate(substr: String, str: Column): substring在str中第一次出现的位置，位置编号从1开 始，0表示未找到。 locate(substr: String, str: Column, pos: Int): 同上，但从pos位置后查找。 lpad(str: Column, len: Int, pad: String):字符串左填充。用pad字符填充str的字符串至len长 度。有对应的rpad，右填充。 ltrim(e: Column):剪掉左边的空格、空白字符，对应有rtrim. ltrim(e: Column, trimString: String):剪掉左边的指定字符,对应有rtrim. trim(e: Column, trimString: String):剪掉左右两边的指定字符 trim(e: Column):剪掉左右两边的空格、空白字符 regexp_extract(e: Column, exp: String, groupIdx: Int): 正则提取匹配的组 regexp_replace(e: Column, pattern: Column, replacement: Column): 正则替换匹配的部 分，这里参数为列。 regexp_replace(e: Column, pattern: String, replacement: String): 正则替换匹配的部分 repeat(str: Column, n: Int):将str重复n次返回 reverse(str: Column): 将str反转 soundex(e: Column): 计算桑迪克斯代码（soundex code）PS:用于按英语发音来索引姓名, 发音相同但拼写不同的单词，会映射成同一个码。 split(str: Column, pattern: String): 用pattern分割str substring(str: Column, pos: Int, len: Int): 在str上截取从pos位置开始长度为len的子字符串。 substring_index(str: Column, delim: String, count: Int):Returns the substring from string str before count occurrences of the delimiter delim. If count is positive, everything the left of the final delimiter (counting from left) is returned. If count is negative, every to the right of the final delimiter (counting from the right) is returned. substring_index performs a case?sensitive match when searching for delim. translate(src: Column, matchingString: String, replaceString: String):把src中的 matchingString全换成replaceString。 窗口函数 cume_dist(): cumulative distribution of values within a window partition currentRow(): returns the special frame boundary that represents the current row in the window partition. rank():排名，返回数据项在分组中的排名，排名相等会在名次中留下空位 1,2,2,4。 dense_rank(): 排名，返回数据项在分组中的排名，排名相等会在名次中不会留下空位 1,2,2,3。 row_number():行号，为每条记录返回一个数字 1,2,3,4 percent_rank():returns the relative rank (i.e. percentile) of rows within a window partition. lag(e: Column, offset: Int, defaultValue: Any): offset rows before the current row lead(e: Column, offset: Int, defaultValue: Any): returns the value that is offset rows after the current row ntile(n: Int): returns the ntile group id (from 1 to n inclusive) in an ordered window partition. unboundedFollowing():returns the special frame boundary that represents the last row in the window partition.
温馨提示: 本文由Jm博客推荐，转载请保留链接: https://www.jmwww.net/a/22982.html
