Spark数据挖掘全流程实战：从数据清洗到模型部署

1. 单机数据挖掘的天花板：为什么要换Spark

1.1 先说我踩过的那个内存爆炸

第一次让我下定决心系统学Spark，是我用Pandas跑一份千万级订单数据，机器内存直接被干爆的时候。任务管理器里内存占用拉满，Python进程直接被杀，两三个小时算出来的中间结果全没了。那种感觉经历过一次就明白了：单机环境下做数据挖掘，天花板太清晰了。后来我把整套流程搬到Spark上，数据量到了千万级反而顺畅许多，也才开始真正理解大家经常挂在嘴边的"大数据"到底在说什么。

传统的数据挖掘项目里，无论是做用户画像、商品推荐还是风险预测，流程无非是数据获取、清洗预处理、特征工程、模型训练与评估这几个环节。只要数据规模还停留在百万行以内，Pandas加scikit-learn完全能应付。但一旦跑到千万行、上亿行，单机内存就开始出问题：read_csv直接溢出、merge跑半天、GridSearch跑了几个小时候意外崩掉，连中间结果都没了。问题不在算法，而在数据规模对单机算力的碾压。

Spark解决的是"算力横向扩展"的问题。它的核心思想是：把一份大任务切分成无数小任务，分发到多台机器上并行处理，节点之间通过内存通信而不是反复读写磁盘。配合lazy evaluation（惰性求值），Spark只记录数据处理的血缘关系，等到真正需要结果时才触发计算，并且通过DAG调度和容错机制保证某个节点挂了之后数据可以从血缘重新恢复。这套设计，天生适合数据挖掘里那些"反复迭代、反复读取、反复聚合"的场景。

1.2 数据挖掘四个阶段，Spark分别怎么接招

拿一次完整的数据挖掘项目来说，Spark几乎在每个环节都有对应的工具。

数据获取和存储阶段，Spark可以通过DataFrameReader对接HDFS、Hive、MySQL、Parquet、JSON、CSV等各种数据源，统一的读取方式让下游清洗环节不用关心数据来自哪里。清洗和预处理阶段，Spark SQL可以把DataFrame当成一张表，用SQL做过滤、去重、补缺、join；如果你更习惯代码，DataFrame API同样顺手。特征工程阶段，除了手动写SQL，MLlib里还封装了VectorAssembler、StringIndexer、StandardScaler、OneHotEncoder等常用转换器。到了模型训练阶段，MLlib直接提供分类、回归、聚类、协同过滤、频繁项集挖掘等算法，模型训练和交叉验证接口也都有。

这里想强调一个观点：数据挖掘的方法论，比如特征怎么构造、模型怎么选择、效果怎么评估，并不会因为换到Spark就改变。变的是计算载体——数据不再是一份整体装进内存的二维数组，而是切分在多节点上的分布式数据集。你以前会玩SQL、会做特征、懂机器学习的原理，搬到Spark之后需要补的只是分布式编程的思维方式和几个API习惯，核心数学基础依然有复用价值。这个认知很重要，它决定了你是一上来硬啃源码，还是先把全流程跑通。

1.3 和MapReduce、单机Python做个对比

为了搞清楚Spark在数据挖掘里扮演的角色，我把它和单机Pandas、Hadoop MapReduce放在一起对比过：

对比之后会发现，MapReduce能把数据算出来，但不适合做数据挖掘里高频迭代的算法。比如K-Means这种需要多轮迭代的算法，如果每次迭代都重新读写一次磁盘，代价太高，实际效率反而不如Spark。而Spark的内存迭代让它既能处理批任务，也能在同一个引擎里跑流式作业、图计算和机器学习，这种"一套工具养全家"的特性，在做数据挖掘项目时非常省事。

2. 环境搭建与核心概念：先把Spark跑起来

2.1 选哪种部署模式，学习期和生产期怎么选

第一次接触Spark，最容易被各种部署模式搞懵：Local、Standalone、YARN、Kubernetes，每个都有人说好。

如果你的目的是学习和验证，Local模式就够了。它的意思是Spark在一台机器上模拟集群，不用HDFS、不用额外的资源调度器，下载解压就可以跑spark-shell。毕业设计或者小规模项目，可以搭一个3节点的Standalone集群，连上HDFS之后你能看到任务怎么打散到多个worker上执行。生产环境下企业一般走YARN或Kubernetes，因为要跟Hadoop生态统一资源调度。

我的建议是：除非你在公司接触的就是YARN，否则个人学习不要一上来就搞YARN。先在Local模式把代码逻辑跑通，再搭Standalone把分布式特性体验一遍，把Spark UI、任务调度这些概念真正看到，比看十篇架构文章都有效。

2.2 安装Spark时最容易忽略的版本兼容问题

安装本身不复杂，照着官方文档下载解压就行：

# 以Spark 3.5 + Hadoop 3为例 wget https://dlcdn.apache.org/spark/spark-3.5.0/spark-3.5.0-bin-hadoop3.tgz tar -zxvf spark-3.5.0-bin-hadoop3.tgz cd spark-3.5.0-bin-hadoop3 # 配置环境变量 export SPARK_HOME=/path/to/spark export PATH=$SPARK_HOME/bin:$PATH # 启动本地spark-shell验证 spark-shell

真正容易踩坑的是版本兼容。Spark 3.x要求JDK8或JDK11，但有些老项目的CDH发行版里默认JDK7，跑起来会莫名其妙报各种ClassNotFound。另外Spark和Scala的版本也要对齐：Spark 3.2以前的版本大多基于Scala 2.12，3.3以后逐步迁移到Scala 2.13，如果你要通过spark-submit提交自己写好的Scala作业，scalaVersion不匹配，直接编译不过。

Windows用户还会遇到一个经典问题：本地跑Spark时hadoop的winutils.exe缺失，导致原生库文件找不到。解决方法是在Hadoop的bin目录下放一个对应版本的winutils.exe，或者改用带embedded Hadoop的发行包。很多以前在Windows上初学Spark的人卡在这一步卡了很久，所以我特别提一句。

2.3 RDD、DataFrame、Dataset：到底该用哪个

Spark有三种数据抽象：RDD、DataFrame、Dataset。

RDD是最底层的弹性分布式数据集，所有操作都用算子来表达（map、flatMap、reduceByKey、filter），灵活但繁琐。DataFrame相当于分布式环境下的二维表，带Schema，Spark SQL优化器会对它做谓词下推、列裁剪等自动优化，所以同样的操作，DataFrame往往比直接写RDD快，而且代码更短。Dataset在DataFrame之上加了强类型和编译期检查，主要在Scala里用得比较多，Python里我们通常不需要区分Dataset和DataFrame。

我的结论很直接：做数据挖掘，默认用DataFrame和Spark SQL。能用SQL表达的处理，不要手动去写RDD算子。比如判断缺失值比例、去重、join这两张表，SQL一眼就能看懂，换成RDD就得写一长串Lambda，而优化器能帮你做的列裁剪和谓词下推，自己写RDD时反而很容易写歪。

2.4 用WordCount理解分布式计算的秒懂模型

学任何Spark入门教程，WordCount都是第一个例子。它虽然简单，但能把分布式计算最核心的"数据打散再聚合"讲清楚：

from pyspark.sql import SparkSession spark = SparkSession.builder.appName("WordCount").getOrCreate() lines_rdd = spark.sparkContext.textFile("hdfs://.../input/*.txt") word_counts = lines_rdd.flatMap(lambda line: line.split(" ")) \ .map(lambda word: (word, 1)) \ .reduceByKey(lambda a, b: a + b) word_counts.collect()

执行逻辑是这样的：flatMap把每一行文本按空格打散，变成无数个单词；map把每个单词转成(word, 1)的键值对；reduceByKey把所有相同单词的计数累加到一个值上。其中最关键的是reduceByKey，它会触发一次shuffle——把分布在各个节点上同一key的数据拉到一个节点做合并，这个动作既是分布式计算的精髓，也是性能瓶颈的来源。

理解了shuffle，后面那些数据倾斜、分区调优的问题就有了基础。因为任何groupBy、join、reduceByKey都会产生shuffle，shuffle一旦不合理，整个job就慢下来。所以我认为WordCount的意义不在于"会写这段代码"，而在于理解了"数据是怎么在分布式环境下流动的"。

3. 数据挖掘的第一公里：用Spark SQL完成数据清洗和特征工程

3.1 从网约车订单数据说起：业务背景决定清洗规则

在网约车大数据综合项目里，我处理的订单数据大概是这样的：一行一个订单，包含order_id、user_id、driver_id、pickup_time、pickup_lng/pickup_lat、dropoff_lng/dropoff_lat、distance_km、fare_amount、coupon_amount、status（completed/canceled）等字段。规模在千万行级别。

这个案例最合适拿来讲数据清洗，因为字段类型够全：有时间、有数值、有经纬度、有枚举状态。而清洗规则不能拍脑袋定，得先想清楚每个字段的业务场景。比如fare_amount出现负值，可能是退款单；distance_km为0但订单已完成，可能是司机没开定位，系统用默认值填充；同一笔order_id出现两次，可能是业务侧重复上报。不结合业务判断，直接删数据或补数据都很危险。

我习惯的做法是：先打印Schema，再用describe和groupBy大致看一遍每个字段的分布，找出异常的字段再逐个定规则。不要指望一次把所有脏数据清干净，数据挖掘项目里的清洗步骤往往是迭代的——每做完一轮特征，可能又发现新的数据问题，再回来补一轮清洗规则。

3.2 清洗三板斧：去重、补缺、滤异常

清洗的常用手段归纳下来就是三招：去重、补缺、滤异常。

加载数据后，第一个动作通常是去重。同一订单在重复上报场景下可能出现多条记录，用订单ID去重：

from pyspark.sql import SparkSession from pyspark.sql import functions as F spark = SparkSession.builder.appName("NetCarCleaning").getOrCreate() df = spark.read.csv("hdfs://.../orders.csv", header=True, inferSchema=True) df.printSchema() # 去重 df_unique = df.dropDuplicates(["order_id"])

然后是缺失值处理。先看看每个字段缺失比例：

missing_ratio = df.select([ (F.sum(F.col(c).isNull().cast("int")) / F.count("*")).alias(c) for c in df.columns ]).collect()

对关键业务字段，比如fare_amount、pickup_time，缺失比例小就直接删除整行；对非关键字段，比如coupon_amount，缺失通常意味着没使用优惠券，填充为0就合理。一个细节：不要所有缺失值都无脑填充0，比如经纬度缺失填0，下游计算距离特征时会算出完全错误的结果，反而不如删掉。

最后是异常值过滤。金额、里程这类数值字段，先根据业务常识设一个合理范围：

df_valid = df_unique.filter( (F.col("fare_amount") >= 0) & (F.col("fare_amount") <= 1000) & (F.col("distance_km") >= 0) & (F.col("distance_km") <= 300) )

如果想更严谨一点，可以用IQR（四分位距）来识别极端值：计算出某字段的Q1、Q3，把超出Q1-1.5IQR和Q3+1.5IQR之外的样本标记出来，人工看一眼再决定要不要删。这个方法在"校园大数据—数据清洗"这类入门项目里也很常用，因为对任何数值字段都能套用，不用每次手动定上下界。

3.3 特征工程：时间、距离、用户聚合特征

数据清洗完之后，就到了数据挖掘里最见功力的部分——特征工程。网上车订单这个场景，我通常从三个方向构造新特征。

第一个方向是时间特征。把pickup_time拆解成小时、星期几、是否周末、是否早晚高峰：

df_feat = df_valid.withColumn("pickup_hour", F.hour("pickup_time")) \ .withColumn("pickup_weekday", F.dayofweek("pickup_time")) \ .withColumn("is_weekend", F.col("pickup_weekday").isin(1, 7).cast("int")) \ .withColumn("is_rush", ((F.col("pickup_hour").between(7, 9)) | (F.col("pickup_hour").between(17, 19))).cast("int"))

第二个方向是距离特征。如果原始数据没有distance_km，但给了起终点经纬度，我习惯用Haversine公式计算直线距离近似值，Spark内置的数学函数可以直接算：

df_feat = df_feat.withColumn( "distance_haversine", F.lit(6371.0) * F.acos( F.sin(F.toRadians("pickup_lat")) * F.sin(F.toRadians("dropoff_lat")) + F.cos(F.toRadians("pickup_lat")) * F.cos(F.toRadians("dropoff_lat")) * F.cos(F.toRadians("dropoff_lng") - F.toRadians("pickup_lng")) ) )

第三个方向也是最有价值的：用户聚合特征。比如每个用户的历史下单次数、平均消费金额、历史取消率，这需要先按user_id做一次聚合，再把聚合结果join回订单表：

user_stats = df_feat.groupBy("user_id").agg( F.count("*").alias("user_order_count"), F.avg("fare_amount").alias("user_avg_fare"), F.avg((F.col("status") == "canceled").cast("int")).alias("user_cancel_rate") ) df_final = df_feat.join(user_stats, on="user_id", how="left")

这里做的其实是"把历史行为编码成当前订单的特征"，在用户画像类项目里是很常见的手段。如果还想更细一点，可以用窗口函数给每个司机最近N单的接单耗时做滚动平均——F.row_number().over(Window.partitionBy("driver_id").orderBy(F.col("pickup_time").desc()))，然后取前N条做均值。

3.4 农产品价格数据：同一套方法换个业务场景怎么落

网约车订单这套流程做熟了之后，你会发现数据清洗和特征工程的大部分代码都是可以复用的。比如农产品价格数据分析，字段大概变成province、market、product、price、date，业务规则也换一套：同市场同产品同日期可能重复采集，要去重；价格出现负值或远高于市场极值，要过滤；月度均价、环比、同比这些就是新的聚合特征。

通用方法是一样的：理解字段含义、写清洗规则、做聚合特征。这也是为什么我一直建议数据挖掘新人至少完整走一遍像"农产品价格数据清洗-Spark"或"校园大数据—数据清洗"这类项目，因为它们能把这套方法论练成肌肉记忆。等你换到下一个数据集，完全可以把自己总结的清洗函数打包成类或配置驱动的方式，把字段映射、过滤阈值都放进配置文件里，复用性会非常强。

4. 从特征到模型：MLlib流水线的落地细节

4.1 为什么主张用Pipeline组织整个建模流程

特征做完之后，很多人会直接把数据交给模型，但我强烈建议用Pipeline来组织整个建模流程。

Pipeline的意义在于，把特征转换和模型训练串成一条完整的流水线：向量化、标准化、模型训练，每个步骤都是一个Stage。好处有三点。第一，参数管理统一，所有超参数都在一个Pipeline里，重跑实验时不用东改一个西改一个。第二，避免数据泄漏，CrossValidator在交叉验证时，每个fold只会用当前训练集的统计量来拟合StandardScaler，不会偷看验证集信息。这在单机机器学习里也是经典要求，但Pipeline能帮你强制做到。第三，模型部署更方便，Pipeline保存之后，新数据只需要做一次transform就能得到预测结果，不用在线上重写一遍特征处理逻辑。

我在网约车订单不要被取消的预测任务里，就是用一个Pipeline把特征处理、标准化和随机森林串起来的。

4.2 一个完整的随机森林建模示例

建模前先把目标变量处理一下：status是completed或canceled，转成数值标签0/1：

df_label = df_final.withColumn( "status_label", F.when(F.col("status") == "canceled", 1).otherwise(0) )

然后定义特征列，组装PipeLine：

from pyspark.ml import Pipeline from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.classification import RandomForestClassifier from pyspark.ml.evaluation import BinaryClassificationEvaluator, MulticlassClassificationEvaluator feature_cols = ["pickup_hour", "pickup_weekday", "is_weekend", "is_rush", "distance_km", "fare_amount", "user_order_count", "user_avg_fare", "user_cancel_rate"] assembler = VectorAssembler(inputCols=feature_cols, outputCol="features") scaler = StandardScaler(inputCol="features", outputCol="scaledFeatures", withStd=True, withMean=True) rf = RandomForestClassifier( labelCol="status_label", featuresCol="scaledFeatures", numTrees=200, maxDepth=10 ) pipeline = Pipeline(stages=[assembler, scaler, rf]) train_df, test_df = df_label.randomSplit([0.8, 0.2], seed=42) model = pipeline.fit(train_df) pred = model.transform(test_df) bce = BinaryClassificationEvaluator(labelCol="status_label", metricName="areaUnderROC") auc = bce.evaluate(pred) print("AUC:", auc) mce = MulticlassClassificationEvaluator(labelCol="status_label", metricName="accuracy") acc = mce.evaluate(pred) print("Accuracy:", acc)

这里有几个细节值得说明。第一，VectorAssembler的作用是把所有特征拼成一列Vector，因为MLlib的算法只认Vector，不像scikit-learn可以直接吃二维数组。第二，StandardScaler做了标准化，随机森林这类树模型对尺度不敏感，但如果后面换逻辑回归或K-Means，标准化是必要的预处理，放在Pipeline里省得以后来回改。第三，随机森林里numTrees控制集成规模，通常100到300够用；maxDepth控制树深，太深容易过拟合，太浅学不到复杂关系，我一般从5试到15，再根据验证集表现定。

4.3 评估与调参：别只看准确率

二分类问题里，如果正负样本不平衡（比如取消订单只占10%），光看accuracy很容易被多数类骗。假设模型把所有订单都预测成不取消，accuracy也有90%，但这个模型没有任何业务价值。所以对这类场景，我习惯同时看AUC、召回率和PR曲线，重点调高业务关心那个类的召回率。

超参数搜索直接用CrossValidator加ParamGridBuilder：

from pyspark.ml.tuning import CrossValidator, ParamGridBuilder param_grid = (ParamGridBuilder() .addGrid(rf.numTrees, [100, 200]) .addGrid(rf.maxDepth, [5, 10, 15]) .build()) cv = CrossValidator( estimator=pipeline, estimatorParamMaps=param_grid, evaluator=bce, numFolds=5 ) cv_model = cv.fit(train_df) best_model = cv_model.bestModel

实操中有个经验：分布式环境下交叉验证的时间成本比单机高得多，尤其当训练数据很大时，numFolds设成5或10会让整个训练时间成倍增长。我的做法是，先用随机抽样取一份较小但类别均衡的数据集快速定参数方向，再拿着确认下来的参数全量跑一次最终模型。别一上来就全量数据十倍交叉验证，等到第二天发现参数搜索矩阵配错了，白白浪费计算资源。

4.4 和scikit-learn的差异，以及模型怎么部署

用过scikit-learn的人切到MLlib，通常会遇到几个不适应的地方。

第一是数据结构不同。scikit-learn直接接受numpy数组或pandas DataFrame，而MLlib要求所有特征必须组装成一个Vector列，所以Pipeline第一步永远是VectorAssembler。第二是调试方式不同。单机模型你可以随时print中间变量，分布式下数据散落在多个executor里，很多中间结果并不会输出，需要靠Spark UI、采样sample()和collect()少量数据来验证。第三是算法细节略有差异。虽然都是随机森林、逻辑回归，但MLlib是对原始论文的分布式实现，部分参数叫法不一样，比如树模型里的maxBins表示连续特征离散化时的最大桶数，特征基数很高的维度如果不用奥尼编码或降基处理，maxBins也要调大。

模型训练完之后，部署分两种路径。离线批量预测最直接：model.save("hdfs://.../rf_model")，新数据来了用PipelineModel.load加载再transform即可。如果要上线为在线服务，可以把Pipeline导出为PMML格式，或者把特征计算和模型transform封装成Python服务接进内部系统。很多毕设和竞赛项目往往忽略了“把模型变成可以重复使用的东西”这一步，但真实工作里这个环节反而很关键。

5. 分析结果不落地等于白做：聚合输出与可视化

5.1 模型跑完之后，真正交付的是什么

跑出一个AUC数字只是开始。在企业里，数据挖掘项目最终交付的不只是模型指标，而是业务结论或可用服务。运营关心的问题往往是：哪个时段、哪个区域订单取消率最高？补贴策略应该怎么设计？这类问题靠Spark做聚合分析最直接：

cancel_stats = df_final.groupBy("city", "pickup_hour").agg( F.count("*").alias("order_count"), F.avg((F.col("status") == "canceled").cast("int")).alias("cancel_rate"), F.avg("fare_amount").alias("avg_fare") ).orderBy(F.desc("cancel_rate")) cancel_stats.write.mode("overwrite").csv("hdfs://.../agg_result")

聚合结果落到CSV或MySQL之后，就可以做成报表或看板。电商、农产品价格、校园数据这些场景都一样：模型精度再高，不能让业务方直接看到"下一步该做什么"，项目价值就大打折扣。

5.2 业务看板：Flask + ECharts最简单的一版

用Flask加ECharts做可视化看板，是我见过最快的一种落地方式。后端读聚合结果，前端用图表展示，几分钟就能出一版能看的原型。

后端代码很薄，核心是这样：

# app.py from flask import Flask, render_template, jsonify import pandas as pd app = Flask(__name__) df = pd.read_csv("agg_result.csv") @app.route("/") def index(): return render_template("index.html") @app.route("/api/cancel_stats") def api_cancel_stats(): return jsonify(df.to_dict(orient="records")) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000)

前端模板里，放一个DIV加一段ECharts脚本就能画折线图：

<div id="chart" style="width: 900px; height: 450px;"></div> <script src="https://cdn.jsdelivr.net/npm/echarts@5/dist/echarts.min.js"></script> <script> fetch("/api/cancel_stats") .then(res => res.json()) .then(data => { const chart = echarts.init(document.getElementById("chart")); chart.setOption({ title: { text: "不同时段订单取消率" }, tooltip: {}, xAxis: { type: "category", data: data.map(d => d["pickup_hour"]) }, yAxis: { type: "value" }, series: [{ type: "line", data: data.map(d => d["cancel_rate"]), areaStyle: {} }] }); }); </script>

这段代码技术含量不高，但决策链里非常实用。哪怕不做前端看板，把聚合表用Excel打开，把取消率最高的几个时段标红，汇报效果都会好很多。真实项目里，数据挖掘工程师一半的精力都在"算出来、展示出去、讲清楚"这三件事上。

5.3 从单个案例到通用模板

做完网约车订单这个项目，你会发现这套流程完全可以抽象成模板：读数据、清洗、特征构造、模型、聚合分析、可视化。换成农产品价格数据分析，就是清掉重复采集记录、按市场聚合出日均价和波动率、再画出价格趋势曲线；换成校园数据分析，就是清洗选课成绩数据、做用户聚类、再展示各院系平均绩点分布。

所以我在系统里都会保留一套可以直接跑通的Spark全流程项目，后续接任何新业务，改的是字段映射和业务规则，不变的是"数据怎样从原始到特征再到结果"的骨架。这种沉淀下来的模板，比单个项目的代码积累更值钱。

6. 实战中绕不开的坑：性能、内存与调优

6.1 数据倾斜：groupBy心中的痛，加盐解决

第一次正式跑网约车订单聚合时，按城市统计订单量配上用户维度做特征，跑了快一个小时都不结束。打开Spark UI，发现某个stage里有少数task运行时间极长，GC时间占比高得吓人。再一看task处理的数据量，少数几个task几十GB，其他task几百MB——这就是典型的key分布不均导致的数据倾斜。

定位方法很简单，先看倾斜的key分布：

df.groupBy("city").count().orderBy(F.desc("count")).show(10)

找到热点key之后，最通用的处理是加盐（salted key）做两阶段聚合。所谓加盐，就是给key拼上一个随机前缀，让原本集中在一个key上的数据被拆散到多个task上：

from pyspark.sql import functions as F # 第一阶段：加盐聚合 df_salted = df.withColumn( "salt", (F.rand() * 10).cast("int") ).withColumn( "salted_city", F.concat(F.col("city"), F.lit("_"), F.col("salt")) ) stage1 = df_salted.groupBy("salted_city").agg(F.sum("fare_amount").alias("sum_fare")) # 第二阶段：去掉盐前缀，再做一次聚合得到真实结果 stage2 = stage1.withColumn( "city", F.split("salted_city", "_")[0] ).groupBy("city").agg(F.sum("sum_fare").alias("total_fare"))

加盐的核心思想是：把热点key随机拆成10份，让这10份分散到不同task上，避免单个task扛下所有数据。这个技巧在特征工程阶段也很好用，尤其是做大规模用户聚合的时候，热点用户带来的倾斜往往比城市维度严重得多。

6.2 小文件问题：写文件前不控制，重新读就想哭

数据清洗完写Parquet到HDFS时，我踩过一个特别典型的坑：清洗过程中用了groupBy和filter，shuffle默认分区数是200，结果每张表都被切成上百个小文件。当时没在意，后来重新读取时发现效率明显下降，HDFS的NameNode压力也大了不少，因为每个小文件都要占一条元数据记录。

解决思路是在写数据之前主动控制分区数。如果只是想减少文件数量且数据量不大，用coalesce()最合适，它尽量不触发shuffle；如果是要重新分布数据到合理粒度，就用repartition()。比如一个3节点集群，写结果前：

df_final.coalesce(12).write.mode("overwrite").parquet("hdfs://.../result.parquet")

另外spark.sql.shuffle.partitions这个参数也很关键。默认200是全国分区数，对大数据集合理，但如果集群规模小、数据量也不大，200反而会制造大量小任务。我通常按executor数量乘以核心数再乘2到3来估算，比如50个executor总共200个核，shuffle分区设400到600是合理的。

6.3 executor内存该怎么给，Spark UI怎么看

数据挖掘项目最容易遇到的问题就是OOM，而且大多数OOM不是你代码逻辑错了，是内存分配不合理。

配置上，我常用的经验值是：每个executor内存4g到8g、核心数2到4个。内存开得太大，GC反而频繁，因为Spark要对一个大堆做标记清理；开得太小，数据放不下就会频繁溢写到磁盘。堆外内存同样要预留，有些算子会用到堆外存储，不设置spark.memory.offHeap.enabled=true和spark.memory.offHeap.size，可能遇到莫名其妙的"Direct buffer memory"报错。

还有一个容易被忽略的参数是广播阈值spark.sql.autoBroadcastJoinThreshold，默认10MB。join时如果有一张小表（比如维表）小于这个阈值，Spark会自动把它广播到每个executor，避免shuffle。可是默认阈值不算大，大一点的维表就不会广播了。遇到join慢时，可以主动用小表做广播：

from pyspark.sql import functions as F df_big.join(F.broadcast(df_small), on="key", how="left")

排查问题离不开Spark UI。Jobs页面看每个Stage的输入输出大小和耗时，Executors页面看每个executor的GC时间和内存使用。如果某个executor GC时间占比很高，多半是内存不足或者数据倾斜；如果某个Stage的shuffle read特别大，就该考虑调大分区数或者优化join方式。还有一个土办法：把问题任务的数据量用df.filter(条件).count()和df.sample(0.01).collect()直接看，很多时候比UI更直观。

6.4 给想走这条路的人一个学习路线建议

很多同学问我，从零开始学Spark做数据挖掘，该按什么顺序走。我的建议很简单：别一上来啃源码或架构论文，先按这个链路把全流程跑通一遍。

第一步是把SQL和DataFrame API搞熟练，能读文件、能过滤、能join、能groupBy。第二步用Spark SQL完成一个真实数据集的清洗和特征工程，比如网约车订单数据、农产品价格数据都可以。第三步用MLlib跑通一个分类或聚类模型，理解Pipeline、VectorAssembler、超参数调优这些概念。第四步再考虑调优和扩展，比如加盐解决数据倾斜、优化内存配置、搭一个三人节点集群。最后再涉猎Streaming和GraphX这类扩展模块。

还有一件事要提醒：入门阶段的高线平台、在线练习里那些题都是"数据是干净的、字段含义是明确的"，而真实项目里数据又脏又乱、字段含义要翻业务文档才知道。所以一定要自己动手处理一份真实脱敏数据，把每个环节都走一遍。等你亲手清洗过一次网约车订单、亲手调过一次数据倾斜、亲手把结果画成看板，再回头看那些大数据的抽象概念，都会变得具体很多。

踩过这么多坑之后，我有个特别深的体会：在Spark里做数据挖掘，大部分时间其实不在写模型，而在搞清楚数据长什么样、该怎么洗、怎么拼特征。很多人学习时容易把精力全放在算法的数学原理上，但到了实际项目里，决定最终效果上限的，往往是数据质量和特征工程。另一个体会是别怕跑得慢，先让它对起来，再让它快起来。我第一次跑通全流程时，代码也算不上优雅，后来一点点调shuffle、调分区、调内存，才慢慢理解了分布式计算到底在做什么。如果你也正卡在"数据太大跑不动"或者"项目不知道从哪下手"这两个问题上，希望这篇内容能给你一个可以照着走的起点。