pyspark 入门Spark SQL 练习的正确打开方式很多朋友学 PySpark一上来就啃 RDD 算子什么 map、flatMap、reduceByKey绕来绕去最后写个 wordcount 就觉得自己入门了。但实际一回到真实业务面对一堆 JSON、CSV、数据库表你发现 RDD 那套东西写起来既啰嗦又不直观。我自己的经验是如果你是新手与其在 RDD 里挣扎不如直接从 Spark SQL 切进去用 DataFrame、用 SQL 语法去理解分布式计算反而上手更快。这篇文章就是围绕 pyspark 里的 Spark SQL 练习来写的。我会从环境准备、DataFrame 构建、SQL 操作、常见的坑到最后一个完整练习案例把实际干活时要用的东西都过一遍。适合刚开始学 PySpark、想做数据分析或数仓 ETL、又不想被概念绕晕的朋友。看完你就能在本地把 Spark 跑起来用 DataFrame API 和 SQL 完成一套完整的数据分析流程。1. 环境准备先把 Spark 跑起来1.1 本地安装与配置说实话本地装 Spark 没有想象中那么复杂但版本匹配问题确实坑过不少人。我的建议是先确定一个组合Java 8 或 Java 11Spark 3.3 或 3.4Python 3.8 以上。这个组合我实测过多次兼容性最省心。步骤大概是这样的去官网下载 Spark 的预编译包选带有 Hadoop 客户端的版本比如spark-3.4.1-bin-hadoop3。解压到一个不含空格的路径比如D:\spark或/opt/spark。配置环境变量SPARK_HOME并把$SPARK_HOME/bin加入PATH。安装 PySparkpip install pyspark。注意这里装的是 Python 端的库它会自动寻找SPARK_HOME或者自己携带一个内置 Spark。有一个特别容易踩的坑是 Python 解释器不一致。你在命令行输入python用的是某个版本但 Spark 启动时可能去找系统默认的 Python结果两边版本不一样一执行就报Py4JJavaError或者Python worker failed to connect back。解决办法是显式设置PYSPARK_PYTHON和PYSPARK_DRIVER_PYTHON让 Spark 知道你用的是哪个 Python。在 Linux 或 Mac 上可以在~/.bashrc里加export SPARK_HOME/opt/spark export PATH$SPARK_HOME/bin:$PATH export PYSPARK_PYTHONpython3 export PYSPARK_DRIVER_PYTHONpython3Windows 上把路径改成你的实际安装位置然后在系统环境变量里加同样的内容改完记得重启终端。1.2 两种运行方式怎么选本地练习环境配好之后你通常会遇到两种启动方式一种是在 Jupyter Notebook 里跑另一种是写脚本用spark-submit跑。这两种方式各有各的用处。Jupyter 模式适合学习和调试因为你可以一段一段地运行随时看 DataFrame 的内容查结果非常方便。在 Notebook 里需要加两行代码import findspark findspark.init()如果没装 findspark直接用pip install findspark就行。它的作用就是帮你在 Python 进程里自动设置 Spark 相关的环境变量省得每次手动去配SPARK_HOME。脚本模式适合批量跑任务比如你写了一个analysis.py然后用spark-submit analysis.py提交到集群或者本地运行。这种方式的好处是贴近生产环境而且没有 Notebook 那种状态缓存的问题每跑一次都是全新的会话。我个人的建议是新手先用 Jupyter 练操作等逻辑成熟了再封装成脚本用spark-submit跑一遍感受一下两种模式的区别。很多人在 Notebook 里跑得好好的一上脚本就各种报错多半是因为环境变量没设置并不是代码本身有问题。对比项Jupyter Notebookspark-submit 脚本适合场景学习、调试、交互分析批量处理、生产任务启动成本需要 findspark 或手动设置环境依赖环境变量直接提交数据查看方便可实时打印 DataFrame需要自己打印或写日志资源管理长驻进程容易保留旧状态每次全新启动无状态残留2. 从零构建 DataFrameSpark SQL 的第一块基石2.1 从 RDD 转换我知道很多人学 PySpark 入门时写过 RDD所以这里先讲一下怎么把 RDD 转成 DataFrame。这是理解 Spark 数据模型的一个关键跳板。from pyspark.sql import SparkSession spark SparkSession.builder.appName(demo).getOrCreate() # 先造一个 RDD rdd spark.sparkContext.parallelize([ (Alice, 25, engineer), (Bob, 30, designer), (Cathy, 28, manager) ]) # 方式一toDF 指定列名 df1 rdd.toDF([name, age, job]) df1.show()toDF是最懒的方法只要 RDD 里的元素是元组或者 Row 对象就可以直接转。但这里有个隐藏问题RDD 是没有 schema 的里面的数据全是 Python 对象Spark 在转换时要自己去推断类型。如果数据量很大这个推断过程也会消耗时间。所以我更推荐下面这种方式先手动定义 schema再创建 DataFrame。from pyspark.sql.types import StructType, StructField, StringType, IntegerType schema StructType([ StructField(name, StringType(), True), StructField(age, IntegerType(), True), StructField(job, StringType(), True) ]) df2 spark.createDataFrame(rdd, schema)指定 schema 的好处很明显类型可控、性能更好、不会出现“推断出个 bigint 还是 long 的差别”这种奇怪问题。后面读文件时同样建议手动指定 schema尤其是生产环境。2.2 从文件读取CSV、JSON、Parquet实际练习中大部分数据都来自文件。PySpark 读取文件非常直接但每个格式都有各自的细节。先看 CSV最常见也最容易踩坑df spark.read.csv( data.csv, headerTrue, inferSchemaTrue, sep, )headerTrue表示第一行是表头inferSchemaTrue表示让 Spark 自动推断字段类型。练习阶段这样写没问题但如果你发现读进来某些列类型不对比如日期变成了字符串或者数字列有 null就要考虑手动指定 schema 了。另一种情况是 CSV 里某些字段本身包含逗号比如张 三,工程师这种带引号的格式需要设置quote参数否则会解析错位。JSON 读取就更简单了df_json spark.read.json(data.json)但 JSON 有个天然的问题嵌套结构。一行 JSON 里可能有个address字段它本身是个字典。读进来之后这个字段会变成 StructType查询时要通过df[address.city]这样点引用的方式去访问。新手很容易在这里被绕晕建议先把数据拉平或者用explode把嵌套数组拆开。Parquet 是列式存储格式也是 Spark 最推荐的存储格式。读取非常快而且自带压缩和 schema 信息df_parquet spark.read.parquet(data.parquet)如果只是练习不一定要特意生成 Parquet 文件但你要知道Spark 在读写 Parquet 时很多优化会自动生效比如列裁剪和谓词下推。用 CSV 做练习体验不到这些特性所以我建议后期一定要试试 Parquet。2.3 从数据库读取JDBC这事练习阶段可能用不上但既然做 pySpark免不了有一天要直接从数据库读表。JDBC 读取要做几件事首先确保数据库驱动 jar 已经让 Spark 能看到其次连接串要写对。df_db spark.read.format(jdbc) \ .option(url, jdbc:mysql://localhost:3306/test) \ .option(driver, com.mysql.cj.jdbc.Driver) \ .option(user, root) \ .option(password, 123456) \ .option(dbtable, user_info) \ .load()dbtable 那里可以写子查询比如(select id, name from user_info where age 18) tmp这样 Spark 只拉符合条件的行过来而不是把整张表全部读进来再做过滤这是 JDBC 读取时一个非常实用的性能优化小技巧。注意连接数据库前要把驱动 jar 放到 Spark 的 jars 目录里或者在提交任务时用--jars指定否则会报ClassNotFound错误。2.4 Schema 的两种定义方式前面提到过 schema 可以自动推断也可以手动定义。这里展开说下因为很多练习做着做着就卡在 schema 上。自动推断适合第一次探索数据的时候你想快速看看字段大概是什么类型。缺点是不稳定数据一多容易推断出意想不到的类型比如把字符串“12345”推断成 long。手动定义 schema 适合在明确字段结构的情况下使用尤其是当字段很多、类型明确或者来源是第三方数据接口时。手动定义 schema 还有另一个好处可以指定字段是否允许为 null。比如有的字段你觉得比较关键不希望它为空就在StructField里把第三参数设为False。这样 Spark 在写入或解析时就会直接报错而不是把脏数据放过去等于多了一层校验。3. Spark SQL 核心操作实战3.1 注册临时表与纯 SQL 查询Spark SQL 最大的魅力就在于你可以像操作关系型数据库那样写 SQL。先把 DataFrame 注册成临时表df.createOrReplaceTempView(user_info)然后就可以用spark.sql()直接写查询了result spark.sql( SELECT job, COUNT(*) AS cnt FROM user_info WHERE age 25 GROUP BY job ORDER BY cnt DESC ) result.show()为什么有时候要用 SQL、有时候用 DataFrame API我的经验是SQL 适合表达复杂关联和聚合逻辑读起来直白DataFrame API 适合在代码里做条件分支、循环调用、动态拼接列名。两者可以随时互通一张临时表既能用 SQL 查也能用 API 继续操作并不冲突。有个小细节要提一下createOrReplaceTempView创建的临时表仅在当前 SparkSession 内有效一旦会话结束表就没了而且它不会进入 Hive 元数据。所以不要幻想临时表能跨会话复用。如果想让表持久化可以用saveAsTable它会把数据真正写到 Hive 仓库或 Spark 管理的存储中。3.2 DataFrame API 常用操作如果不想写 SQL或者你的代码里需要动态处理列名DataFrame API 就是主力。下面几个操作我几乎每个练习都会用到。select用来选列可以做列的重命名和表达式计算from pyspark.sql.functions import col, upper, year df.select( col(name), upper(col(job)).alias(job_upper), year(col(hire_date)).alias(hire_year) ).show()注意这里的col(...)是构造一个 Column 对象不是普通字符串。很多人刚学时写成df.select(name, job)也行但一旦涉及表达式、函数嵌套就必须靠 Column 对象了。filter也就是where条件过滤支持字符串表达式和 Column 表达式两种写法# 字符串写法 df.filter(age 25 AND job engineer) # Column 写法 df.filter((col(age) 25) (col(job) engineer))第二种写法的好处是不容易拼错引号而且可以利用代码补全。注意多个条件用连接而不是and因为 Python 里的and不能重载 Column 对象的布尔运算这是个很经典的报错点。groupBy和agg是聚合操作的灵魂df.groupBy(job) \ .agg( count(*).alias(cnt), avg(age).alias(avg_age), max(age).alias(max_age) ) \ .show()agg里可以放任意多个聚合函数用alias给结果列起名字。这是最推荐的习惯否则生成列名会是一串奇怪的默认名后面引用的时候非常难受。3.3 UDF如何写自定义函数内置函数总有覆盖不到的业务逻辑这时候就要写 UDF。PySpark 的 UDF 其实很简单但性能和序列化问题要注意。举个例子假设要根据年龄字段判断年龄段标签from pyspark.sql.functions import udf from pyspark.sql.types import StringType def age_group(age): if age is None: return unknown if age 25: return junior elif age 35: return mid else: return senior age_group_udf udf(age_group, StringType()) df.withColumn(age_level, age_group_udf(col(age))).show()UDF 的核心逻辑是把 Python 函数序列化之后送到每个 Executor 的 Python worker 里去执行。既然是 Python 代码在跑那就不会享受 Spark 原生算子的优化所以性能会比内置函数差一些。我一般建议能用内置函数解决的就不要写 UDF。比如上面这个例子用when函数也能实现而且快得多from pyspark.sql.functions import when df.withColumn( age_level, when(col(age).isNull(), unknown) .when(col(age) 25, junior) .when(col(age) 35, mid) .otherwise(senior) ).show()如果你确实要写 UDF也建议启动 Pandas UDF也就是pandas_udf。它的执行方式是按批次用 Pandas 来处理数据性能比逐个 Python 函数调用高不少API 却差不多是实际生产里比较常用的折中方案。3.4 用窗口函数实现分组 TopN窗口函数是 Spark SQL 里特别实用的能力尤其是分组取 TopN、累计求和、前后对比这些场景。很多新手不知道用 groupBy 实现 TopN 会写得很绕但窗口函数一条 SQL 就能搞定。比如按部门分组找每个部门薪水最高的前两个人from pyspark.sql.functions import row_number from pyspark.sql.window import Window window_spec Window.partitionBy(dept).orderBy(col(salary).desc()) df.withColumn(rn, row_number().over(window_spec)) \ .filter(col(rn) 2) \ .drop(rn) \ .show()这里的关键是Window.partitionBy(...).orderBy(...)定义了一个窗口范围在每个dept分区内按salary降序排列。row_number()会给每个分区内的行编上序号然后在外面过滤rn 2就拿到了 Top 2。这个写法要多练几次因为窗口函数在数据分析里出现的频率超高。除了row_number还有rank、dense_rank、lag、lead、sum开窗等等它们的组合能解决很多“看起来很难”的业务问题。4. 常见坑与性能排查实录4.1 数据倾斜任务卡在 99%数据倾斜应该是分布式计算里最经典、也最让人头疼的问题。症状很典型Spark 界面里某个 Stage 有十几个 Task前面的 Task 几秒跑完最后一个 Task 跑了几十分钟还没结束甚至直接 OOM。原因无非是某个 key 对应的数据量太大。比如按城市分组可能某个城市的用户量是其他城市的几百倍那么这个 key 所在的分区就会成为瓶颈。处理数据倾斜的思路有不少我实际用下来比较有效的有两个第一个是过滤掉异常大 key。如果这个超大 key 本身是脏数据或者不是业务关心的重点直接过滤掉最简单比如把 null、空串、默认值这些先剔除掉。第二个是加随机前缀打散再聚合。做法是先把一个聚合拆成两步第一步给 key 加上随机前缀让原本聚集在同一个分区的数据分散到多个分区去预聚合第二步去掉前缀再做最终聚合。这里只演示第一个场景第二个因为涉及两步聚合逻辑相对复杂练习阶段可以先了解思路等遇到真实问题再回来实现。我个人遇到过最大的一次数据倾斜就发生在一个用户行为表上某个 app 渠道的埋点记录占了总量的一半学会加随机前缀打散之后整个任务从 40 分钟缩到了 6 分钟效果立竿见影。4.2 小文件问题为什么越跑越慢在分布式文件系统里大量小文件是性能杀手。如果你的表每天写入几十个小文件每个只有几十 KB那么下游查询时每个文件都要启动一个 Task 去读白白浪费大量调度开销。处理小文件最常用的手段是调整分区数和重分区。写入前可以用coalesce减少分区数或者用repartition按某个列重新分区df.coalesce(1).write.mode(overwrite).parquet(output/) # 或者 df.repartition(col(dt)).write.mode(overwrite).parquet(output/dt20250601)这里coalesce(1)是把数据合并到一个分区再写出适合小规模练习场景。生产环境一般不会这么干因为单分区文件会太大反而影响并发度。更通用的是根据数据量估算比如目标每个文件 128 MB就用文件总大小除以 128 MB 来算分区数。但这里我要提醒一个容易误解的细节repartition会触发全量 shuffle很消耗资源。如果数据已经比较均匀只是想把少量小文件合并一下用coalesce更合适因为它是在现有分区基础上合并不会做完整的重洗牌。搞清楚这两个操作的区别是优化写任务的第一步。4.3 查询慢的隐藏原因行式扫描 vs 列式裁剪如果你发现自己明明写了select a, b from tSpark 仍然把整张表的列都读进来了那说明表可能是行式存储或者没有利用到 Parquet 的列裁剪。用 CSV 和 JSON 做练习时是不会有这种优化的因为它们是行式存储格式必须整行读出。这也是我一直建议练习到后期要体验 Parquet 的原因。Parquet 是列式存储Spark 在执行查询时会先解析逻辑计划只读取需要的列来自动跳过无关列。如果你的表按照某个常用过滤字段做了分区比如按日期dt分区那查询时会直接跳过非目标分区的数据这个叫分区裁剪。这些优化在 Spark 里都是自动发生的但前提是你用对了格式和分区策略。在实际练习中你可以做一个简单对比用一个 1 GB 左右的 CSV分别以 CSV 格式和 Parquet 格式写出来再去跑同一个聚合查询看看耗时差异。实测下来 Parquet 通常能快好几倍尤其是当你只查少数几列时差距会非常明显。这样你就能直观感受到存储格式对查询性能的影响。4.4 内存和序列化问题PySpark 和纯 Spark 不一样的是Python worker 和 JVM 之间需要频繁传递数据。如果你在一个 Python 函数里不小心引用了外部大对象这个对象会被序列化并发送到每个 Executor可能导致内存爆掉。另一个常犯的错误是在 UDF 里直接引用 SparkSession。SparkSession 本身不能被序列化一旦在 Executor 端尝试访问它就会报出反序列化相关的错误。我在练习阶段踩过一次之后就记住了UDF 里只放纯 Python 逻辑凡是涉及 Spark 对象的东西都不要放进去。如果遇到内存不足的报错也别急着把内存参数往大了调。先看看代码里有没有不必要的collect()这是把整个 DataFrame 拉回 Driver 的操作数据量一大Driver 端内存瞬间就能被打满。练习时用show()观察结果生产环境用write落盘只有明确知道数据量小、且需要回 Driver 做进一步处理时才用collect()。5. 完整练习模拟用户行为数据分析理论说再多不如动手跑一个完整练习。这一节我造了一份模拟用户行为数据包含注册信息、登录记录和订单流水然后做一套常见的数据分析。你可以直接复制代码在本地跑也可以改造成适合自己的数据集。5.1 造数据首先生成三份数据用户表users、登录表logins、订单表orders。from pyspark.sql import SparkSession from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType from pyspark.sql.functions import col, count, avg, sum, row_number, date_format from pyspark.sql.window import Window spark SparkSession.builder.appName(spark_sql_practice).getOrCreate() # 用户数据 user_schema StructType([ StructField(user_id, IntegerType(), False), StructField(name, StringType(), True), StructField(age, IntegerType(), True), StructField(city, StringType(), True) ]) users_data [ (1, Alice, 25, Beijing), (2, Bob, 30, Shanghai), (3, Cathy, 28, Shenzhen), (4, David, 35, Beijing), (5, Eve, 22, Guangzhou) ] users spark.createDataFrame(users_data, user_schema) # 登录数据 login_schema StructType([ StructField(user_id, IntegerType(), False), StructField(login_date, StringType(), True), StructField(device, StringType(), True) ]) login_data [ (1, 2025-05-01, ios), (1, 2025-05-02, android), (2, 2025-05-01, web), (3, 2025-05-01, ios), (3, 2025-05-02, ios), (4, 2025-05-01, android), (5, 2025-05-02, web) ] logins spark.createDataFrame(login_data, login_schema) # 订单数据 order_schema StructType([ StructField(user_id, IntegerType(), False), StructField(order_id, StringType(), True), StructField(amount, DoubleType(), True), StructField(order_date, StringType(), True) ]) order_data [ (1, o1001, 99.9, 2025-05-01), (1, o1002, 159.0, 2025-05-02), (2, o1003, 299.0, 2025-05-01), (4, o1004, 450.0, 2025-05-02), (4, o1005, 23.5, 2025-05-03) ] orders spark.createDataFrame(order_data, order_schema)5.2 数据清洗与预处理造完数据第一步要处理一下字段格式和空值。这里我们把order_date转成标准日期格式顺便去掉金额列里的 null 行过滤掉年龄异常的记录clean_orders orders.filter(col(amount).isNotNull()) \ .withColumn(order_dt, col(order_date).cast(date)) clean_users users.filter((col(age) 0) (col(age) 120))注意cast(date)这种写法是字符串转日期最方便的方式。如果转换失败字段值会变成 null所以转换完之后可以再查一遍空值数量。日期格式转换往往是练习里最琐碎但最重要的一步后面按天分组、计算间隔都依赖它。5.3 指标计算用户维度、城市维度、TopN先来两个最常用的聚合。第一个是每个用户的登录次数和总消费金额login_cnt logins.groupBy(user_id).agg(count(login_date).alias(login_cnt)) order_sum clean_orders.groupBy(user_id) \ .agg(sum(amount).alias(total_amount)) user_stats users.join(login_cnt, user_id, left) \ .join(order_sum, user_id, left) \ .fillna(0) user_stats.show()这里要注意fillna(0)只能填充数值列如果你有字符串列为空得用fillna(unknown, subset[city])。还有一点两个 DataFrame 做 join 时如果关联字段名字相同直接在join里写user_id就行Spark 会自动处理合并列如果字段名不同就要写成join(logins, users[user_id] logins[user_id])结果会保留两列。第二个是按城市统计用户数、平均年龄、总订单金额city_stats clean_users.join(order_sum, user_id, left) \ .groupBy(city) \ .agg( count(user_id).alias(user_cnt), avg(age).alias(avg_age), sum(col(total_amount)).alias(city_total_amount) ) \ .orderBy(col(city_total_amount).desc()) city_stats.show()第三个是窗口函数练习找出每个城市里订单金额最高的用户。window_spec Window.partitionBy(city).orderBy(col(total_amount).desc()) top_user_per_city user_stats.withColumn(rn, row_number().over(window_spec)) \ .filter(col(rn) 1) \ .select(city, name, total_amount) top_user_per_city.show()这一步其实是把前面的知识串起来用了。你会发现一开始觉得窗口函数难懂但只要配合完整的数据场景多写几遍就能理解它的执行逻辑无非就是“分组、排序、编号”这三件事。5.4 结果输出与验证练习的最后一步把结果写到本地文件并重新读回来验证一下user_stats.write.mode(overwrite).parquet(output/user_stats.parquet) city_stats.write.mode(overwrite).json(output/city_stats.json) # 读回来验证 check_df spark.read.parquet(output/user_stats.parquet) check_df.show()写文件时我习惯把输出目录放在项目下的output文件夹里这样每次重跑之前清空一下目录避免mode(overwrite)没生效时因为目录已存在而报错。这里有个细节mode(overwrite)默认会覆盖整个目录但如果你指定了format(json)生成的文件并不是一个单个的.json文件而是一个目录下多个 part 文件这是正常的不要一看到目录结构就以为自己写错了。如果你希望结果是在一个文件里可以用coalesce(1).write...但前面说过这只是练习场景或者结果集很小的情况下才推荐因为数据量一大单文件反而制约后续读取性能。验证完成后可以在 Jupyter 里再跑几个验证 SQL比如查一下城市排名结果是否符合预期看看签到数与消费金额是否匹配这些都能帮助发现前面步骤里可能埋下的 bug。6. 关于学习和练习的一点心得我这几年带过不少新人也看过很多网上所谓的“入门教程”最大的感受是PySpark 入门容易写出来能跑的代码也容易但真正跑到大数据量上不崩、不慢、结果对靠的是对 Spark 运行机制的理解和持续踩坑的积累。如果你现在刚开始我建议的学习路径是先掌握 DataFrame 的构建和基础操作再练 SQL 和 DataFrame API 的相互转换然后用窗口函数解决实际问题最后再去涉猎 RDD 底层、调优和集群部署。很多人一上来就从 RDD 开始学结果学了一堆算子遇到业务需求还是不知道怎么写 DataFrame实在是绕了弯路。最后分享一个小技巧每次练习前在代码开头打印一下 Spark 版本然后强制设置合理的分区数。很多时候新手抱怨“Spark 跑得太慢”其实不是 Spark 慢而是默认的 200 个 shuffle 分区在本地环境里太夸张了每个分区处理的数据太少调度开销反而超过了计算时间。把spark.sql.shuffle.partitions调小一点比如本地设成4很多任务会明显变快。这个参数在集群上要重新评估但在入门练习阶段它帮你省下的时间能让你更专注于理解逻辑本身。