Pyspark DataFrame常用操作函数和示例

2024-09-06 19:12

本文主要是介绍Pyspark DataFrame常用操作函数和示例,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

针对类型:pyspark.sql.dataframe.DataFrame

目录

1.打印前几行

1.1 show()函数

1.2 take()函数

2. 读取文件

2.1 spark.read.csv

3. 获取某行某列的值(具体值)

4.查看列名

5.修改列名

5.1 修改单个列名

5.2 修改多个列名

5.2.1 链式调用 withColumnRenamed 方法

5.2.2 使用 selectExpr 方法

6. pandas类型转化为pyspark  pandas

7.选择特定的列,创建一个新的 DataFrame

8.列表套字典格式转化为pyspark DataFrame

9. 根据某列或者某列进行去重

10. pyspark 的两个dataframe合并

11.查看 pyspark dataframe中某列为空的数量

12.删除 pyspark dataframe中 第一行数据

13.pyspark dataframe用空格拼接两列得到新的列

14.将pyspark dataframe 保存到集群(分片)

16.将pyspark dataframe 保存为csv

实际场景1

实际场景2


1.打印前几行

1.1 show()函数

  • show() 函数会将指定数量的行(默认是 20 行)转换为字符串并打印到控制台。
  • 无返回值,直接打印数据到控制台。

用法:

df.show()  # 默认显示前 20 行
df.show(10)  # 显示前 10 行

1.2 take()函数

  • 用于获取 DataFrame 的前 N 行数据,返回一个包含 Row 对象的列表。
  • 返回一个包含 Row 对象的列表。
  • 返回一个包含前 N 行数据的列表,每行数据以 Row 对象的形式存在。你可以通过索引访问这些行,并进一步处理它们。
rows = df.take(5)  # 获取前 5 行数据
for row in rows:print(row)

2. 读取文件

2.1 spark.read.csv

df = spark.read.csv(path, sep="\t", header=False, inferSchema=True).toDF('id','time','label','feature')
  • inferSchema=True: 让 Spark 自动推断 CSV 文件中各列的数据类型
  • toDF: 这是一个 DataFrame 方法,用于为 DataFrame 的列指定新的列名。

3. 获取某行某列的值(具体值)

直接获取 DataFrame 的特定行(例如第 562962 行)并不是一个高效的操作,因为 Spark 是

分布式计算框架,数据被分割并在多个节点上并行处理

# 获取第一行
first_row = df.first()# 获取 feature 列的值
first_row['feature_1']
# 获取前两行
rows = df.take(2)# 获取第二行
second_row = rows[1]# 获取 feature 列的值
second_row['feature']

4.查看列名

df.columns

5.修改列名

5.1 修改单个列名

# 修改列名
df_renamed = df.withColumnRenamed("name", "new_name")

5.2 修改多个列名

5.2.1 链式调用 withColumnRenamed 方法

# 修改多个列名
df_renamed = df.withColumnRenamed("id", "new_id").withColumnRenamed("name", "new_name")

5.2.2 使用 selectExpr 方法

注意:使用 selectExpr 方法时,最后只会得到你修改的列,即,在函数参数中的列名

如果想使用该方法时,还想要原来的列名,就直接, 在参数中加入,"原列名 as 原列名"

# 使用 selectExpr 修改列名
df_renamed = df.selectExpr("id as new_id", "name as new_name")

6. pandas类型转化为pyspark  pandas

pandas.core.frame.DataFrame 类型转化为 pyspark.sql.dataframe.DataFrame
# 将 Pandas DataFrame 转换为 PySpark DataFrame
pyspark_df = spark.createDataFrame(pandas_df)

7.选择特定的列,创建一个新的 DataFrame

# 选择某几列并创建新的 DataFrame
new_df = df.select("name", "age")

8.列表套字典格式转化为pyspark DataFrame

# 示例列表套字典
data = [{"name": "Alice", "age": 25, "id": 1},{"name": "Bob", "age": 30, "id": 2},{"name": "Cathy", "age": 35, "id": 3}
]# 将列表套字典转换为 PySpark DataFrame
df = spark.createDataFrame(data)# 显示 DataFrame
df.show()

9. 根据某列或者某列进行去重

duyuv3_1_df = duyuv3_1_df.dropDuplicates(['md5', 'time', 'label'])

10. pyspark 的两个dataframe合并

merged_v3_1_df = duyuv3_1_df.join(passid_md5_df, on=['md5'], how='left')

11.查看 pyspark dataframe中某列为空的数量

null_passid_count = merged_v3_1_df.filter(merged_v3_1_df['passid'].isNull()).count()
print(f"passid is null:{null_passid_count}")

12.删除 pyspark dataframe中 第一行数据

data_df = data_df.filter(col("_c0") != data_df.first()[0])
  • data_df.first(): 获取 DataFrame 的第一行数据。

  • col("_c0"): 获取 DataFrame 的第一列(默认情况下,Spark 会将 CSV 文件的列命名为 _c0_c1_c2, ...)。

  • data_df.filter(col("_c0") != data_df.first()[0]): 过滤掉第一行数据。这里假设第一行的第一列值与后续行的第一列值不同,因此通过比较第一列的值来过滤掉第一行。

13.pyspark dataframe用空格拼接两列得到新的列

# 拼接特征列replace_df = replace_df.withColumn('merged_feature',when(col('featurev3').isNotNull() & col('feature_v3_1').isNotNull(),concat_ws(' ', col('featurev3'), col('feature_v3_1'))).when(col('featurev3').isNotNull(), col('featurev3')).when(col('feature_v3_1').isNotNull(), col('feature_v3_1')).otherwise(lit('')))

14.将pyspark dataframe 保存到集群(分片)

save_path =f'afs://szth.afs.****.com:9902/user/fsi/duyuv3_1_feature/result_duyuv3_1/'
rdd_combined_duyuv3_1 = feature_cgc.rdd.map(lambda x: "\t".join(x))
rdd_combined_duyuv3_1.saveAsTextFile(save_path)

16.将pyspark dataframe 保存为csv

output_path = "afs://szth.afs.baidu.com:9902/user/fsi/tongweiwei/duyuv3_1_feature/data.csv"
final_df.write.csv(output_path, header=True, mode="overwrite")

实际场景1

对某列的值进行按照空格进行切割,然后在对切割后的数据判断冒号前面的字符串判断是否在某一个字符串中,如果在则去除掉

from pyspark.sql.types import StringType
from pyspark.sql.functions import concat_ws, col, when, lit, udfdef filter_feature(feature_str, filter_list):parts = feature_str.split()filtered_parts = [part for part in parts if str(part.split(':')[0]) not in filter_list.split(',')]return ' '.join(filtered_parts)filter_feature_udf = udf(filter_feature, StringType())df = duyuv3_df.withColumn("featurev3", filter_feature_udf(col("featurev3"), lit(duyuv3_str)))

实际场景2

对某列的值,按照空格进行切割后,按照冒号前面的进行排序

from pyspark.sql.types import StringType
from pyspark.sql.functions import concat_ws, col, when, lit, udfdef sort_by_number(value):# 将输入字符串按空格分割为列表value = value.strip().split(" ")value_list = []# 遍历列表中的每个元素,提取数字部分并排序for val in value:try:feat_num = int(val.split(":")[0])value_list.append(val)except:continuesorted_pairs = sorted(value_list, key=lambda x: int(x.split(":")[0]))return " ".join(sorted_pairs)sort_by_number_udf = udf(sort_by_number, StringType())feature_cgc = replace_df.withColumn("sorted_feat",sort_by_number_udf(replace_df["merged_feature"]))

这篇关于Pyspark DataFrame常用操作函数和示例的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1142872

相关文章

Java使用Curator进行ZooKeeper操作的详细教程

《Java使用Curator进行ZooKeeper操作的详细教程》ApacheCurator是一个基于ZooKeeper的Java客户端库,它极大地简化了使用ZooKeeper的开发工作,在分布式系统... 目录1、简述2、核心功能2.1 CuratorFramework2.2 Recipes3、示例实践3

Spring Boot 3.4.3 基于 Spring WebFlux 实现 SSE 功能(代码示例)

《SpringBoot3.4.3基于SpringWebFlux实现SSE功能(代码示例)》SpringBoot3.4.3结合SpringWebFlux实现SSE功能,为实时数据推送提供... 目录1. SSE 简介1.1 什么是 SSE?1.2 SSE 的优点1.3 适用场景2. Spring WebFlu

Java利用JSONPath操作JSON数据的技术指南

《Java利用JSONPath操作JSON数据的技术指南》JSONPath是一种强大的工具,用于查询和操作JSON数据,类似于SQL的语法,它为处理复杂的JSON数据结构提供了简单且高效... 目录1、简述2、什么是 jsONPath?3、Java 示例3.1 基本查询3.2 过滤查询3.3 递归搜索3.4

springboot security快速使用示例详解

《springbootsecurity快速使用示例详解》:本文主要介绍springbootsecurity快速使用示例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝... 目录创www.chinasem.cn建spring boot项目生成脚手架配置依赖接口示例代码项目结构启用s

golang 日志log与logrus示例详解

《golang日志log与logrus示例详解》log是Go语言标准库中一个简单的日志库,本文给大家介绍golang日志log与logrus示例详解,感兴趣的朋友一起看看吧... 目录一、Go 标准库 log 详解1. 功能特点2. 常用函数3. 示例代码4. 优势和局限二、第三方库 logrus 详解1.

SpringBoot实现MD5加盐算法的示例代码

《SpringBoot实现MD5加盐算法的示例代码》加盐算法是一种用于增强密码安全性的技术,本文主要介绍了SpringBoot实现MD5加盐算法的示例代码,文中通过示例代码介绍的非常详细,对大家的学习... 目录一、什么是加盐算法二、如何实现加盐算法2.1 加盐算法代码实现2.2 注册页面中进行密码加盐2.

Redis 中的热点键和数据倾斜示例详解

《Redis中的热点键和数据倾斜示例详解》热点键是指在Redis中被频繁访问的特定键,这些键由于其高访问频率,可能导致Redis服务器的性能问题,尤其是在高并发场景下,本文给大家介绍Redis中的热... 目录Redis 中的热点键和数据倾斜热点键(Hot Key)定义特点应对策略示例数据倾斜(Data S

Android Kotlin 高阶函数详解及其在协程中的应用小结

《AndroidKotlin高阶函数详解及其在协程中的应用小结》高阶函数是Kotlin中的一个重要特性,它能够将函数作为一等公民(First-ClassCitizen),使得代码更加简洁、灵活和可... 目录1. 引言2. 什么是高阶函数?3. 高阶函数的基础用法3.1 传递函数作为参数3.2 Lambda

JavaScript Array.from及其相关用法详解(示例演示)

《JavaScriptArray.from及其相关用法详解(示例演示)》Array.from方法是ES6引入的一个静态方法,用于从类数组对象或可迭代对象创建一个新的数组实例,本文将详细介绍Array... 目录一、Array.from 方法概述1. 方法介绍2. 示例演示二、结合实际场景的使用1. 初始化二

C#中的 StreamReader/StreamWriter 使用示例详解

《C#中的StreamReader/StreamWriter使用示例详解》在C#开发中,StreamReader和StreamWriter是处理文本文件的核心类,属于System.IO命名空间,本... 目录前言一、什么是 StreamReader 和 StreamWriter?1. 定义2. 特点3. 用