Spark学习(2) Spark SQL

Spark SQL是Spark用来处理结构化数据的一个模块，它提供了一个编程抽象叫做DataFrame并且作为分布式SQL查询引擎的作用, 它是将Spark SQL转换成RDD，然后提交到集群执行，执行效率非常快
1）易整合
2）统一的数据访问方式
3）兼容Hive
4）标准的数据连接

SparkSQL可以看做是一个转换层，向下对接各种不同的结构化数据源，向上提供不同的数据访问方式

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是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()

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 显示物理计划

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语句，就需要创建一个临时视图
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)
  }
}

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(‘2018­01­01‘,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)
同上，默认格式为 yyyy­MM­dd HH:mm:ss
unix_timestamp():当前时间戳(秒),底层实现为unix_timestamp(current_timestamp(), yyyy?MM­dd 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 cube­root 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):计算SHA­1摘要，返回40位，16进制字符串
sha2(e: Column, numBits: Int):计算SHA­1摘要，返回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支持 ‘US­ASCII‘, ‘ISO­8859­1‘,
‘UTF­8‘, ‘UTF­16BE‘, ‘UTF­16LE‘, ‘UTF­16‘。
format_number(x: Column, d: Int):格式化‘#,###,###.##‘形式的字符串
format_string(format: String, arguments: Column*): 将arguments按format格式化，格式为
printf­style。
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.