Spark SQL基础知识

一.DataFrame详解

1.清洗相关的API

去重API:dropDuplicates

总结:用来删除重复数据,如果没有指定参数subset,那么要比对行中的所有字段内容,如果全部相同,就认为是重复数据,会被删除;如果有指定参数subset,那么只比对subset中指定的字段范围,如果指定不存在的字段会报错.

删除缺失值的API:dropna

总结:

1- 如果不传递任何参数,只要有任意一个字段值为null,那么就删除整行数据

2-如果只指定了subset,那么空值的检查,就只会限定在subset指定的范围内

3-如果只指定了thresh,那么空值检查的这些字段中,至少需要有thresh(>=thresh)个字段的值不为空,才不会被删除

替换缺失值的API:fillna

?总结:替换缺失值数据

value:必须要传递参数,是用来填充缺失值的

subset:限定缺失值替换范围

注意:

1- value如果不是字典,那么只会替换字段类型匹配空值

2- 最常用的是value传递字典的形式

?2. Spark SQL的shuffle分区设置

如何调整shuffle分区数量呢?

方案一(不推荐):直接修改spark的配置文件spark-defaults.conf,全局设置,默认值为200,设置为:

spark.sql.shuffle.partitions? 20

方案二(常用,推荐使用):在客户端通过submit命令提交的时候,动态设置shuffle的分区数量,部署,上线的时候,基于spark-submit提交运行的时候

./spark-submit --conf "spark.sql.shuffle.partitions=20"

方案三(比较常用):在代码中设置,主要在测试环境中使用,但是一般在部署上线的时候,会删除.优先级也是最高的,一般的使用场景,当你的数据量未来不会发生太大的波动.

sparkSession.conf.set('spark.sql.shuffle.partition',20)

3.数据写出操作

统一的输出语法:

对应的简写API格式如下,以CSV为例:

init_df.write.csv(

? ? ? ? path='存储路径',

? ? ? ? mode='模式',

? ? ? ? header=True,

? ? ? ? sep='\001',

? ? ? ? encoding='UTF-8'?

)

常用参数说明：
? ? 1- path：指定结果数据输出路径。支持本地文件系统和HDFS文件系统
? ? 2- mode：当输出目录中文件已经存在的时候处理办法
? ? ? ? 2.1- append：追加。如果文件已经存在，那么继续在该目录下产生新的文件
? ? ? ? 2.2- overwrite：覆盖。如果文件已经存在，那么就先将已有的文件清除，再写入进去
? ? ? ? 2.3- ignore：忽略。如果文件已经存在，那么不执行任何操作
? ? ? ? 2.4- error：报错。如果文件已经存在，那么直接报错。会报错AnalysisException: path ?? ?
? ? ? ? ?? ??? ??? ?file:xxx already exists.
? ? ? ??
? ? 3- sep：字段间的分隔符
? ? 4- header：数据输出的时候，是否要将字段名称输出到文件的第一行。推荐设置为True
? ? 5- encoding：文件输出的编码方式

二.常见的DSL代码

分类	格式	含义	示例
API/方法	select	查询字段	select('id1', 'id2')
	where	对数据过滤	where('avg_score>3')
	groupBy	对数据分组	groupBy('userid')
	orderBy	对数据排序	orderBy('cnt', ascending=False)
	limit	取前几条数据	orderBy('cnt', ascending=False).limit(1)
	agg	聚合操作，里面可以写多个聚合表达式	agg(F.round(F.avg('score'), 2).alias('avg_score'))
	show	打印数据	init_df.show()
	printSchema	打印数据的schema信息，也就是元数据信息	init_df.printSchema()
	alias	对字段取别名	F.count('movieid').alias('cnt')
	join	关联2个DataFrame	etl_df.join(avg_score_dsl_df, 'movieid')
	withColumn	基于目前的数据产生一个新列	init_df.withColumn('word',F.explode(F.split('value', ' ')))
	dropDuplicates	删除重复数据	init_df.dropDuplicates(subset=["id","name"])
	dropna	删除缺失值	init_df.dropna(thresh=2,subset=["name","age","address"])
	fillna	替换缺失值	init_df.fillna(value={"id":111,"name":"未知姓名","age":100,"address":"北京"})
	first	取DataFrame中的第一行数据
	over	创建一个窗口列
函数	avg	计算均值
	count	计数
	col	将字段包装成Column对象，一般用于对新列的包装
	round	保留小数位
	desc	降序排序
	row_number	行号。从1开始编号
窗口	partitionBy	对数据分区
	orderBy	对数据排序

1-什么时候使用select实现聚合,什么时候使用groupBy+agg/select实现聚合?

如果不需要对数据分组,那么可以直接使用select实现聚合;如果有分组操作,需要使用groupBy+agg/select,推荐使用agg

2- first()总结

如果某个DataFrame只有一行数据,并且不使用join来对比数据,那么一般需要使用first()明确指定和第一行进行比较

3-F.col()总结

对于在计算过程中临时产生的字段,需要使用F.col()封装成Column对象?

API/方法:是由DataFrame来调用

函数:需要先通过import pyspark.sql.functions as F导入,使用F调用.

窗口:需要先通过from pyspark.sql import Window导入?