企业官网建设流程全解析

把 Spark 当成“能跑更大数据的 Pandas”，是我见过最多、也最致命的理解偏差。身边做推荐、风控、交易特征的同学，Pandas 用得再熟练，一接 Spark 或 Dask 就开始写 for 循环、频繁 collect，跑出来的任务比单机还慢，然后第一反应就是加内存、加节点。这篇文章想聊清楚一件事：大规模特征计算真正的瓶颈和正确姿势到底是什么。我会结合网约车日志清洗、订单特征聚合这类实战场景，把 Spark/Dask 里真正能救命的技巧、参数和坑一次性讲透。适合两种人看：正在从 Pandas 迁移到 Spark/Dask 的算法或数据工程师，以及刚搭完集群但任务总跑不快的同学。

1. 为什么 Spark / Dask 绝不是“放大版的 Pandas”

1.1 一个被忽略的差异：Eager 与 Lazy 的思维切换

Pandas 是典型的 eager 执行。你写df[df["amount"] > 100]，这一行执行完，结果立刻就在内存里了，可以直接看shape、head()、.values。这种“拿到数据再想下一步”的交互方式，让人非常舒服，但也把人的思维养懒了。

Spark 和 Dask 不一样，它们是 lazy 执行。你写df.filter(col("amount") > 100).groupBy("driver_id").count()，这行代码执行完，Spark 并没有真的去过滤、分组、计数，它只是往内存里塞进了一个计算图（DAG）。只有当你调用.collect()、.show()、.write()这类 action 操作时，任务才真正开跑。

这个差异直接导致了一个经典翻车现场：有人想看看中间结果长什么样，随手敲了个.count()或者.show()，结果整个任务被强制触发了一次完整的 action 计算；后面再写下一段代码，又触发一次 action。底层的父 RDD/DataFrame 被重复计算了好几遍，比单机 Pandas 慢十倍都不奇怪。

Dask 的 lazy 机制相对轻量一点，本质也是一样：.compute()之前都是“描述计算”，算完之后才出数据。所以从 Pandas 切过来，第一件事不是学 API，而是接受“你在描述一个计算流程，而不是在操作一批数据”这个事实。

1.2 数据分区与 Shuffle：真正决定性能的底层机制

Pandas 只有一个进程、一个内存空间，所有数据都在同一个进程里躺着。Spark/Dask 则把数据切成很多块，每个块叫一个 partition（分区），分到不同 worker 上并行处理。

理解分区之后，还必须理解 Shuffle。所谓 shuffle，就是数据在节点之间搬运重组的动作。join、groupBy、distinct、orderBy这些操作都会引发 shuffle，因为原本分散在不同分区的数据，需要按某个 key 重新分组，这就必须把数据通过网络和磁盘搬家。

我打一个比方：Pandas 就像一个小工作室，三个员工共用一张大桌子，谁拿材料都方便。Spark 是一栋写字楼，几十个工位分布在不同的楼层和房间，每个工位只有自己面前的抽屉。做特征计算就是“按部门重新分组”，Pandas 只需要抬头喊一嗓子，Spark 却要让所有人搬着电脑和工位换楼层。这个“搬家”的成本，就是 shuffle 的成本。

理解了这一点，你就明白为什么 Spark 任务常见的优化原则是：减少 shuffle 次数、缩小 shuffle 数据量。很多在 Pandas 里习以为常的写法，比如多次 groupBy、频繁 join，在分布式环境下都会变成灾难。

1.3 内存、容错与执行模型：挂了不能直接拍大腿

Pandas 跑挂了，进程崩了，数据在内存里的东西全没了，你能做的只有修复代码然后重跑。Spark/Dask 不一样，它们有血缘机制——每个 DataFrame 都记得自己是“从哪个父 DataFrame、做了什么操作”变来的。某个节点挂了以后，调度器会找到对应的父分区，重新计算一次就能恢复，这就是容错。

但容错不是免费的。如果血缘链特别长（比如连续做了 20 步操作），一旦深层节点失败，重算的代价会非常大，甚至比从头跑还慢。所以我后面会重点讲 checkpoint 的用法——它就是用来“截断血缘”的。

内存模型更是完全不同。Pandas 是“一个大内存空间，全塞进去，不够就 OOM”。Spark 的 executor 内存被明确分成 execution（执行）和 storage（存储）两部分，shuffle 数据、聚合中间结果、缓存数据都各有各的地盘。这意味着，光给 Spark 加内存不一定能解决问题，反而可能导致 GC 变长、任务更慢。

这三层差异，构成了全文所有技巧的底层逻辑：不要再问“为什么我把 Pandas 的一行代码搬过来这么慢”，而要问“我的计算到底触发了几次 shuffle、多少数据被搬动了”。

2. 大规模特征计算的六大实用技巧

2.1 先过滤、再投影、后聚合：特征是“减”出来的

我见过太多人做特征计算，上来就select *，然后全列做聚合。这个习惯在 Pandas 里不算大问题，但在 Spark 里会直接影响扫描成本、序列化成本和 shuffle 量。

正确顺序是：

1. 尽早过滤。比如只需要近 7 天数据，就在读表或读文件之后立刻where dt >= ...，把数据量缩到最小。
2. 只保留需要的列。列越少，传输到 shuffle 的数据就越少。
3. 尽量少 groupBy。把多个特征合并到一次聚合里，而不是同一个 key 聚合三次。
4. 如果中间结果要反复使用，才考虑 cache；否则别乱缓存。

举个实际例子。你有一个订单表，里面有 50 列，但司机特征只需要订单金额、订单时间、城市 ID、司机 ID。如果你直接df.groupBy("driver_id").agg(sum("amount"))，Spark 会扫描全部 50 列，虽然它可能只读必要的列，但很多情况下你需要显式写select来帮助优化器裁剪。我常用的写法是：

order_df = ( spark.read.parquet("orders") .filter(col("dt") >= "2025-01-01") .select("driver_id", "amount", "order_time", "city_id") )

然后在order_df上做聚合。这一步对性能的影响，往往比后面任何调参都大。特征工程的本质不是“造出来的”，而是“先缩小数据空间，再在缩小后的空间里做变换”。

2.2 窗口函数：时间序列特征的利器

跟时间相关的特征，几乎都离不开窗口函数。常见需求包括：最近 N 天的累计订单量、上一单和当前单的时间间隔、按司机排名的金额、每个司机每天的第几单、未来 N 天的均值等。

窗口函数的写法并不难，难点在于理解它的执行代价。窗口函数通常会在指定的partitionBy键上做一次全部分区的排序和聚合，这会触发 shuffle。如果partitionBy的 key 基数特别大（比如一亿个司机），每个分区又排一次序，代价可不小。

我放一个网约车场景里的经典案例：计算每个司机的“上一单距离当前单的时间间隔”。这个特征对风控、司机行为分析都很重要，做法是用lag窗口函数：

from pyspark.sql import functions as F from pyspark.sql.window import Window window_spec = Window.partitionBy("driver_id").orderBy("order_time") feature_df = order_df.withColumn( "prev_order_time", F.lag("order_time").over(window_spec) ).withColumn( "interval_seconds", (F.col("order_time").cast("long") - F.col("prev_order_time").cast("long")) )

在这个例子里，partitionBy("driver_id").orderBy("order_time")就相当于按司机分桶、再按时间排序，Spark 会对这个 key 做全局 shuffle。如果司机数量有几百万，这个窗口的执行时间会非常可观。一个实测有效的优化：如果业务上不需要“所有司机全局排序”，可以先按city_id做前置过滤或分组，让窗口的 partition key 只覆盖当前需要的数据范围。

Dask 里对应也可以用dask.dataframe的groupby+apply做类似操作，不过 Dask 的窗口函数支持不如 Spark 完整，复杂场景建议落到 SQL 上，或者直接用dask-sql。

2.3 Broadcast Join：小表别和大表硬 Join

特征计算里最常见的操作之一，是把一个大事件表和一个小维表 join 在一起。比如订单表有 1 亿行，城市维表只有 300 行，天气表可能只有几万行。Pandas 里merge一下就完事，Spark 里如果直接 join，Spark 会默认把两个表都做 shuffle，把小表的数据广播到所有节点再和大表匹配。其实 Spark 有一个自动的广播优化spark.sql.autoBroadcastJoinThreshold，默认 10MB，当小表小于阈值时，自动转 broadcast join。

但自动优化不一定总生效。我习惯手动指定，尤其是知道维表很小的时候：

from pyspark.sql import functions as F city_df = spark.createDataFrame([(1, "上海"), (2, "北京")], ["city_id", "city_name"]) order_df = order_df.join(F.broadcast(city_df), "city_id", "left")

加了F.broadcast之后，Spark 会把city_df分发到每个 executor 的本地内存里，join 的时候完全不做 shuffle。这里有一个重要原则：广播的表要真的小。如果广播一个 50GB 的表，那么每个 executor 都塞 50GB，Driver 分配广播任务时先把自己撑爆。一般超过 100MB 就需要重新考虑方案。

在 Dask 中，类似的需求可以用map_partitions+ Python 字典来做，先收集小表到内存，然后在每个分区内用字典映射新列，效果更直接，也不容易触发复杂的图优化。

2.4 别用 Python 原生循环处理行数据，用向量化/内置函数

有一部分同学从 Pandas 迁移过来，会习惯性写df.rdd.map(lambda row: ...)或者for row in df.collect()。在 Pandas 里这是可以接受的，因为数据都在内存里；在 Spark 里，每map一次都是一次分布式任务调度，每 collect 一次都是把所有数据拉回 Driver，这两件事都是性能毒药。

处理大规模数据，正统做法是用 Spark 的内置函数和表达式。比如判断一个订单是否属于高峰时段，不要写 UDF：

# 不建议 def is_peak(hour): return 1 if (hour >= 7 and hour <= 9) or (hour >= 17 and hour <= 20) else 0 udf_is_peak = F.udf(is_peak, IntegerType()) df.withColumn("is_peak", udf_is_peak(F.hour("order_time")))

推荐写法：

df.withColumn( "is_peak", F.when( (F.hour("order_time").between(7, 9)) | (F.hour("order_time").between(17, 20)), 1 ).otherwise(0) )

内置函数的好处是：不需要序列化、不需要将 Python 和 JVM 之间的数据来回转换、可以被 Spark 催化剂优化器进一步优化。真遇到复杂业务逻辑非写 UDF 不可，在 Spark 3.0+ 建议用 pandas UDF（基于 Arrow），它把数据处理放到 Python 的 pandas 里批量执行，序列化和函数调用开销都小很多。Dask 里则是map_partitions，把整块分区丢给 pandas 函数，效果类似。记住一句话：从“逐行逻辑”切换到“批量/向量化逻辑”，是 Pandas 打法的核心迁移点。

2.5 分区数不是越多越好，Repartition / Coalesce 怎么用

分区数是 Spark 任务性能最直接的“旋钮”。分区太少，并行度不够，CPU 空闲；分区太多，每个任务只干一点点活，调度开销和文件碎片却可能拖垮整台集群。

经验法则有两个：

1. 对于读文件进来的 DataFrame，分区数通常由文件块数量决定，不需要刻意调整。
2. 对于 shuffle 之后的 DataFrame，分区数由spark.sql.shuffle.partitions决定，默认是 200。这个默认值很坑：你只有 500MB 数据，它也会给你拆成 200 个分区，一群小任务，白白增加调度开销；你有 100TB 数据，只有 200 个分区，又严重不够。所以实践中要根据数据量和集群规模动态调整。

常用的调法：

spark.conf.set("spark.sql.shuffle.partitions", "400") # 或者直接对 DataFrame 重分区 df = df.repartition(400, "driver_id")

repartition会真正触发一次 shuffle，把数据按指定列和分区数重新分布；coalesce只是把现有分区合并，不产生 shuffle，所以当需要减少分区但不关心数据分布时用coalesce更便宜。

还有一个高频场景：处理数据倾斜。如果某个 key（比如某个大司机）的数据量远超其他 key，聚合时单个 executor 就容易 OOM。常见解法是加随机盐，比如把 key 后面拼一个 0-9 的随机数，让一个大 key 分布到 10 个分区上，做两阶段聚合：先按加盐后的 key 聚合，再去掉盐做第二次聚合。这个手法在特征计算里非常实用，尤其是按热门商品、热门用户做聚合时。

2.6 缓存与检查点：用对了省事，用错了更慢

cache/persist 是 Spark 里最容易被滥用、也最容易用错的功能。很多人以为只要df.cache()就能快，结果发现任务越跑越奇怪。

cache 的真相是：

• cache()是 lazy 的。调用.cache()之后，如果后面没有 action，Spark 根本不会缓存，等于白写。
• 缓存级别要选对。默认MEMORY_ONLY，数据放不进内存就扔掉，下次还是要重算。如果中间结果很大，建议用MEMORY_AND_DISK，放不下的落盘，至少不会全丢。
• 缓存是有成本的。缓存一份数据需要把数据序列化放内存或磁盘，如果某个结果只被用一次，缓存反而更慢。

更重要的一个技巧是 checkpoint。当血缘图谱很长（比如做了 50 个步骤），某一步失败后 Spark 要从头重算最原始的父 DataFrame，代价极高。checkpoint 会把当前 DataFrame 的中间结果直接写到可靠存储（如 HDFS），然后切断血缘，后续计算从 checkpoint 位置重新开始。

spark.sparkContext.setCheckpointDir("hdfs:///tmp/checkpoint") df = df.checkpoint()

这个操作在分布式任务里相当关键。特征计算的 Pipeline 往往又长又重，在中间结果做一次 checkpoint，相当于给计算图拍了个快照，后续随便怎么折腾，都从快照开始，不用回溯整条血缘。Dask 没有完全等价的全局 checkpoint，但可以分步计算后把结果存 parquet 再接 reads，思想是同一套。

3. 实战：从 JSON 原始日志到特征宽表的 Spark 完整过程

3.1 场景定义与数据形态

用一个经典场景：网约车订单日志。原始数据是 JSON 文件，每条记录包含订单 ID、司机 ID、乘客 ID、上车时间、下车时间、订单金额、城市 ID、上车点经纬度、订单状态等。现在要按司机维度，产出最近 7 天的特征宽表，包括订单量、日均订单金额、高峰时段占比、上一单间隔均值、平均每单时长等，供下游模型使用。

原始 JSON 大概长这样：

{ "order_id": "123456", "driver_id": "d_9981", "passenger_id": "p_2233", "status": "completed", "order_time": "2025-05-06 18:33:12", "finish_time": "2025-05-06 18:51:40", "amount": 42.5, "city_id": 101, "start_location": {"lat": 31.2304, "lng": 121.4737} }

这种数据在 Pandas 里直接json_normalize展开也能做，但一旦文件有好几百 GB，Pandas 连读都读不动，Spark 反而是首选。

3.2 读取 JSON 并处理嵌套结构与时间字段

读取 JSON 最直接的方式是：

df = spark.read.json("hdfs:///raw_data/orders/*.json")

Spark 会自己推断 schema。但这里有两个坑：

1. inferSchema 开销巨大。Spark 默认需要遍历一遍数据才能推断出所有字段类型，文件越多越慢。如果你已经知道字段格式，最好直接显式定义 schema，使用spark.read.schema(...)读取。
2. 嵌套 Struct 解析。读取之后，经纬度字段会变成一个struct类型，需要通过col("start_location.lat")访问，不能直接当普通 DataFrame 列处理。

时间字段也是一个重点。原始 JSON 里order_time是字符串，Spark 推断 schema 时很可能推断成 StringType。要参与窗口排序、时间差计算，必须先转成 timestamp：

df = df.withColumn("order_time", F.to_timestamp("order_time", "yyyy-MM-dd HH:mm:ss")) df = df.withColumn("finish_time", F.to_timestamp("finish_time", "yyyy-MM-dd HH:mm:ss"))

然后做初步清洗：过滤掉金额异常（小于 0）、状态不是 completed 的记录、上下车时间颠倒的记录。清洗这一步的代码：

clean_df = ( df.filter(F.col("status") == "completed") .filter(F.col("amount") >= 0) .filter(F.col("finish_time") > F.col("order_time")) .select( "order_id", "driver_id", "order_time", "finish_time", "amount", "city_id", F.col("start_location.lat").alias("start_lat"), F.col("start_location.lng").alias("start_lng") ) )

到这里，数据形态已经变成一张比较规整的订单明细宽表。接下来要做特征计算。

3.3 特征计算的 Pipeline 实现

现在看具体特征怎么算。我列出几个有代表性的特征及实现思路。

特征一：近 7 天订单量、总金额、平均金额

一次性聚合，避免多次 groupBy：

driver_agg = clean_df.groupBy("driver_id").agg( F.count("*").alias("order_cnt_7d"), F.sum("amount").alias("total_amount_7d"), F.avg("amount").alias("avg_amount_7d") )

特征二：高峰时段订单占比

先算每笔订单是否属于高峰时段，再按司机汇总。用when函数生成 0/1 标签，不需要 UDF：

peak_df = clean_df.withColumn( "is_peak", F.when( (F.hour("order_time").between(7, 9)) | (F.hour("order_time").between(17, 20)), 1 ).otherwise(0) ) peak_agg = peak_df.groupBy("driver_id").agg( F.mean("is_peak").alias("peak_ratio") )

特征三：上一单间隔的均值、中位数

用窗口函数先算出每一单的间隔，再按司机聚合：

window_spec = Window.partitionBy("driver_id").orderBy("order_time") gap_df = clean_df.withColumn( "prev_order_time", F.lag("order_time").over(window_spec) ).withColumn( "gap_seconds", (F.col("order_time").cast("long") - F.col("prev_order_time").cast("long")) ).filter(F.col("gap_seconds").isNotNull()) gap_agg = gap_df.groupBy("driver_id").agg( F.avg("gap_seconds").alias("avg_gap_seconds"), F.expr("percentile_approx(gap_seconds, 0.5)").alias("median_gap_seconds") )

percentile_approx在超大分组数据里比精确百分位函数快得多，误差通常可接受，是生成分布类特征的常用替代方案。

特征四：平均每单时长

直接两个时间戳相减：

duration_agg = clean_df.withColumn( "duration_seconds", (F.col("finish_time").cast("long") - F.col("order_time").cast("long")) ).groupBy("driver_id").agg( F.avg("duration_seconds").alias("avg_duration_seconds") )

最后把多个特征 join 成宽表。这里有个性能细节：多个聚合结果都是按driver_id这个 key 做 groupBy，理论上会各自产生一次 shuffle。为避免过多 shuffle，可以把多个聚合合并到一次groupBy里用多个agg表达式一次性算完，如果窗口特征没法合并，就把窗口生成的结果作为一列，放回原表之后再一起聚合，尽量少拆分组动作。

把聚合结果合并成宽表：

feature_table = ( driver_agg .join(gap_agg, "driver_id", "left") .join(peak_agg, "driver_id", "left") .join(duration_agg, "driver_id", "left") )

然后按实际业务需要，对空值做填充：

feature_table = feature_table.fillna( { "avg_gap_seconds": 0, "median_gap_seconds": 0, "peak_ratio": 0, "avg_duration_seconds": 0 } )

3.4 写出 Parquet 并处理类型转换、分区策略

特征宽表算完之后，最后一步是落地。我强烈推荐 Parquet 而不是 CSV：

• Parquet 是列式存储，下游模型如果只需要其中几列，可以跳过大部分数据扫描。
• 体积比 CSV 小很多，自带压缩。
• 保留类型信息，不会出现 CSV 写出后时间字段全变字符串、金额精度丢失的问题。

写出方式按日期分区：

feature_table.write.mode("overwrite") \ .partitionBy("dt") \ .format("parquet") \ .save("/warehouse/features/driver_feature_7d")

这里需要先把当前计算日期dt加到 DataFrame 里，然后按天分区，将来按天做增量特征更新就很方便。注意一点：partitionBy的列会自动参与写入，但如果不小心把dt加进来后又去做过滤，Spark 可能没法做分区裁剪，所以写完后查数要记得带上分区过滤条件。

还有一个类型转换的细节：Spark 在 join 时如果两个 DataFrame 的 join key 类型不一致，会静默做 cast，但性能会下降。我在实际项目中遇到过LongType和IntegerType字段 join 的情况，小表还好，大表 join 直接慢了几倍。所以建议在写 pipeline 时提前统一 key 类型，比如：

driver_agg = driver_agg.withColumn("driver_id", F.col("driver_id").cast("string")) gap_agg = gap_agg.withColumn("driver_id", F.col("driver_id").cast("string"))

这一步骤看起来不起眼，但在大规模场景里非常关键，也延续了 Pandas 里astype对齐类型的好习惯。

4. 常见问题与排查技巧实录

4.1 Spark 内存问题：OOM、Executor 丢失、GC 飙升

几乎每个 Spark 项目都会碰到内存问题，现象各有不同：任务在某个 stage 卡死、Container killed by YARN、ExecutorLostFailure、GC 时间占比超高。排查的第一步永远是看 Spark UI，而不是猜参数。

具体看几个关键页面：

• Executors 页面：看到某个 executor 的 Shuffle Spill 非常大，说明内存不够，数据一直在溢写到磁盘，需要提高 executor 内存或者减少单分区数据量。
• Stages 页面：看每个 task 的 Shuffle Read/Write 量。如果某个 task 的输入量是其他 task 的几十倍，那基本是数据倾斜。
• GC 时间：如果 Executor 的 GC 时间占总运行时间 30% 以上，说明内存管理出了问题，可能需要调整spark.memory.fraction。

常用参数调整：

--executor-memory 8g --conf spark.memory.fraction=0.75 --conf spark.memory.storageFraction=0.3

spark.memory.fraction表示用于执行和存储的内存占 executor 总内存的比例，默认 0.6，如果 GC 严重可以适当调大到 0.75；storageFraction表示其中专用于 cache 的占比，如果不太用 cache 可以调小。但请记住，内存参数永远排在不产生 shuffle 和减少数据规模的优化之后。很多人一遇到 OOM 就疯狂加内存，加到 64G 还是挂，原因就是 shuffle 量太大或者分区严重倾斜，加机器根本缓解不了。

4.2 JSON 读取与嵌套解析：Schema 推断慢、时间字段解析失败

刚才实战环节已经提了一些，这里再补充排查要点。

如果 JSON 文件特别多，spark.read.json自动推断 schema 会触发一整轮全量扫描。解决办法之一是加samplingRatio：

spark.read.option("samplingRatio", 0.1).json("hdfs:///raw_data/*.json")

只抽样 10% 的数据来推断 schema，速度能快不少。但抽样有风险，少数文件里多出的字段可能没被推断出来，后面读出来变成 null。我的经验是，线上固定格式的数据文件直接显式写 schema 最稳妥。

嵌套数组的展开也是常见难点。JSON 里如果某个字段是数组，需要先explode再处理。比如每个订单可能包含多个途经点，你要按途经点展开成多行再做特征：

from pyspark.sql import functions as F df = df.withColumn("waypoint", F.explode(F.col("waypoints")))

展开之后行数会增多，下一阶段的聚合要格外小心数据量膨胀。我见过有人对一个大 JSON 展开之后总行数翻了 100 倍，后面 join 直接把集群跑挂。展开之前先想清楚：这个字段是不是真的需要进特征。

时间字段解析失败也是高频问题。JSON 里的时间字符串可能是2025-05-06T18:33:12Z、2025-05-06 18:33:12、甚至20250506183312，Spark 默认的不一定能正确识别。最简单粗暴的办法是显式指定格式：

F.to_timestamp(F.col("order_time"), "yyyy-MM-dd'T'HH:mm:ssXXX")

在写样例数据测试时，尽量把各种时间格式都覆盖到，避免上线以后在凌晨三点被某个异常格式打断。

4.3 Pandas 数据类型转换和环境安装的坑

热词里有一堆跟 Pandas 有关的问题，比如“pandas 数据类型转换”“pycharm 怎么安装 pandas 包”“import pandas as pd 清华源下载”“手机 python 安装 pandas”。这些看似是小问题，但工程里确实耗人。

先解决环境安装这个比较常见的点。如果你遇到：

ERROR: Could not find a version that satisfies the requirement pandas (from versions: none) ERROR: No matching distribution found for pandas

通常原因有三个：pip 源里没有当前 Python 版本对应的 pandas 轮子、Python 版本太新或太旧、缺少依赖库导致安装脚本失败。最快的解决办法之一是切换到豆瓣或清华的镜像源：

pip install pandas -i https://pypi.tuna.tsinghua.edu.cn/simple/

Pandas 的依赖版本如果和你已有环境里的 NumPy 冲突，装完之后import pandas还可能报错，建议用 venv 或 conda 单独建环境，避免污染系统 Python。手机 Python 环境一般用 Termux 或 Pydroid，直接 pip 也一样，但要注意手机架构可能没有预编译包，只能走源码编译，速度慢且容易失败，等不了建议直接放弃。

再说数据类型转换。Pandas 里常见的坑包括：字符串列里有空格，astype("int")直接报错；时间字段混了两种格式，pd.to_datetime性能骤降；超大整数被当成int64，转成 Spark 后变LongType没问题，但转回 pandas 时溢出等等。在 Pandas 里正确做法是用pd.to_numeric(..., errors="coerce")或pd.to_datetime(..., format="...")来强制转换，不要裸用astype。

这个习惯要带进 Spark。数据处理 pipeline 里经常能看到col("amount").cast("double")、col("order_time").cast("timestamp")，凡是不确定类型的地方，都要考虑转换失败的兜底逻辑，比如先when(col("amount").cast("double").isNull(), default)再使用。大规模数据处理最忌讳的就是某个脏数据在最后阶段才炸出来，全链路排查成本极高。

4.4 集群搭建与脚本运行中的经典翻车现场

最后一个章节，聊聊集群搭建和 Spark 脚本运行时最容易踩的坑。看热词里“spark 集群搭建”“spark 搭建教程”“spark 数据分析案例”“spark 脚本示例”特别多，说明这块是真的难倒不少人。

本机装 PySpark 开发，和真正搭一个集群跑生产任务，完全是两码事。我见过最经典的几个问题：

• Java 版本不匹配。Spark 3.x 对 Java 8/11 有不同支持，装错版本开局就跪。
• 节点间 hostname 无法解析。Standalone 模式下，worker 注册不到 master，报Failed to connect to master，检查/etc/hosts是最优先动作。
• Python 环境不一致。每个节点上的 Python 版本、pandas/numpy 等依赖不一致，任务分发到某个节点就报ModuleNotFoundError。生产环境强烈建议用 conda 或虚拟环境打包分发，而不是逐台机器手工装。
• 端口被防火墙挡住。Spark master 默认 7077，Web UI 默认 8080，YARN 的调度端口也有可能冲突。排查命令先跑jps看进程是否都在，再用netstat查端口监听情况。

还有一种资源抢占场景也值得专门说。现在不少团队会采购带 GPU 的 AI 工作站或一体机，像 NVIDIA DGX Spark 这类设备就经常被装上一堆“一键安装脚本”。我见过有人在同一个机器上同时跑 ComfyUI、训练脚本、Spark，结果 Spark 任务一跑就 OOM。后来发现是 ComfyUI 之类的服务常驻占掉了大量显存和内存。很多人在问“DGX Spark 怎么关机”，其实重点不是关机，而是关掉不用的服务、给 Spark 预留资源。如果你也在类似设备上跑 Spark，先看free -h和nvidia-smi，确认内存/显存还有多少余量，再调整 Spark 的 executor 内存，别指望一台机器同时扛下所有任务。

spark-submit的常用参数也顺手整理一下：

spark-submit \ --master spark://your-master:7077 \ --deploy-mode cluster \ --driver-memory 4g \ --executor-memory 8g \ --executor-cores 4 \ --num-executors 10 \ --conf spark.sql.shuffle.partitions=400 \ your_job.py

这几个参数基本覆盖了资源分配的核心。每个参数不是越大越好，要根据集群总资源反算，比如 100GB 内存的机器，executor-memory 开 32g，那一个节点最多也就两三个 executor，还要留给系统余量。总之，先看机器能提供多少，再决定参数，并预留 20% 以上的余量给系统和 shuffle 缓存。

4.5 常见问题速查表

从 Pandas 迁移到 Spark/Dask，我踩过的最大的坑其实不是 API 记不熟，而是思维没有转过来。单机 Pandas 里数据是“躺”在内存里的，你想怎么拿就怎么拿；分布式世界里，数据是“流动”的，你每做一个操作，背后的网络和磁盘都要付出代价。所以再遇到慢任务，先别急着加资源，去 Spark UI 看看数据到底被搬动了多少次、每个分区扛了多少数据。把这些看透，比记住一百个 API 都管用。