不用纠结这个项目标题是不是太“基础”很多人一开始都容易把它当成“又一个大数据框架介绍”但实际上Spark的Python版本也就是PySpark已经把大数据的门槛拉得非常低了。我见过不少只写过几年Python、连分布式系统都没碰过的开发者靠着PySpark直接上手处理几十个G甚至上T的数据。这篇文章就把Spark当作一个“用Python就能玩转的分布式计算工具”来讲讲清楚它到底是什么、核心概念有哪些、底层是怎么跑的以及你第一次用PySpark写任务时最容易踩的那些坑。不适合什么人看呢如果你已经用Spark写过一年以上的生产任务那这篇对你来说确实偏入门了。但如果你正在纠结“要不要学Spark”“PySpark和普通Python处理数据到底有什么区别”“分布式到底是怎么个分布法”那这篇可以把这些问题一次性讲明白。1. 到底什么是Spark为什么大家都在用1.1 Spark是什么批处理、流处理、SQL一把抓先说结论Spark是一个通用的分布式计算引擎。这里的“通用”不是随便说说的因为它能干的活确实很杂处理大规模结构化数据跑SQL分析做ETL也就是从各种数据源里抽取数据、清洗转换、再写回目标存储处理实时流数据比如从消息队列里消费数据做实时统计跑机器学习算法像分类、聚类、协同过滤这类常见算法它都内置了处理图数据做图计算。一个引擎能把这些事全包了这是它流行起来的最核心原因。在我早年做数据开发的时候公司的技术栈是割裂的离线分析用一套工具实时计算又用另一套框架机器学习还要再起一个单独的环境。维护成本高数据从离线到实时之间的口径还经常对不上。Spark出来后至少批处理和流处理在API层面统一了写离线任务的代码改一改就能跑实时场景这省下来的工作量不是一星半点。有一件事需要先明确Spark本身不是存储系统。它不负责把你的数据存起来它负责的是“算”。你的数据可能放在分布式文件系统里可能放在各种云存储里也可能在数据库里Spark只是把它们读进来在内存里做分布式计算再把结果写出去。所以你看Spark解决的核心问题就是“单机算不动”的那类场景。单机Python处理数据内存上限决定了你最多能处理多大的数据量。就算你的机器有256G内存一旦数据超过这个规模要么等死要么OOM。Spark的思路是把计算分散到很多台机器上每台机器只处理自己那一份数据然后合并结果。这个思路不复杂但把它做得稳定、高效、易用就非常难了。1.2 为什么Python版PySpark这么火这个要追溯到开发者的使用习惯。早期Spark只提供Scala和Java接口那个年代写大数据基本是Java工程师的活。后来Spark加入了Python API也就是PySpark局面立刻不一样了——Python生态在数据分析和机器学习领域本来就有绝对统治力PySpark相当于给这批人开了一条通往分布式计算的路。PySpark能火还有一个很实在的原因代码写起来比Scala版本省心太多。我对比过同一个逻辑的两种实现用Scala写要定义各种类型、处理序列化和编译问题而用Python写几乎就是按着数据处理的直觉来。尤其是你本来就会pandas的话PySpark的DataFrame API上手会非常快很多方法名和习惯都是类似的。另外PySpark还能配合Python生态里的其他库一起用。你在一个PySpark任务里引入pandas、numpy、scikit-learn这些库完全没障碍。虽然跑分布式任务时不建议在主流程里大量用纯Python库但做一些小规模预处理、写UDF的时候Python生态的优势就体现出来了。我接触到的团队里现在用PySpark的比例已经明显超过了Scala。不是说Scala不好而是对一个业务团队来说招一个会Python又能理解数据的工程师比招一个Scala专家容易太多了。而且PySpark的社区活跃度很高踩到坑基本都能搜到解决方案。2. Spark核心概念盘点先搞懂这几个词文档也能读下去了2.1 RDD最基础的数据抽象RDD全称是弹性分布式数据集Resilient Distributed Dataset。这个概念是Spark的骨架级抽象虽然有经验的开发者现在写代码很少直接操作RDD都是操作DataFrame但不懂RDD的话你很难真正理解Spark的容错和调度原理。你可以把RDD理解为“一个被分片存储、可以被并行操作的数据集合”。它有几个关键特性分区数据被切分成若干个分片每个分片就是一台机器上的一个计算单元不可变一旦创建就不能修改任何“修改”操作都会生成新的RDD懒加载真正触发计算的是动作而不是转换容错每个RDD都记录着自己是从哪个父RDD通过什么操作得来的也就是血缘关系某个分片坏了可以按血缘重新计算。用生活化的比喻来理解RDD就是一摞被拆分成好几叠的卡片每张卡片写着一行数据。你没法直接改某张卡片只能拿着一叠卡片去做变换最后交给统计员汇总。这个过程不动原卡片而是生成一叠新卡片。虽然现在日常开发已经很少写sc.parallelize这种代码了但理解RDD对后面理解DataFrame的底层逻辑非常有帮助。DataFrame其实就是带有schema信息的RDD如果你遇到DataFrame层面的诡异问题最终还是要到RDD层面去排查。2.2 DataFrame与DataSet面向结构化的好朋友先说Dataset这是Spark 1.6开始引入的强类型数据抽象提供了编译时类型安全检查但主要在Scala/Java里体现价值。在PySpark里你接触到的Dataset其实就是DataFrame因为Python是动态类型语言所谓强类型检查在Python里不存在。DataFrame在PySpark里就是最核心的数据抽象。它有两个特点有Schema也就是“每一列叫什么名字、是什么类型”这一点和pandas、SQL表很像底层分布式存储自动分区并行计算。为什么DataFrame比RDD更流行主要是两个原因一是DataFrame的优化空间大。Spark的优化器Catalyst优化器会对DataFrame操作做逻辑优化和物理优化就像SQL数据库里的查询优化器一样。你写的代码会被重写成执行效率更高的计划。而RDD层面没有这种优化空间你写的什么逻辑就怎么执行。二是DataFrame更接近数据分析师和工程师的思考习惯。大部分人对数据处理的理解天然是“表”和“行列”而不是分布式抽象。我实际用下来的感觉是能用DataFrame就不用RDD。DataFrame的API足够丰富而且不断在更新除非你要做非常底层的自定义逻辑否则RDD真的没必要直接碰。2.3 SparkSession、Transformation与Action先搞清楚懒执行Spark 2.0开始SparkSession成了统一的入口。以前要分别创建SparkContext、SQLContext、HiveContext现在一个入口全搞定。你写PySpark代码的第一行基本就是创建或者获取一个SparkSessionfrom pyspark.sql import SparkSession spark SparkSession.builder \ .appName(my_first_job) \ .master(local[*]) \ .getOrCreate()appName是这个任务的名字会在集群管理界面上显示用来区分不同任务。master决定了任务跑在哪里local[*]表示本地模式使用本机所有CPU核。接下来是神仙打架级别的关键概念Transformation和Action。Transformation叫转换操作比如select、filter、groupBy这类操作的特点是懒执行——你调用了它Spark并不会立刻计算只是记录了一个操作步骤。Action叫动作操作比如count、collect、show这类操作会真正触发Spark去执行计算流程。用一个比喻解释懒执行比如你要做一道菜转换操作就是“把菜谱看了一遍又一遍”动作操作才是“真正下锅开炒”。懒执行带来的好处是高性能。Spark会把一连串转换操作记录下来形成一个DAG有向无环图然后在执行前对整个链条做优化调整。比如你连续做了多个filterSpark可能会把它们合并成一个过滤条件减少数据扫描次数。这就是为什么Spark能在大规模数据上保持高效率的秘密之一。刚接触PySpark的人最常见的疑惑就是“我明明写了df.filter(...)为什么数据没变”答案就是你只写了转换操作没有触发的动作数据当然不会变。记住一句话没有Action就没有计算也没有结果。3. 从宏观到微观Spark架构与运行原理3.1 集群里到底谁在干活Driver、Executor、Cluster ManagerSpark是典型的主从架构跑一个Spark任务时集群里有三类角色Driver任务的“大脑”负责解析你的代码、生成执行计划、把任务分发给各个工作节点、汇总结果。它运行在你的提交端或者集群的一个专用节点上。Executor任务的“苦力”真正干活的进程负责执行具体的计算任务、把结果缓存到内存或磁盘。每个节点上可以有多个Executor进程。Cluster Manager负责“分配地盘”也就是管理集群资源决定在哪些机器上启动Executor。Spark支持不同的集群管理器有自带的Standalone也有通用的资源调度系统。我经常把这三个角色类比成项目团队的组成Driver是项目经理Executor是干活执行的工程师Cluster Manager是行政人事负责组团队和分配场地。一个容易让人迷惑的点是Spark任务执行时你的代码到底在哪里运行答案是Driver端和Executor端都有运行。比如你写了一个df.groupBy(city).count()Driver把这段逻辑翻译成执行计划分发给Executor执行Executor会把分组和计数的具体计算跑起来。如果是自定义UDFSpark会把这个函数的序列化副本发给每个Executor各自在本地执行。3.2 一个Spark任务从提交到出结果的完整旅程我来完整地梳理一遍当你执行一个PySpark任务时后台到底发生了什么你提交PySpark应用Driver进程启动创建SparkContext或者SparkSession底层一样。SparkContext向Cluster Manager申请资源即在多台机器上启动Executor进程。你的代码被解析成一系列转换操作构建成DAG有向无环图DAG描述了整个计算步骤的依赖关系。当遇到Action操作时DAG调度器把DAG切割成多个Stage。切割规则是遇到宽依赖就切分Stage。每个Stage由多个Task组成这些Task被分发到各个Executor执行。Executor执行任务把结果写回Driver或者外部存储比如分布式文件系统、数据库。全部任务完成后Driver退出释放资源。这里面最关键的是第4步提到的“宽依赖”。什么是宽依赖呢简单说就是父RDD的某个分区数据被多个子RDD分区使用比如groupByKey、join这种操作通常需要发生数据混淆Shuffle也就是数据要在集群节点间重新分配。而窄依赖就是父RDD的每个分区只对应一个子RDD分区比如map、filter不需要网络传输。窄依赖的阶段执行效率很高宽依赖所在的Stage则是优化和调优的重点。3.3 部署模式怎么选local、Standalone、YARN还是K8s部署模式决定了你的Spark任务跑在哪里。下面这几种我都实际用过可以给你一些选择参考。Local模式就是本地跑不依赖集群。适合开发调试、跑演示Demomaster(local[*])就是这种。它好在零部署成本坏在只能用一台机器的资源。Standalone模式是Spark自带的集群部署方式不依赖其他系统就能搭一个简单的Spark集群。适合中小规模实验环境、学习环境或者不想引入额外组件的时候。YARN模式是目前很多生产环境的选择因为它可以把Spark任务和其他计算框架的资源统一管理起来集群利用率高。适合在Hadoop生态体系成熟的大数据平台里跑。如果你所在的公司已经有一套大数据平台大概率用的YARN。Kubernetes模式是近几年越来越流行的方式Spark可以直接以容器的方式跑在K8s集群里资源隔离性、弹性都更强。适合云原生环境尤其是容器化架构已经比较完整的团队。我的建议很简单学习就Local生产环境哪个和你现有体系融合度高就选哪个。不要为了追新而强行迁移部署模式分布式系统最怕迁移时的各种环境问题。4. 用Python版Spark实操从零跑通一个PySpark任务4.1 本地环境搭建一条命令还真能搞定如果你是Python派环境搭建非常简单。前提是你机器上已经装了Python 3.8以上版本。我用的是pip直接装pip install pyspark没错就这么简单。pyspark这个PyPI包里面自带Spark的JVM运行环境装完就能直接跑Local模式。不用单独下载Spark、不用配置JAVA_HOME什么都不用配。我当年第一次用Spark的时候配环境花了一整天现在新入门的朋友真的幸福太多。装完之后你在命令行敲一下pyspark就能进到Spark的交互式界面可以直接写SQL、执行DataFrame操作跟用Jupyter差不多。如果你是更习惯写脚本的人那就建一个Python文件比如叫app.py里面写SparkSession初始化然后直接spark-submit app.py提交运行。不过本地测试时直接python app.py也行因为Local模式下SparkSession自己就能把JVM拉起来。4.2 读取数据并做转换最小可用示例我拿一个最常见的场景举例读一个CSV文件做简单的统计和过滤操作。假设现在有个CSV文件里面是用户订单数据字段包括order_iduser_idcityamountorder_date。from pyspark.sql import SparkSession from pyspark.sql.functions import col, sum spark SparkSession.builder \ .appName(order_analysis) \ .master(local[*]) \ .getOrCreate() # 读取CSV自动推断schema df spark.read.option(header, True) \ .option(inferSchema, True) \ .csv(orders.csv) # 查看数据 df.show(5) # 计算每个城市的总订单金额 result df.groupBy(city).agg(sum(amount).alias(total_amount)) # 只保留金额超过10000的城市 result result.filter(col(total_amount) 10000) # 触发Action真正执行计算 result.show()看到这坨代码如果你是写pandas的应该会觉得很多地方很眼熟。groupBy、agg、filter这些API和SQL的语法非常贴近。不同点是pandas是在内存里一口气算完而PySpark把这个计算分布式地拆到了多台机器上数据量更大时优势非常明显。有一点要注意不要直接用df.collect()把全量数据拉到本地。collect()会把分布式所有分区数据汇总到Driver端内存里如果数据量太大你的Driver会直接OOM。正确方式是show()看几条或者write到目标存储。4.3 PySpark新手必踩的几个坑第一个坑是类型问题。CSV自带字段类型需要推理但推理经常“猜错”比如日期被读成了字符串、空值被读成了null导致类型变复杂。我的建议是正式任务里不要单纯依赖inferSchema最好在建表或读取时手动指定schema。这样做虽然多写几行代码但能避免无数后续问题。from pyspark.sql.types import StructType, StructField, StringType, DoubleType schema StructType([ StructField(order_id, StringType(), True), StructField(user_id, StringType(), True), StructField(city, StringType(), True), StructField(amount, DoubleType(), True), StructField(order_date, StringType(), True), ]) df spark.read \ .option(header, True) \ .schema(schema) \ .csv(orders.csv)第二个坑是UDF性能太差。很多从pandas转过来的开发者习惯把Python函数包成udf在PySpark里跑。但UDF有一个致命问题它会打破Spark的优化机制数据要在JVM和Python进程之间来回序列化传输。如果数据量大UDF上的数据量等于“火山式”性能灾难。我的经验是能用内置函数解决的就用内置函数比如pyspark.sql.functions提供了一大批常见函数实在要写UDF优先用pandas_udf它通过Apache Arrow做了优化性能比普通UDF好很多。第三个坑是分区数不调。刚用PySpark时我把数据读进来直接处理跑得特别慢。后来发现默认的分区数并不一定适合你的数据和集群配置。分区太多调度开销大分区太少每块数据太大执行不均匀。具体怎么调我放在下一节详细说。5. 调优与排查跑大数据不能只靠玄学5.1 分区设置影响速度的关键参数分区的数量和大小是PySpark任务性能的决定性因素之一。每个分区对应一个计算单元数据被分到多少分区就意味着最多能有多少任务并行执行。分区太少有些Executor可能没活干资源闲置单个分区数据量过大处理时间变长内存压力也大。分区太多任务调度和元数据开销增加每个任务启动、结束都有成本你的时间都浪费在“发任务”上了。怎么判断分区数合不合适先看两个指标如果每个任务执行时间只有几百毫秒那很可能分区太多了如果某个Stage只有几个任务但每个任务跑了几十分钟那很可能分区太少或者数据倾斜了。一个简单的经验公式基于我的实践不是什么玄学目标让每个分区的数据量在100MB到200MB左右。假设你的数据有2GB那设置10到20个分区比较合理。调整分区的常用方式# 重新分区会触发Shuffle df df.repartition(20) # 合并分区减少分区数尽量不触发Shuffle df df.coalesce(5)注意repartition和coalesce是有区别的。repartition会把数据重新均匀打散代价是触发一次完整的Shuffle适合需要调整并行度的时候。coalesce只能减少分区数而且尽量不移动数据适合输出结果前缩小分区避免产生太多小文件。5.2 内存管理执行内存和存储内存怎么平衡Spark把Executor的内存分成了两大块执行内存和存储内存。执行内存用于计算比如Shuffle、Join、Aggregation时的临时数据存储内存用于缓存RDD或DataFrame。在Spark 1.6之后引入了统一内存管理器执行内存和存储内存不再严格隔离而是可以互相抢占。举个例子如果缓存数据占用了存储内存而执行时需要更多内存它会主动驱逐一部分缓存数据来给执行让路。这解释了一个常见的困惑为什么我看到Storage Memory里有大量缓存任务还是OOM了因为这个缓存占的是存储内存而执行时需要的内存不够时虽然会让缓存让路但如果缓存没有及时释放就会触发额外开销甚至失败。实际调优时不需要太激进改内存比例除非你明确知道自己在做什么。我更建议你先关注Executor数量与CPU核数的关系每个Executor的内存大小是否每个Executor都能获得足够的“并发任务”资源。spark.executor.memory控制Executor的堆内存spark.executor.cores控制并发任务数。不需要机械遵循某个“标准配置”但有一个原则单个Executor的内存不要设置太大因为内存越大的JVMGC垃圾回收压力越大。一般推荐单个Executor的堆内内存在2GB到8GB之间这个区间是我踩过各种坑之后觉得比较稳的范围。5.3 数据倾斜最常见的性能杀手及缓解思路数据倾斜是分布式计算中“人人见面都叹气”的问题。它的表现是其他任务都跑完了只有一两个任务卡在那里跑不动整个Stage都等在那几个慢任务上。数据倾斜的本质是某个分区里的数据量远大于其他分区。比如你按城市分组统计订单量结果“上海”这一个城市的订单占了总数据量的一半那么处理上海这个分区的任务就会比处理其他小城市分区的任务慢几十倍。解决倾斜有几种常用手段我按推荐顺序列一下提高Shuffle并行度让每个任务处理的数据变少。这是最省事的解法spark.sql.shuffle.partitions调大比如从默认200调到500甚至更高。但它只是缓解不能根治。加盐salting。针对热点Key给它加上一个随机前缀或后缀把它打散成多个Key再处理。比如把“上海”变成“上海_0”、“上海_1”……这样就能把原先一个分区的数据分散到多个分区去算算完再合并结果。from pyspark.sql.functions import concat, lit, rand, split, col # 给城市加随机盐打散热点 salted df.withColumn( salted_city, concat(col(city), lit(_), (rand() * 10).cast(int)) ) # 按加盐后的key聚合 salted_result salted.groupBy(salted_city).agg(sum(amount).alias(total)) # 把盐去掉后再聚合一次 result salted_result.withColumn( city, split(col(salted_city), _)[0] ).groupBy(city).agg(sum(total).alias(total_amount))数据量悬殊的Join可以用广播。如果小表很小几百MB以内可以把小表广播到每个Executor避免Shuffle。PySpark里用join时指定广播提示from pyspark.sql.functions import broadcast result large_df.join(broadcast(small_df), key)广播Join在数据量差距悬殊时效果立竿见影是我最常用的优化手段之一。如果倾斜严重考虑对倾斜Key单独处理。把热点数据和非热点数据分开计算最后再union回来。操作复杂一些但能真正解决问题。调优这条路没有银弹关键是要学会看Spark UI。每个Stage的统计信息里能看到任务运行时间分布、Shuffle数据量等定位到具体瓶颈后再用上面的方法对症下药。6. 写在最后一点个人体会我从第一次接触Spark到现在最大的感受是这玩意儿的学习曲线其实没有传说中那么陡。只要你抓住“懒执行、分区、Shuffle、宽窄依赖”这几个核心概念再带着几个真实的数据集去跑一跑很快就能建立感觉。用Python写Spark更是把门槛又降了一层。你在pandas里的数据处理思路完全可以迁移到PySpark上只是要多记一个前提你的数据是分布式的你的操作是懒执行的你的结果最终要靠Action才能触发。如果你正要开始做一个PySpark项目我的建议是先从一份小数据集开始把流程整个跑通包括读取、转换、写出。然后慢慢换成大数据集在压力下观察Spark UI上的指标再反推去理解那些调优参数是怎么回事。最后再多记一句话不要一开始就想着“调优”先让任务正确跑起来再去追踪慢在哪里。分布式计算这门手艺跑得通只是起点跑得快才是经验。希望这些内容能帮你把起点站稳了。