当前位置: 首页 > news >正文

Spark的DataFrame使用

内容目录

  • 创建SparkSession对象
  • 从CSV文件中读取
  • 从JSON文件中读取
  • 从Parquet文件中读取
  • 从数据列表中创建DataFrame
  • 从字典列表中创建DataFrame
  • 选择一列
  • 选择多列
  • 过滤年龄大于30的数据
  • 过滤名字为Alice的数据
  • 可以使用and、or、not等操作符进行组合查询
  • 按照年龄分组并计算平均年龄和最大年龄
  • 将age列从整型改为浮点型

Spark的DataFrame是一种类似于表格的数据结构,可以进行各种数据处理操作。使用DataFrame可以很方便地处理结构化数据(如CSV、JSON、Parquet等格式的数据)。

DataFrame是Spark SQL中的概念,因此需要首先引入Spark SQL库:

from pyspark.sql import SparkSession

创建SparkSession对象

spark = SparkSession.builder.appName(“example”).getOrCreate()
创建DataFrame
从文件中读取
可以使用SparkSession的read方法从不同类型的文件中读取数据,例如:

从CSV文件中读取

df = spark.read.csv(“file.csv”, header=True, inferSchema=True)

从JSON文件中读取

df = spark.read.json(“file.json”)

从Parquet文件中读取

df = spark.read.parquet(“file.parquet”)
直接创建
也可以使用SparkSession的createDataFrame方法直接创建DataFrame,例如:

从数据列表中创建DataFrame

data = [(“Alice”, 25), (“Bob”, 30), (“Charlie”, 35)]
df = spark.createDataFrame(data, [“name”, “age”])

从字典列表中创建DataFrame

data = [{“name”: “Alice”, “age”: 25}, {“name”: “Bob”, “age”: 30}, {“name”: “Charlie”, “age”: 35}]
df = spark.createDataFrame(data)
DataFrame的基本操作
显示DataFrame
使用show方法可以将DataFrame中的数据显示出来,例如:

df.show()
查看DataFrame的结构
使用printSchema方法可以查看DataFrame的结构,例如:

df.printSchema()
选择列
使用select方法可以选择一列或多列,例如:

选择一列

df.select(“name”).show()

选择多列

df.select(“name”, “age”).show()
过滤数据
使用filter方法可以根据条件过滤数据,例如:

过滤年龄大于30的数据

df.filter(df[“age”] > 30).show()

过滤名字为Alice的数据

df.filter(df[“name”] == “Alice”).show()

可以使用and、or、not等操作符进行组合查询

df.filter((df[“age”] > 30) & (df[“name”] != “Alice”)).show()
分组聚合
使用groupBy方法可以对数据进行分组聚合操作,例如:

按照年龄分组并计算平均年龄和最大年龄

df.groupBy(“age”).agg({“age”: “avg”, “age”: “max”}).show()
DataFrame的类型转换
更改列名
使用withColumnRenamed方法可以更改列名,例如:

df = df.withColumnRenamed(“name”, “person_name”)
df.show()
更改列类型
使用withColumn方法可以更改列的数据类型,例如:

from pyspark.sql.functions import col

将age列从整型改为浮点型

df = df.withColumn(“age”, col(“age”).cast(“float”))
df.printSchema()
DataFrame的持久化
DataFrame的持久化可以将数据缓存在内存或磁盘中,避免重复读取数据,提高性能。

使用cache方法可以将DataFrame缓存在内存中,例如:

df.cache()
使用persist方法可以将DataFrame缓存在磁盘中,例如:

df.persist()

http://www.lryc.cn/news/33294.html

相关文章:

  • 【Flutter】入门Dart语言:操作符的基本用法
  • Linux线程概念
  • C#基础教程10 方法
  • java高性能并发计数器之巅峰对决
  • HTTPS简介
  • K-means聚类
  • 04-SQL基础(表管理,约束,多表连接,子查询)
  • 统计学 一元线性回归
  • 【软件开发】基于PyQt5开发的标注软件
  • CSS3新特性
  • 35 openEuler搭建repo(yum)服务器-创建、更新本地repo源
  • 【三.项目引入axios、申明全局变量、设置跨域】
  • 启动u盘还原成普通u盘(Windows Diskpart)
  • 深入理解机器学习——偏差(Bias)与方差(Variance)
  • 分布式新闻项目实战 - 13.项目部署_持续集成(Jenkins) ^_^ 完结啦 ~
  • Linux c/c++技术方向分析
  • JavaScript 高级3 :函数进阶
  • 【项目】Java树形结构集合分页,java对list集合进行分页
  • java.lang.IllegalArgumentException: itemView may not be null
  • [ 攻防演练演示篇 ] 利用 shiro 反序列化漏洞获取主机权限
  • 达人合作加持品牌布局,3.8女神玩转流量策略!
  • 观点丨Fortinet谈ChatGPT火爆引发的网络安全行业剧变
  • 工业企业用电损耗和降损措施研究
  • 高并发、高性能、高可用
  • 剑指 Offer 62. 圆圈中最后剩下的数字
  • 概率论小课堂:高斯分布(正确认识大概率事件)
  • 剑指 Offer 43. 1~n 整数中 1 出现的次数
  • 如何成为程序员中的牛人/高手?
  • 云原生时代顶流消息中间件Apache Pulsar部署实操之轻量级计算框架
  • 数据结构刷题(十九):77组合、216组合总和III