1. 引言PySpark 是 Spark 的 Python 接口它让我们可以用熟悉的 Python 语法操作分布式数据集。本文从一个最简单的 JSON 读取示例出发逐步带你掌握 DataFrame 的创建、SQL 查询、DSL 风格操作、窗口函数以及炸裂函数等核心用法。无论你是刚接触大数据的新手还是想快速回顾 PySpark 常用 API 的开发者这篇文章都能给你一份可直接运行的实战参考。2. 环境准备与第一个 DataFrame在开始之前请确保你的环境中已经安装好 PySpark并准备好一份测试用的 JSON 数据文件。下面是最基础的读取与展示代码from pyspark.sql import SparkSession if __name__ __main__: # 创建环境 spark SparkSession.builder.master(local[*]).appName(第一个DF).getOrCreate() # 读取数据 dataFrame spark.read.json(rC:\pythoncode\shixun\pyspark_code\day03\data\students.json) # 打印数据 dataFrame.show() # 关闭环境 spark.stop()很多初学者会混淆print(dataFrame)和dataFrame.show()的区别。这里需要特别说明print(dataFrame)只是 Python 层面的打印。DataFrame 在未触发行动算子前只是一个“逻辑计划”数据并没有真正被拉取到你的驱动程序Driver内存里。print 打印出来的是“对象地址”而不是数据。dataFrame.show()这是一个 Action。它不仅会把数据从各个分布式节点拉取汇总到 Driver 端还会对数据进行格式化排版。如果数据量特别大show() 也只会安全地取前 20 条不会把内存撑爆。简单来说show() 是专门为 DataFrame 设计的“格式化展示”方法而 print 只是 Python 对象的默认字符串表示。3. 临时视图与 SQL 风格查询Spark SQL 允许我们用熟悉的 SQL 语法来查询 DataFrame。但 SQL 是字符串字符串本身不知道怎么操作对象只能通过“表名”去找数据。因此我们需要先把 DataFrame 注册成一张临时表from pyspark.sql import SparkSession if __name__ __main__: # 创建环境 spark SparkSession.builder.appName(sql).master(local[*]).getOrCreate() # 读取数据 dataFrame spark.read.json(rC:\pythoncode\shixun\pyspark_code\day03\data\students.json) # 创建临时表因为临时表虽然不存“数据本身”但它存了“获取数据的路线图逻辑计划 # 数据一直都在你的 JSON 文件里临时表只是帮你搭了一座通向这些数据的“查询桥梁”。 dataFrame.createOrReplaceTempView(students) # 查询数据使用 sql 风格 df spark.sql(select * from students) df.show() # 关闭 spark.stop()关于临时表有几个关键点需要理解物理位置不在磁盘在 Driver 端的内存里逻辑映射。生命周期随着 SparkSession 的创建而存在随着 spark.stop() 而消失。内容存的不是数据而是“表名 - 数据来源DataFrame 计算逻辑”的对应关系。换句话说数据一直都在你的 JSON 文件里临时表只是帮你搭了一座通向这些数据的“查询桥梁”。4. DSL 风格操作除了 SQLPySpark 还提供了更贴近编程思维的 DSLDomain Specific Language风格。DSL 直接通过对象引用来操作数据不需要建表映射from pyspark.sql import SparkSession if __name__ __main__: # 创建环境 spark SparkSession.builder.appName(DSL).master(local[*]).getOrCreate() # 读取数据 dataFrame spark.read.json(rC:\pythoncode\shixun\pyspark_code\day03\data\students.json) # 查看数据结构 # dataFrame.printSchema() # 查询学生并使 age 10 dataFrame.select(name, (dataFrame.age 10).alias(age)).show() # alias 类似 as 起别名 # 查询 age 大于 20 的学生信息 dataFrame.where(dataFrame.age 20).show() # 按照年龄分组求 count dataFrame.groupby(age).count().show()这里可以做一个对比总结SQL 是“字符串”字符串不知道怎么操作对象只能通过“表名”去找数据。DSL 是“代码”代码可以直接通过“变量名”操作对象。SQL 作为声明式语言必须通过字符串和元数据Catalog来定位数据DSL 作为命令式/函数式语言直接通过对象引用Reference来操作数据。这就是为什么一个需要建表映射一个直接就能用。5. 窗口函数实战窗口函数Window Function是数据分析中非常强大的工具常用于分组排名、累计求和等场景。下面我们通过一个学生成绩表来体验 row_number 的用法from pyspark.sql import SparkSession, DataFrame, Column, Window, functions from pyspark import SparkConf from pyspark.sql.types import StructType, StructField, StringType, IntegerType if __name__ __main__: spark SparkSession.builder.master(local[*]).appName(windows).getOrCreate() # 梳理表结构 schema schema1 StructType([ StructField(nameid, dataTypeStringType(), nullableFalse), StructField(namename, dataTypeStringType(), nullableFalse), StructField(namescore, dataTypeIntegerType(), nullableFalse), StructField(namecountry, dataTypeStringType(), nullableFalse), ]) df1 spark.read.option(header, True).schema(schema1).csv(C:\pythoncode\shixun\pyspark_code\day03\data\student.csv) # 加个 option 是因为数据里面有表头 df1.printSchema() df1.show() # withColumn通过添加一列或替换具有相同名称的现有列返回一个新的 DataFrame。 df2 df1.withColumn(rnk, functions.row_number().over( Window.partitionBy(df1.country).orderBy(functions.desc(score)) )) df2.where(df2.rnk 3).select(id, df2.name, df2.score, df2.country).show()这里有几个要点为什么 CSV 要手动指定 schema不是“因为读了 CSV 必须写”而是“因为 CSV 本身没结构如果不手动写后续的数值计算、SQL 查询就会出大问题”。读 Parquet、JSON、TXT 都不用建 schema。withColumn用于添加一列或替换具有相同名称的现有列返回一个新的 DataFrame。Window.partitionBy按 country 分组在每个组内按 score 降序排名。6. 炸裂函数与 WordCount 实战炸裂函数explode是处理一对多关系的利器最常见的应用就是 WordCount。下面分别用 SQL 和 DSL 两种方式实现from pyspark import SparkConf from pyspark.sql import SparkSession, functions, DataFrame if __name__ __main__: # 创建环境 spark SparkSession.builder.appName(func).master(local[*]).getOrCreate() # 读取数据 df1 spark.read.text(rC:\pythoncode\shixun\pyspark_code\day03\data\a.txt) # read.text() 这个方法的底层逻辑非常死板 # 它会把文件里的每一行都当成一个字符串 # 强行塞进一个名叫 value 的列里。 # 使用 sql必须建表写多行查询语句的时候要用三引号 # 在这个环境里Spark 已经把所有内置函数如 count、sum、split、explode # 全部预先加载好了 df1.createTempView(t_words) spark.sql( select word, count(*) as c1 from (select explode(split(value, )) word from t_words) as t1 group by word ).show() # 使用 DSL # a.txt 里面的数据 word 不是表头 df2 df1.withColumn(word, functions.explode(functions.split(value, ))) df2.groupby(word).count().show()关于value这个列名需要特别提醒它是.text()方法自带的固有列名。在你的 WordCount 代码里你必须老老实实地用value去引出数据。在 SQL 里value 既没有单引号也没有双引号而在 Python DSL 里我们却经常写F.col(value)或者F.split(value, )。7. 总结本文通过五个实战示例带你走完了 PySpark 最核心的入门路径DataFrame 基础理解 print 与 show 的区别掌握 JSON 读取。临时视图与 SQL理解临时表的本质是“逻辑映射”学会用 SQL 查询。DSL 风格对比 SQL 与 DSL 的差异掌握 select、where、groupby 等操作。窗口函数用 row_number 实现分组排名理解 schema 对 CSV 的重要性。炸裂函数用 explode 实现 WordCount掌握 SQL 与 DSL 两种写法。建议你亲手运行每一段代码观察输出结果再尝试修改参数比如把 show() 改成 show(5)或者把 row_number 换成 rank这样能更快建立对 PySpark 的直觉。后续可以继续深入学习 DataFrame 的 join、UDF 自定义函数、以及性能调优等内容。