- 1Oracle又裁员！_oracle裁员
- 2KeeperErrorCode = ConnectionLoss 问题解决
- 37 Series FPGAs Integrated Block for PCI Express IP核基本模式配置详解
- 4If you want an embedded database (H2, HSQL or Derby), please put it on the classpath.
- 5python数组求和函数_python数据分析之Numpy数据库第三期数组的运算
- 6第三阶段应用层——1.7 数码相册—电子书（2）—编写通用的Makefile_在主目录下创建dianzi1和dianzi2两个文件夹,在dianzi1文件夹中创建file1 fi
- 7＜网络安全＞《70 微课堂＜第11课 0day、1day、nday（精华版）＞》_0day、1day、n day
- 8计算机领域顶级会议归纳（计算机视觉CV、机器学习ML、人工智能AI）_wacv
- 9RabbitMQ消费者Consumer实现_java rabbitmq consumer
- 10CKA 06_Kubernetes 工作负载与调度 Pod 管理 yaml 资源清单 标签 Pod 生命周期 容器探针_kubernetes工作负载 yaml
spark- Dataframe基本操作-查询_spark dataframe
赞
踩
前言： DataFrame需要重点关注的是如何取使用，利用DataFrame来解决实际业务中的数据问题。熟练掌握DataFrame的各种api的使用就像相当于手握一把非常重要且高效的利器，实际数据工程可以大量的被用到。 本文是对DataFrame概念和过去实际工作中用到过的DataFrame中关于数据查看的一个回忆性总结。
目录
.distinct() 行去重和.dropDuplicates按指定列去重
.head(num) 以及.take(num) 获取前num行记录
.unionByName 两个DataFrame根据列名进行合并。
withColumn中，df.column.cast('type')数据类型转换
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'>
.distinct() 行去重和.dropDuplicates按指定列去重
.distinct() 去掉重复的行，只保留第一次出现的行，得到不同的行，返回类型为DataFrame
- ### 去重
- 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'))
