北京网站建设新鸿微信号,做网站的公司哪家有名,简单展示网站模板,丹阳建站推广管理内容目录创建SparkSession对象从CSV文件中读取从JSON文件中读取从Parquet文件中读取从数据列表中创建DataFrame从字典列表中创建DataFrame选择一列选择多列过滤年龄大于30的数据过滤名字为Alice的数据可以使用and、or、not等操作符进行组合查询按照年龄分组并计算平均年龄和最大…
内容目录创建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”, headerTrue, inferSchemaTrue)
从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()
