spark- Dataframe基本操作-查询

前言： DataFrame需要重点关注的是如何取使用，利用DataFrame来解决实际业务中的数据问题。熟练掌握DataFrame的各种api的使用就像相当于手握一把非常重要且高效的利器，实际数据工程可以大量的被用到。 本文是对DataFrame概念和过去实际工作中用到过的DataFrame中关于数据查看的一个回忆性总结。

**DataFrame概念(了解即可） **

• DataFrame是从Spark 1.3版本开始引入的。从Spark 2.0开始，DataFrame和DataSet的API合并在一起，实现了跨库统一成为一套API。这样，开发人员的学习成本就降低了。只需要学习一个High Level的、类型安全的DataSet API就可以了
• 通过DataFrame可以简化Spark程序的开发，让Spark处理结构化数据变得更简单。DataFrame可以使用SQL的方式来处理数据。例如：业务分析人员可以基于编写Spark SQL来进行数据开发，而不仅仅是Spark开发人员。
• DataFrame和RDD有一些共同点，也是不可变的分布式数据集。但与RDD不一样的是，DataFrame是有schema的，有点类似于关系型数据库中的表，每一行的数据都是一样的，因为。有了schema，这也表明了DataFrame是比RDD提供更高层次的抽象。
• DataFrame支持各种数据格式的读取和写入，例如：CSV、JSON、AVRO、HDFS、Hive表。
• DataFrame底层使用Catalyst进行优化。

DataFrame的优点(了解即可）

• 因为DataFrame是有统一的schema的，所以序列化和反序列无需存储schema。这样节省了一定的空间。
• DataFrame存储在off-heap（堆外内存）中，由操作系统直接管理（RDD是JVM管理），可以将数据直接序列化为二进制存入off-heap中。操作数据也是直接操作off-heap。

查询相关操作

**样例数据，简单建一个DataFrame **

sc = spark.sparkContext 
# 创建一个SparkDataFrame
rdd = sc.parallelize([("Sam", 28, 88, "M"),
                      ("Flora", 28, 90, "F"),
                      ("Run", 1, 60, None),
                      ("Peter", 55, 100, "M"),
                      ("Mei", 54, 95, "F")])
test_exp_data = rdd.toDF(["name", "age", "score", "sex"]) # 创建DataFrame的一种方式，用rdd.toDF 来建

.show() 显示头n 行

DataFrame.show( num) 以表格的形式显示头num行，函数返回值为NoneType 。

tmp = test_exp_data.show(5)
print(type(tmp)) # 返回结果为 <class 'NoneType'>.

输出：
+-----+---+-----+----+
| name|age|score| sex|
+-----+---+-----+----+
|  Sam| 28|   88|   M|
|Flora| 28|   90|   F|
|  Run|  1|   60|null|
|Peter| 55|  100|   M|
|  Mei| 54|   95|   F|
+-----+---+-----+----+

<class 'NoneType'>

.printSchema() 打印表概要，

以树的形式显示表中各个列的数据结构 ,返回类型为None

print(test_exp_data.printSchema())

输出：

root
 |-- name: string (nullable = true)
 |-- age: long (nullable = true)
 |-- score: long (nullable = true)
 |-- sex: string (nullable = true)

也可以直接输出，不用括号，也不会报错如

print(test_exp_data.printSchema)

.count() 查询统计表中数据行数

返回一个int类型结果值

print(test_exp_data.count())
tmp  = test_exp_data.count()
print(type(tmp)) # <class 'int'>

输出：

5
<class 'int'>

### 去重
rdd = sc.parallelize([("Sam", 28, 88.52, "M"),
                      ("Flora", 28, 90.55, "F"),
                      ("Flora", 28, 90.55, "F"),
                      ("Flora", 45, 99.5, "F"),
                      ("Mey", 1, None, "M"),
                      ("Chery", 7, 80.23, "F")])
test_exp_data = rdd.toDF(["name", "age", "score", "sex"])
#test_exp_data.show()
### .distinct()对数据集去重，去掉重复的行
test_exp_data.distinct().show()

结果：

+-----+---+-----+---+
| name|age|score|sex|
+-----+---+-----+---+
|  Sam| 28|88.52|  M|
|Flora| 28|90.55|  F|
|Flora| 45| 99.5|  F|
|  Mey|  1| null|  M|
|Chery|  7|80.23|  F|
+-----+---+-----+---+

.dropDuplicates 按指定列去重,#对于改列的不同值，只保留在第一次出现的行
test_exp_data.dropDuplicates(['sex']).show()

结果：

+-----+---+-----+---+
| name|age|score|sex|
+-----+---+-----+---+
|Flora| 28|90.55|  F|
|  Sam| 28|88.52|  M|
+-----+---+-----+---+

.limit() 获取前n行并得到新的DataFrame

limit

方法获取指定DataFrame的前n行记录，得到一个新的DataFrame对象。

tmp = test_exp_data.limit(4)
print(type(tmp)) # DataFrame
tmp.show(2)

输出

<class 'pyspark.sql.dataframe.DataFrame'>
+-----+---+-----+---+
| name|age|score|sex|
+-----+---+-----+---+
|  Sam| 28|   88|  M|
|Flora| 28|   90|  F|
+-----+---+-----+---+

.collect() 获取所有数据到数组

不同于前面的

show

方法，这里的

collect

方法会将

DF

中的所有数据都获取到，并返回一个list。

print(test_exp_data.collect())
tmp= test_exp_data.collect()
print(type(tmp))

输出：

[Row(name='Sam', age=28, score=88, sex='M'), Row(name='Flora', age=28, score=90, sex='F'), Row(name='Run', age=1, score=60, sex=None), Row(name='Peter', age=55, score=100, sex='M'), Row(name='Mei', age=54, score=95, sex='F')]
<class 'list'>

.head(num) 以及.take(num) 获取前num行记录

两个函数功能都是获取前num行记录汇总到一个list中。list中的每个元素都是一个row结构。

tmp = test_exp_data.head(3)
print(tmp)
print(type(tmp))  ## list 
print(type(tmp[0]))

tmp = test_exp_data.take(3)
print(tmp)
print(type(tmp))  # <class 'list'>
print(type(tmp[0]))

输出：

[Row(name='Sam', age=28, score=88, sex='M'), Row(name='Flora', age=28, score=90, sex='F'), Row(name='Run', age=1, score=60, sex=None)]
<class 'list'>
<class 'pyspark.sql.types.Row'>

另一个输出完全一模一样。

.sample() 采样

df.sample(fraction=float_num) ，按fraction比例对行进行随机采样

print(test_exp_data.count())
samp = test_exp_data.sample(fraction=0.5)
print(samp.count())

输出：

5
3

.select() 选择列数据

选择的列数据构成一个新的dataframe返回。 可以有如下三种表示方法来选 。且都可以进行数据运算。

test_exp_data.select("name").show(1)
test_exp_data.select(test_exp_data.name).show(1)
test_exp_data.select(test_exp_data['name']).show(1)

都是输出：

+----+
|name|
+----+
| Sam|
+----+

选择多列的表示方式。注意第一种如果是单独只有列名时需要用list拼起来

test_exp_data.select(["name",'score']).show(1) # 选择多个列，可以是由string类型列名组成的list。注意要用list表示
test_exp_data.select(test_exp_data.name,test_exp_data.score).show(1) #也可以是df中逐个带着列名
test_exp_data.select(test_exp_data["name"],test_exp_data["score"]).show(1) #也可以是df中逐个带着列名

都是输出

+----+-----+
|name|score|
+----+-----+
| Sam|   88|
+----+-----+

另外，可以基于某列直接进行数据运算(数据类型要为非string)

test_exp_data.select(test_exp_data["name"],test_exp_data["score"]+1).show(1) # df中逐个带着列的数据还可以进行运算操作
test_exp_data.select(test_exp_data.name,test_exp_data.score+1).show(1)

.selectExpr() 选择数据

selectExpr() 也可以选择数据并进行更加复杂的数据处理，直接选择的语句中就处理了。selectExpr方法本质与select方法中使用expr函数是一样的，可以直接对指定字段调用UDF函数，或者指定别名等。传入

String

类型参数，得到DataFrame对象。

如下实例：里面用as 重命名、用round四舍五入并指定保留小数位；

# 创建一个SparkDataFrame
rdd = sc.parallelize([("Sam", 28, 88.52, "M"),
                      ("Flora", 28, 90.55, "F"),
                      ("Run", 1, 60, None),
                      ("Peter", 55, 100, "M"),
                      ("Mei", 54, 95, "F")])
test_exp_data = rdd.toDF(["name", "age", "score", "sex"])
test_exp_data.selectExpr('name as new_name','round(score)','round(score,1)').show(3) # 第一个处理是起新命名，第二个是直接调用round函数进行四舍五入处理默认不保留小数位，第三个是保留一位小数

输出：

+--------+---------------+---------------+
|new_name|round(score, 0)|round(score, 1)|
+--------+---------------+---------------+
|     Sam|           89.0|           88.5|
|   Flora|           91.0|           90.5|
|     Run|           null|           null|
+--------+---------------+---------------+

.unionByName 两个DataFrame根据列名进行合并。

rdd = sc.parallelize([("Sam", 28, 88.52, "M"),
("Flora", 28, 90.55, "F"),
("Chery", 7, 80.23, "F")])
test_exp_data = rdd.toDF(["name", "age", "score", "sex"])
test_exp_data.show()#df1

+-----+---+-----+---+
| name|age|score|sex|
+-----+---+-----+---+
|  Sam| 28|88.52|  M|
|Flora| 28|90.55|  F|
|Chery|  7|80.23|  F|
+-----+---+-----+---+

rdd = sc.parallelize([("Sam", 28, 88.52, "M"),
("Flora", 28, 90.55, "F"),
("Chery", 7, 80.23, "F")])
test_exp_data = rdd.toDF(["name", "age", "score", "sex"])
test_exp_data.show()# df2

+---+------+-----+---+
|age|  name|score|sex|
+---+------+-----+---+
|  7|Chery2|80.23|  F|
| 28|  Sam2|88.52|  M|
+---+------+-----+---+

## 两个df 根据列名进行合并
merge = test_exp_data.unionByName(test2).show()

结果：

+------+---+-----+---+
|  name|age|score|sex|
+------+---+-----+---+
|   Sam| 28|88.52|  M|
| Flora| 28|90.55|  F|
| Chery|  7|80.23|  F|
|Chery2|  7|80.23|  F|
|  Sam2| 28|88.52|  M|
+------+---+-----+---+

.describe() 查看数据的统计分布

返回关于列的统计分布(个数统计count、数值均值mean、标准差stddev、最小值min、最大值max），结果展现形式为dataframe类型，可以用show查看。注意字符串类型的列并不会报错！且Min 和max会按照字典序进行排序 。

对于pyspark的dataframe的describe函数并不能统计出对应的分位数，如果想统计对应的百分位数，可以转换为pandas的dataframe后再通过参数percentiles去统计。

rdd = sc.parallelize([("Sam", 28, 88.52, "M"),
                      ("Flora", 28, 90.55, "F"),
                      ("Mey", 1, None, "M"),
                      ("Chery", 7, 80.23, "F")])
test_exp_data = rdd.toDF(["name", "age", "score", "sex"])
test_exp_data.describe().show()

输出：

+-------+-----+------------------+-----------------+----+
|summary| name|               age|            score| sex|
+-------+-----+------------------+-----------------+----+
|  count|    4|                 4|                3|   4|
|   mean| null|              16.0|86.43333333333334|null|
| stddev| null|14.071247279470288|5.467287566365362|null|
|    min|Chery|                 1|            80.23|   F|
|    max|  Sam|                28|            90.55|   M|
+-------+-----+------------------+-----------------+----+

也可以选择某列单独进行统计

test_exp_data.select('score').describe().show()

想统计百分位数，.toPandas().describe(percentiles=[....])

test_exp_data.toPandas().describe(percentiles=[0.0,0.01,0.2,0.5,0.90,1.0])

             age      score
count   4.000000   3.000000
mean   16.000000  86.433333
std    14.071247   5.467288
min     1.000000  80.230000
0%      1.000000  80.230000
1%      1.180000  80.395800
20%     4.600000  83.546000
50%    17.500000  88.520000
90%    28.000000  90.144000
100%   28.000000  90.550000
max    28.000000  90.550000

关于百分位数的概念：第p百分位数（p%分位数）值满足：有p%的观察值小于或等于该值，至少有（100-p）%的观察值大于或等于该值。注意0%分位数是这组值中的最小值，100%分位数是这组值中的最大值。

分位数的概念如

.summary统计数据分布

功能类似describe，多了25%，50%，75%分位数，结果返回仍然是一个dataframe

test_exp_data.summary().show()

结果显示：

+-------+------------------+-----------------+
|summary|               age|            score|
+-------+------------------+-----------------+
|  count|                 4|                3|
|   mean|              16.0|86.43333333333334|
| stddev|14.071247279470288|5.467287566365362|
|    min|                 1|            80.23|
|    25%|                 1|            80.23|
|    50%|                 7|            88.52|
|    75%|                28|            90.55|
|    max|                28|            90.55|
+-------+------------------+-----------------+

withColumn中，df.column.cast('type')数据类型转换

如： flash_res = flash_res.withColumn("label_float",flash_res.label.cast('float'))