PySpark处理海量数据的关键在于规避环境、读取和执行陷阱：需匹配Java版本（如Spark 3.5+用Java 17），显式定义schema避免inferSchema内存暴增，慎用count/collect改用take/cache，优化groupBy+orderBy为limit减少shuffle。
PySpark 处理海量结构化数据，不是“能不能”，而是“怎么避免掉进坑里就跑起来”。本地单机跑 SparkSession 也能处理 GB 级 CSV，但一上真实数据就卡住、OOM 或结果不对——问题通常不在代码逻辑，而在环境、读取方式和执行语义。
Java 版本不匹配导致 Spark 启动失败或静默崩溃
这是最常被忽略的拦路虎。Spark 3.5+ 要求 Java 17，而 Spark 3.2–3.4 多数依赖 Java 11；Spark 2.4 则只认 Java 8。版本错配不会报明确错误，而是表现为：java.lang.NoClassDefFoundError、Driver 进程启动后立即退出、或 spark.read.csv(...) 返回空 DataFrame。
- 运行
java -version确认当前默认 JDK 版本 - 用
update-alternatives --config java（Linux）或手动修改JAVA_HOME（macOS/Windows）切换到匹配版本 - 验证：启动 PySpark shell 后执行
sc.version和spark.sparkContext._jvm不报错才算稳
读取 CSV 时 inferSchema=True 导致内存暴增或类型推断错误
对 TB 级结构化数据，inferSchema=True 会强制 Spark 扫描全部数据样本来猜字段类型——不仅慢，还极易把整数列误判为字符串（尤其含空值或前导零），后续 groupBy 或 agg 就失效。
- 始终显式传入
schema：用StructType定义字段名和类型，例如StructField("age", IntegerType(), True) - 若必须推断，先用
sampleRatio=0.01限制采样比例，再配合columnNameOfCorruptRecord="_corrupt_record"捕获脏数据 - 生产环境禁用
header=True自动解析列名——CSV 头部可能含不可见字符，建议用columns = ["id", "name", "salary"]显式指定
count() 和 collect() 在分布式场景下是性能炸弹
df.count() 看似 harmless，但它触发全量 shuffle 和 reducer 汇总，TB 数据可能跑十几分钟；df.collect() 更危险——它把所有分区数据拉到 Driver 内存，哪怕只有 10GB 也大概率 OOM。
- 用
df.take(5)替代collect()查看样本 - 用
df.select("col").distinct().count()前加df.cache()避免重复扫描 - 统计行数需求强烈时，优先查底层存储元数据：Parquet 文件自带 row count，S3/HDFS 可用
hadoop fs -du -s path
groupby + agg 后 order by 引发非必要 shuffle
写成 df.groupBy("dept").agg(avg("salary")).orderBy("avg(salary)") 看似自然，但 Spark 会先 shuffle 分组，再 shuffle 排序——两次网络传输。实际中，90% 的业务只需要 top-N 结果，没必要全局排序。
立即学习“Python免费学习笔记（深入）”；
- 改用
df.groupBy("dept").agg(avg("salary").alias("avg_salary")).orderBy("avg_salary", ascending=False).limit(10) - 如果 N 很大（比如 top 10000），考虑先用
approxQuantile估算阈值，再filter缩小范围 - 注意：
orderBy必须在agg之后，否则语法报错——这是 DataFrame API 的执行顺序约束，不是 bug
df.filter(col("status") == "active")，而是当数据从 GB 涨到 TB，你能否一眼看出哪行代码正在拖慢整个 Stage，以及 Driver 日志里那条 Stage X skipped 是不是掩盖了数据倾斜。这些细节不写在文档里，只藏在失败重试三次后的日志末尾。 
