刚拿到Spark Machine Learning(SparkML):机器学习(部分一)这个项目标题的时候我第一反应是这八成又是一个装了Spark但只会写WordCount的兄弟要开始整活了。毕竟我在社区里见到太多人问Spark怎么跑机器学习结果一聊才发现他是用Spark的RDD硬撸逻辑回归或者压根不知道SparkML和MLlib到底有啥区别。先说清楚这篇博文是干嘛的。它是Spark机器学习入门系列的第一篇目标只有一个帮你把SparkML这套分布式机器学习库的底裤扒干净让你知道它是什么、能干什么、怎么上手、踩过哪些坑。它适合这么几类人一是公司数据量已经大到单机Python跑不动、需要往分布式迁移的算法工程师二是刚学完Spark基础、想往机器学习方向进阶的大数据开发三是准备面试时被SparkML原理这类问题难住的同学。看完整篇文章你能自己搭起一个Spark环境用Pipeline把特征工程到模型训练串起来跑通一个完整的回归预测任务还能避开我当年踩过的内存和兼容性大坑。1. SparkML到底是何方神圣1.1 一个被问烂了的问题SparkML和MLlib什么关系这个事儿几乎每次技术分享都会有人问。很多老教程还停留在MLlib的叫法上这不能怪大家记性差因为历史原因确实绕。最早Spark贡献给Apache基金会的时候自带了一个机器学习库叫MLlib里面全是基于RDD的API。RDD是Spark最早的数据抽象优点是灵活缺点是太底层你在写机器学习代码时要自己操心数据怎么分布、哪个环节该cache、shuffle要怎么优化。后来Spark团队想通了推出了新的DataFrame API并且基于它重写了整个机器学习库新库就叫SparkML。所以现在的官方定位是SparkML是新一代基于DataFrame的机器学习库老版MLlib基于RDD的API进入维护状态不会再加新功能。判断你用的是哪个版本很简单看包名。org.apache.spark.mllib是老的RDD接口org.apache.spark.ml是新的DataFrame接口。我现在的项目全部基于ml这个包写。原因不复杂DataFrame自带Schema、Catalyst优化器、Tungsten执行引擎这些好东西Spark能做列裁剪、谓词下推和优化内存布局性能和易用性全面碾压RDD。1.2 为什么不能继续用Python直接撸很多算法出身的朋友问我我scikit-learn用得好好的为什么要学SparkML我的回答是如果数据量在单机内存能装下的范围内比如两三GB以内模型又不太复杂那scikit-learn确实是更顺手的工具。你没必要为了用Spark而用Spark这是典型的过度设计。但当你面临这几个场景时单机就扛不住了数据量到了几十GB甚至TB级别单机内存完全装不下训练样本是PB级日志里提取出来的需要用Spark做大规模特征工程你需要在同一个Pipeline里完成特征处理、模型训练、批量预测而且数据源分布在HDFS或S3上SparkML的核心优势是横向扩展。你在单机版sklearn里训练一个GBDT可能要几小时甚至OOMSparkML把训练任务拆分到集群的多个Executor上并行算资源够多的话能把时间压缩到几分钟。这个差异在真实业务里决定了一个模型能不能每天按时更新。1.3 SparkML官方生态的7个核心模块SparkML官网把功能分成几个大模块新手容易迷失在茫茫API里,这里直接给你整理成一张速查表模块官方包名核心内容典型场景特征工程ml.featureVectorAssembler、StandardScaler、StringIndexer、OneHotEncoder、PCA数据清洗、特征变换分类算法ml.classificationLogisticRegression、DecisionTreeClassifier、RandomForestClassifier、GBTClassifier、LinearSVC、NaiveBayes二分类、多分类回归算法ml.regressionLinearRegression、DecisionTreeRegressor、RandomForestRegressor、GBTRegressor、AFTSurvivalRegression价格预测、销量预测聚类算法ml.clusteringKMeans、BisectingKMeans、GaussianMixture、LDA用户分群、文本主题协同过滤ml.recommendationALS推荐系统、用户商品TopN模型调优ml.tuningCrossValidator、TrainValidationSplit、ParamGridBuilder超参数搜索模型评估ml.evaluationBinaryClassificationEvaluator、RegressionEvaluator、MulticlassClassificationEvaluator效果评估这篇先重点讲特征工程、回归和Pipeline这几个是后续所有算法的基础。分类和聚类放到系列后面几篇。2. 环境搭建Spark集群是绕不过去的一道坎2.1 本地模式还是集群模式很多人一开始就被搭建集群劝退了。我给你的建议是学习阶段完全可以用本地模式先把API和流程跑通再考虑集群。本地模式就是在你电脑上起一个Spark进程它能模拟Executor并行执行。对入门来说够用了也能体验到DataFrame懒加载、Stage划分这些特性。当你真正要把模型跑在几十GB的数据上时才需要考虑集群。集群形态有三类Standalone模式Spark自带的集群管理器配置最简单适合测试YARN模式Hadoop生态的集群管理器企业内部最常见资源管理能力强Kubernetes模式容器化部署适合云原生环境现在越来越流行本地学习环境建议直接装Spark并启用local模式。去官网下载一个与JDK版本兼容的二进制包比如JDK8对应Spark 3.x早期版本JDK11/17对应更新版本。解压后配置好JAVA_HOME然后跑一下以下命令# 解压Spark tar -zxvf spark-3.5.1-bin-hadoop3.tgz cd spark-3.5.1-bin-hadoop3 # 配置JAVA_HOME以mac或Linux为例 export JAVA_HOME/path/to/your/jdk export PATH$PATH:$JAVA_HOME/bin # 启动Spark本地模式的交互式Python环境 ./bin/pyspark能进入pyspark命令行就算安装成功了。2.2 集群搭建的核心步骤和三个大坑如果你真想搭建一个小的Standalone集群做实验思路很简单。一台主节点机器跑Master进程几台从节点机器跑Worker进程Worker启动时指定同一个Master地址即可。启动Master./sbin/start-master.sh启动Worker./sbin/start-worker.sh spark://master-host:7077启动后在浏览器里打开http://master-host:8080能看到当前集群有哪些Worker。搭建过程中最容易踩的坑有三个。第一个是通信超时问题集群机器之间如果不在同一个网段或者开了防火墙Worker注册到Master时经常注册不上表现就是启动Worker后Master页面一直看不到新节点。解决办法是确保机器之间网络互通检查防火墙或安全组尤其是在云服务器上8080和7077端口必须放行。第二个坑是内存超卖。Spark默认会给每个Worker分配机器全部内存作为Executor的候选但机器本身还要跑操作系统和Spark的Master进程真能被Executor用满时机器早就卡死了。配置时建议设置SPARK_WORKER_MEMORY为你机器实际可用内存的75%左右留点余量给系统。第三个坑是hosts配置混乱。集群之间的主机名互相解析不了会导致各种诡异的连接失败建议在所有节点上把master和worker的主机名都写进/etc/hosts。2.3 内存如何配置才不会频繁OOM我在社区回答过太多OOM问题了新手通病是只调spark.executor.memory实际上这不够。Spark内存分成了好几块核心参数有这四个参数作用建议初始值spark.executor.memoryExecutor堆内内存根据机器内存设一般4~8Gspark.executor.memoryOverheadJVM堆外内存堆内内存的10%~20%spark.driver.memoryDriver端内存一般2~4Gspark.sql.shuffle.partitionsShuffle分区数默认200根据数据量调整如果数据量大报错java.lang.OutOfMemoryError: Java heap space优先调大executor.memory同时按比例调大memoryOverhead。如果报的是Unable to acquire X bytes of memory这种往往是因为memoryOverhead给得太少。在yarn或standalone模式里这两块内存合起来不能超过spark.executor.memory加上overhead的上限。提示不要为了跑大数据量无脑把executor-memory搞到几十G。单个Executor内存太大时JVM的GC停顿会非常明显反而拖慢速度。更合理的做法是增加Executor数量而不是无限加大单个Executor内存。3. DataFrame是SparkML的操作对象3.1 从RDD到DataFrame数据结构的一次进化用过Spark旧版本的人都知道RDD用起来有多原始。你处理数据时脑子里想着的不是数据长什么样而是我该怎么把数据拆成一个个分区然后给每个分区写一个函数。DataFrame就不一样了它本质上是一张分布在集群里的表。每一列有类型、有名字每次操作Spark引擎会自动推演数据分布、决定哪些数据可以放在同一台机器上做计算哪些必须走shuffle。你不用手动管理分区它自己来。用代码说话。读一个CSV建DataFrame只需要一行df spark.read.option(header, true).option(inferSchema, true).csv(hdfs:///data/train.csv) df.printSchema() # 查看列名和类型 df.show(5) # 查看前5行这还不算完DataFrame背后拥有Catalyst优化器会自动对执行计划做优化。比如你只select两列Spark扫描文件时压根不会把其余列全部读进内存。这在处理宽表时收益巨大。3.2 Transformers和EstimatorsSparkML的两个核心抽象很多初学者对SparkML的API感到陌生是因为它引入了两个有门槛的概念Transformer和Estimator。Transformer是有transform方法的类作用是给DataFrame加列。最典型的特征是VectorAssembler把多列数值拼成一个特征向量。比如你有年龄、收入、点击次数三列经过VectorAssembler处理后DataFrame里会多出一列features这个column里装的是三元组向量。Estimator是有fit方法的类作用是学习参数。拿线性回归举例LinearRegression的fit方法会在DataFrame上训练产出LinearRegressionModel。这个产出的model本身也是一个Transformer它的transform方法能对新的DataFrame做预测追加一列predictions。简化理解Transformer是将一个DataFrame变成另一个DataFrame的函数;Estimator是在数据上训练出Transformer即模型的算法。Pipeline把这两者串起来。3.3 Pipeline如何避免数据泄漏Pipeline这个概念借鉴了scikit-learn它的核心是让多个stage按顺序执行。特征列拼装、标准化、训练、预测全串成一条流水线。from pyspark.ml import Pipeline from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.regression import RandomForestRegressor # 组装特征列 assembler VectorAssembler(inputCols[age, income, clicks], outputColraw_features) # 对特征做标准化均值0方差1 scaler StandardScaler(inputColraw_features, outputColfeatures) # 随机森林回归 rf RandomForestRegressor(featuresColfeatures, labelColprice) # 串成流水线 pipeline Pipeline(stages[assembler, scaler, rf]) # 训练并得到一个PipelineModel model pipeline.fit(train_df)Pipeline最好的地方在于避免数据泄漏。你在验证集上做预测时标准化的均值和方差应该来自训练集而不是验证集本身。如果这个操作手动做很容易出错。用Pipelinefit过程在训练集上完成所有stage的参数学习transform新数据时直接复用学到的参数不会把新数据的信息混进模型里。3.4 Pipeline模型的保存与部署训练好的PipelineModel可以一行代码保存到磁盘或HDFSmodel.write().overwrite().save(hdfs:///models/house_price_model)上线预测时另一个Spark任务直接加载from pyspark.ml import PipelineModel loaded_model PipelineModel.load(hdfs:///models/house_price_model) predictions loaded_model.transform(batch_df)这里有个关键坑加载预测时必须保证数据结构与训练时完全一致。你说训练时有age和income列预测时也必须提供同名同类型的列。VectorAssembler不会帮你自动对齐缺失特征它只会报Field mismatch错误。4. 实战用SparkML Pipeline搭建一个房价预测模型4.1 数据和场景选择选什么数据集练手一直是入门的老大难问题。Kaggle上的House Prices: Advanced Regression Techniques是个很好的起点有79个特征有缺失值有偏态分布很适合练特征工程。但它的文件有几百MB本地模式跑起来也不算快而且79列新手容易迷失。我这里换一个思路用Spark自带的样例代码自动生成一份模拟数据集数据量可控能快速验证整个Pipeline。主要是为了让刚上手的人先吃透流程等流程通了再上真实数据集。构造一份包含3个特征、1个标签的数据import random from pyspark.sql import SparkSession spark SparkSession.builder.appName(SparkMLDemo).getOrCreate() # 生成10000条模拟数据 rows [] for i in range(10000): size random.uniform(50, 200) # 房屋面积 rooms random.randint(1, 6) # 房间数 age random.randint(0, 50) # 房龄 price 3000 * size 20000 * rooms - 800 * age random.uniform(-10000, 10000) rows.append((size, rooms, age, price)) df spark.createDataFrame(rows, [size, rooms, age, price]) train_df, test_df df.randomSplit([0.8, 0.2], seed42)这里price是从三个特征线性组合再加噪声生成的所以模型能学到比较完美的规律。用这种模拟数据的好处是如果模型效果很差那肯定是代码写错了而不是数据本身太复杂。4.2 搭建Pipeline完整代码训练部分的代码思路如下from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.regression import LinearRegression from pyspark.ml.evaluation import RegressionEvaluator # 1. 特征拼装 assembler VectorAssembler(inputCols[size, rooms, age], outputColraw_features) # 2. 标准化 scaler StandardScaler(inputColraw_features, outputColfeatures) # 3. 线性回归 lr LinearRegression(featuresColfeatures, labelColprice) # 4. 串成Pipeline pipeline Pipeline(stages[assembler, scaler, lr]) # 5. 训练 lr_model pipeline.fit(train_df) # 6. 预测 predictions lr_model.transform(test_df) predictions.select(price, prediction).show(10) # 7. 评估 evaluator RegressionEvaluator(labelColprice, predictionColprediction, metricNamermse) rmse evaluator.evaluate(predictions) print(fRoot Mean Squared Error: {rmse})逻辑很简单但有几个细节必须说明。StandardScaler的作用是把量纲不同的特征统一到类似的尺度上。size的数值是几十到二百rooms是1到6age是0到50。如果不做标准化线性回归的梯度下降收敛会很慢size这个特征会主导更新方向。虽然在这份模拟数据里影响不大但在真实业务里特征量级差距可能达到几十万倍标准化是所有模型的准入门槛。输出大概是这样---------------------------------- | price| prediction| ---------------------------------- | 321269.64237511| 321211.014662157| | 348319.33933678| 348381.968999354| | 264703.44017008| 264611.853467603| ... Root Mean Squared Error: 8524.412RMSE是8524考虑到价格本身是几十万量级这个误差率不到3%对一个模拟数据管道来说已经足够说明Pipeline跑通了。4.3 从回归模型切换到随机森林线性回归是入门的第一个模型但它假设特征和标签是线性关系。真实业务里这个假设很难成立尤其是遇到用户行为、文本Embedding这类非线性特征时树模型往往效果更好。SparkML切模型非常顺滑。只改一行代码from pyspark.ml.regression import RandomForestRegressor rf RandomForestRegressor(featuresColfeatures, labelColprice, numTrees50, maxDepth10) pipeline Pipeline(stages[assembler, scaler, rf]) rf_model pipeline.fit(train_df)numTrees和maxDepth是随机森林最重要的两个超参数。numTrees越大、方差越小但训练时间线性增长一般50~200够用。maxDepth控制单棵树的复杂度太大容易过拟合太小欠拟合在调优时可以用ParamGridBuilder配合CrossValidator自动搜索。4.4 超参数搜索的正确姿势新手调参都是手动改参数反复跑既费时间又容易漏掉组合。SparkML自带的CrossValidator和ParamGridBuilder就是干这个的。from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import RegressionEvaluator param_grid ParamGridBuilder() \ .addGrid(rf.numTrees, [20, 50, 100]) \ .addGrid(rf.maxDepth, [5, 10, 15]) \ .build() evaluator RegressionEvaluator(labelColprice, predictionColprediction, metricNamermse) crossval CrossValidator( estimatorpipeline, estimatorParamMapsparam_grid, evaluatorevaluator, numFolds3, seed42 ) cv_model crossval.fit(train_df)CrossValidator会把训练集切成3份轮流把1份当验证集、2份当训练集每个超参数组合都跑3次。3个参数组合就是9次模型训练在小数据量上还行数据一多就很耗时。所以一般先用小范围粗网格搜索找到合适区间后再细调。在模拟数据上这样跑基本秒级完成你可以试试。5. 常见问题与排查技巧实录5.1 DataFrame列类型导致模型训练报错这个坑我踩过好多次。CSV文件里有一列叫income看起来是数字但Spark读取时把它推断成了字符串类型可能因为某些行里混了逗号或者空值。结果VectorAssembler在做类型检查时就报错了提示Data type StringType is not supported。排查思路不能瞎猜先用printSchema()看每一列的类型。如果确实推断了错误的类型用withColumn加cast来做转换from pyspark.sql.functions import col df df.withColumn(income, col(income).cast(double))5.2 数据倾斜导致训练慢如蜗牛数据倾斜是分布式计算绕不开的难题。数据里某个Key的样本特别多比如做用户分群时未知用户这个类别可能占了总量的70%。Spark会把这个类别对应的分区计算量搞得特别大其他Executor早算完了干等着整个任务就卡在那里。最简单的排查方法是用Spark UI看Stage的耗时柱状图如果某几个Task的Shuffle Read和计算时间明显长于其他Task基本就是倾斜了。处理倾斜的思路有三种一是过滤掉异常多的类别二是把倾斜的Key加上随机后缀打散后重分区这个办法有点脏但是能用三是用repartition重新调整分区数量和分布让数据尽量均匀。5.3 Pipeline保存后transform阶段报列找不到这个问题在模型上线时特别常见。你本地开发时DataFrame列名叫size到了生产环境上游给的列名叫house_size虽然数据内容一样但PipelineModel加载后会直接报错。对策是在预测入口处加一个数据对齐步骤先select出模型需要的列并改名。我习惯在训练之前就把所有的列名和类型约束固定好用一个字典维护列名映射避免预测阶段到处找名字。5.4 面试高频考点读取CSV与缓存策略热搜词里有绝密100个spark面试题这里先提前堵几个高频问题。第一个是CSV读取时的小文件问题。如果你直接spark.read.csv读几千个小文件会造成大量task调度开销远远大于计算开销。建议读入后写一次Parquet格式落地再后续分析性能提升非常明显。Parquet是列式存储读列很快还自带Schema和压缩是生产环境的首选。第二个是cache与persist的使用时机。如果一个DataFrame会被复用很多次再cache一下能省去重复计算。但不是所有DataFrame都该cache。如果只用一次cache反而白白占用内存。而且cache之后如果数据源原文件更新了你的缓存还是旧数据坑不吭。常见做法是在迭代训练或多次采样时cache训练集用完马上unpersist()。6. 关于面试、期末和更远的SparkML之路话题往回拉一下热搜词里出现了机器学习期末、吴恩达机器学习、机器学习 应用流程这些词说明很多同学是被课程或者面试推着来学的。这很正常我自己当年也是被面试题逼着把Spark源码翻了一遍。但学SparkML不能只背面试题得建立自己的知识网络。很多人学机器学习只学模型不知道如何把模型落到生产环境这就好比学做饭却不会生火。SparkML恰好提供了一个完整的工程视角数据读取、特征工程、模型训练、调优、保存、部署、预测一条链路下来你才算真的懂了一个模型的生命周期。给后来者三个建议。第一先把Spark基础数据结构吃透。DataFrame、Transformations、Actions、宽窄依赖这些概念不牢玩SparkML就像瘸着腿跑步。DataFrame的惰性求值机制会让你在不经意间写出重复计算的烂代码。第二多刷官方文档的示例程序。Spark官方仓库的examples/src/main/python/ml目录下全是算法实例。我当年就是把这些示例逐个跑了一遍理解才扎实。看一百篇博客不如手敲一遍代码。第三算法原理和分布式实现要结合起来学。你在吴恩达的课上学了逻辑回归的损失函数和梯度下降公式SparkML里LogisticRegression封装了这些但你要了解它是通过迭代多次MapReduce式的梯度聚合来完成训练的这直接影响到你对训练时间、资源占用、收敛速度的判断。回到热搜词里的dgx spark部署和spark 数据分析案例这些属于更加垂直的场景。DGX这类GPU服务器跑SparkML主要用于深度学习与特征工程的混合场景涉及GPU调度和Spark的集成复杂度更高也更吃硬件。这类我后面在深度学习的篇章里再展开讲。关于spark etl脚本和spark dbt 转换区别这属于Spark在数据工程侧的应用。ETL和机器学习并不是割裂的在真实的机器学习项目中ETL占的时间往往比训练模型还多。能用好DataFrame做清洗、聚合、宽表拼接才是做好SparkML的前提。dbt和Spark的关系一句话概括dbt负责在数仓里做建模和转换Spark负责更底层的分布式数据加工两者任务边界不同但经常配合使用。最后一句话收尾也是我做这一系列内容最想强调的学SparkML别贪多先把DataFrame、Pipeline、特征工程、一个回归模型完整跑通再逐步扩展。这一篇的内容你跟着做完SparkML的地基就打好了。下一篇我会接着写分类模型和模型调优的实战部分到时你想不踩坑地进阶跟着走就行。我自己的体会是技术这东西入门时最怕的就是一次接触太多概念。这篇把环境、数据结构、Pipeline和回归实战这几个主线掰开了讲就是希望你把骨架立起来。骨架稳了后面再怎么加肉都不慌。 Spark机器学习入门：DataFrame与Pipeline构建回归模型
刚拿到Spark Machine Learning(SparkML):机器学习(部分一)这个项目标题的时候我第一反应是这八成又是一个装了Spark但只会写WordCount的兄弟要开始整活了。毕竟我在社区里见到太多人问Spark怎么跑机器学习结果一聊才发现他是用Spark的RDD硬撸逻辑回归或者压根不知道SparkML和MLlib到底有啥区别。先说清楚这篇博文是干嘛的。它是Spark机器学习入门系列的第一篇目标只有一个帮你把SparkML这套分布式机器学习库的底裤扒干净让你知道它是什么、能干什么、怎么上手、踩过哪些坑。它适合这么几类人一是公司数据量已经大到单机Python跑不动、需要往分布式迁移的算法工程师二是刚学完Spark基础、想往机器学习方向进阶的大数据开发三是准备面试时被SparkML原理这类问题难住的同学。看完整篇文章你能自己搭起一个Spark环境用Pipeline把特征工程到模型训练串起来跑通一个完整的回归预测任务还能避开我当年踩过的内存和兼容性大坑。1. SparkML到底是何方神圣1.1 一个被问烂了的问题SparkML和MLlib什么关系这个事儿几乎每次技术分享都会有人问。很多老教程还停留在MLlib的叫法上这不能怪大家记性差因为历史原因确实绕。最早Spark贡献给Apache基金会的时候自带了一个机器学习库叫MLlib里面全是基于RDD的API。RDD是Spark最早的数据抽象优点是灵活缺点是太底层你在写机器学习代码时要自己操心数据怎么分布、哪个环节该cache、shuffle要怎么优化。后来Spark团队想通了推出了新的DataFrame API并且基于它重写了整个机器学习库新库就叫SparkML。所以现在的官方定位是SparkML是新一代基于DataFrame的机器学习库老版MLlib基于RDD的API进入维护状态不会再加新功能。判断你用的是哪个版本很简单看包名。org.apache.spark.mllib是老的RDD接口org.apache.spark.ml是新的DataFrame接口。我现在的项目全部基于ml这个包写。原因不复杂DataFrame自带Schema、Catalyst优化器、Tungsten执行引擎这些好东西Spark能做列裁剪、谓词下推和优化内存布局性能和易用性全面碾压RDD。1.2 为什么不能继续用Python直接撸很多算法出身的朋友问我我scikit-learn用得好好的为什么要学SparkML我的回答是如果数据量在单机内存能装下的范围内比如两三GB以内模型又不太复杂那scikit-learn确实是更顺手的工具。你没必要为了用Spark而用Spark这是典型的过度设计。但当你面临这几个场景时单机就扛不住了数据量到了几十GB甚至TB级别单机内存完全装不下训练样本是PB级日志里提取出来的需要用Spark做大规模特征工程你需要在同一个Pipeline里完成特征处理、模型训练、批量预测而且数据源分布在HDFS或S3上SparkML的核心优势是横向扩展。你在单机版sklearn里训练一个GBDT可能要几小时甚至OOMSparkML把训练任务拆分到集群的多个Executor上并行算资源够多的话能把时间压缩到几分钟。这个差异在真实业务里决定了一个模型能不能每天按时更新。1.3 SparkML官方生态的7个核心模块SparkML官网把功能分成几个大模块新手容易迷失在茫茫API里,这里直接给你整理成一张速查表模块官方包名核心内容典型场景特征工程ml.featureVectorAssembler、StandardScaler、StringIndexer、OneHotEncoder、PCA数据清洗、特征变换分类算法ml.classificationLogisticRegression、DecisionTreeClassifier、RandomForestClassifier、GBTClassifier、LinearSVC、NaiveBayes二分类、多分类回归算法ml.regressionLinearRegression、DecisionTreeRegressor、RandomForestRegressor、GBTRegressor、AFTSurvivalRegression价格预测、销量预测聚类算法ml.clusteringKMeans、BisectingKMeans、GaussianMixture、LDA用户分群、文本主题协同过滤ml.recommendationALS推荐系统、用户商品TopN模型调优ml.tuningCrossValidator、TrainValidationSplit、ParamGridBuilder超参数搜索模型评估ml.evaluationBinaryClassificationEvaluator、RegressionEvaluator、MulticlassClassificationEvaluator效果评估这篇先重点讲特征工程、回归和Pipeline这几个是后续所有算法的基础。分类和聚类放到系列后面几篇。2. 环境搭建Spark集群是绕不过去的一道坎2.1 本地模式还是集群模式很多人一开始就被搭建集群劝退了。我给你的建议是学习阶段完全可以用本地模式先把API和流程跑通再考虑集群。本地模式就是在你电脑上起一个Spark进程它能模拟Executor并行执行。对入门来说够用了也能体验到DataFrame懒加载、Stage划分这些特性。当你真正要把模型跑在几十GB的数据上时才需要考虑集群。集群形态有三类Standalone模式Spark自带的集群管理器配置最简单适合测试YARN模式Hadoop生态的集群管理器企业内部最常见资源管理能力强Kubernetes模式容器化部署适合云原生环境现在越来越流行本地学习环境建议直接装Spark并启用local模式。去官网下载一个与JDK版本兼容的二进制包比如JDK8对应Spark 3.x早期版本JDK11/17对应更新版本。解压后配置好JAVA_HOME然后跑一下以下命令# 解压Spark tar -zxvf spark-3.5.1-bin-hadoop3.tgz cd spark-3.5.1-bin-hadoop3 # 配置JAVA_HOME以mac或Linux为例 export JAVA_HOME/path/to/your/jdk export PATH$PATH:$JAVA_HOME/bin # 启动Spark本地模式的交互式Python环境 ./bin/pyspark能进入pyspark命令行就算安装成功了。2.2 集群搭建的核心步骤和三个大坑如果你真想搭建一个小的Standalone集群做实验思路很简单。一台主节点机器跑Master进程几台从节点机器跑Worker进程Worker启动时指定同一个Master地址即可。启动Master./sbin/start-master.sh启动Worker./sbin/start-worker.sh spark://master-host:7077启动后在浏览器里打开http://master-host:8080能看到当前集群有哪些Worker。搭建过程中最容易踩的坑有三个。第一个是通信超时问题集群机器之间如果不在同一个网段或者开了防火墙Worker注册到Master时经常注册不上表现就是启动Worker后Master页面一直看不到新节点。解决办法是确保机器之间网络互通检查防火墙或安全组尤其是在云服务器上8080和7077端口必须放行。第二个坑是内存超卖。Spark默认会给每个Worker分配机器全部内存作为Executor的候选但机器本身还要跑操作系统和Spark的Master进程真能被Executor用满时机器早就卡死了。配置时建议设置SPARK_WORKER_MEMORY为你机器实际可用内存的75%左右留点余量给系统。第三个坑是hosts配置混乱。集群之间的主机名互相解析不了会导致各种诡异的连接失败建议在所有节点上把master和worker的主机名都写进/etc/hosts。2.3 内存如何配置才不会频繁OOM我在社区回答过太多OOM问题了新手通病是只调spark.executor.memory实际上这不够。Spark内存分成了好几块核心参数有这四个参数作用建议初始值spark.executor.memoryExecutor堆内内存根据机器内存设一般4~8Gspark.executor.memoryOverheadJVM堆外内存堆内内存的10%~20%spark.driver.memoryDriver端内存一般2~4Gspark.sql.shuffle.partitionsShuffle分区数默认200根据数据量调整如果数据量大报错java.lang.OutOfMemoryError: Java heap space优先调大executor.memory同时按比例调大memoryOverhead。如果报的是Unable to acquire X bytes of memory这种往往是因为memoryOverhead给得太少。在yarn或standalone模式里这两块内存合起来不能超过spark.executor.memory加上overhead的上限。提示不要为了跑大数据量无脑把executor-memory搞到几十G。单个Executor内存太大时JVM的GC停顿会非常明显反而拖慢速度。更合理的做法是增加Executor数量而不是无限加大单个Executor内存。3. DataFrame是SparkML的操作对象3.1 从RDD到DataFrame数据结构的一次进化用过Spark旧版本的人都知道RDD用起来有多原始。你处理数据时脑子里想着的不是数据长什么样而是我该怎么把数据拆成一个个分区然后给每个分区写一个函数。DataFrame就不一样了它本质上是一张分布在集群里的表。每一列有类型、有名字每次操作Spark引擎会自动推演数据分布、决定哪些数据可以放在同一台机器上做计算哪些必须走shuffle。你不用手动管理分区它自己来。用代码说话。读一个CSV建DataFrame只需要一行df spark.read.option(header, true).option(inferSchema, true).csv(hdfs:///data/train.csv) df.printSchema() # 查看列名和类型 df.show(5) # 查看前5行这还不算完DataFrame背后拥有Catalyst优化器会自动对执行计划做优化。比如你只select两列Spark扫描文件时压根不会把其余列全部读进内存。这在处理宽表时收益巨大。3.2 Transformers和EstimatorsSparkML的两个核心抽象很多初学者对SparkML的API感到陌生是因为它引入了两个有门槛的概念Transformer和Estimator。Transformer是有transform方法的类作用是给DataFrame加列。最典型的特征是VectorAssembler把多列数值拼成一个特征向量。比如你有年龄、收入、点击次数三列经过VectorAssembler处理后DataFrame里会多出一列features这个column里装的是三元组向量。Estimator是有fit方法的类作用是学习参数。拿线性回归举例LinearRegression的fit方法会在DataFrame上训练产出LinearRegressionModel。这个产出的model本身也是一个Transformer它的transform方法能对新的DataFrame做预测追加一列predictions。简化理解Transformer是将一个DataFrame变成另一个DataFrame的函数;Estimator是在数据上训练出Transformer即模型的算法。Pipeline把这两者串起来。3.3 Pipeline如何避免数据泄漏Pipeline这个概念借鉴了scikit-learn它的核心是让多个stage按顺序执行。特征列拼装、标准化、训练、预测全串成一条流水线。from pyspark.ml import Pipeline from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.regression import RandomForestRegressor # 组装特征列 assembler VectorAssembler(inputCols[age, income, clicks], outputColraw_features) # 对特征做标准化均值0方差1 scaler StandardScaler(inputColraw_features, outputColfeatures) # 随机森林回归 rf RandomForestRegressor(featuresColfeatures, labelColprice) # 串成流水线 pipeline Pipeline(stages[assembler, scaler, rf]) # 训练并得到一个PipelineModel model pipeline.fit(train_df)Pipeline最好的地方在于避免数据泄漏。你在验证集上做预测时标准化的均值和方差应该来自训练集而不是验证集本身。如果这个操作手动做很容易出错。用Pipelinefit过程在训练集上完成所有stage的参数学习transform新数据时直接复用学到的参数不会把新数据的信息混进模型里。3.4 Pipeline模型的保存与部署训练好的PipelineModel可以一行代码保存到磁盘或HDFSmodel.write().overwrite().save(hdfs:///models/house_price_model)上线预测时另一个Spark任务直接加载from pyspark.ml import PipelineModel loaded_model PipelineModel.load(hdfs:///models/house_price_model) predictions loaded_model.transform(batch_df)这里有个关键坑加载预测时必须保证数据结构与训练时完全一致。你说训练时有age和income列预测时也必须提供同名同类型的列。VectorAssembler不会帮你自动对齐缺失特征它只会报Field mismatch错误。4. 实战用SparkML Pipeline搭建一个房价预测模型4.1 数据和场景选择选什么数据集练手一直是入门的老大难问题。Kaggle上的House Prices: Advanced Regression Techniques是个很好的起点有79个特征有缺失值有偏态分布很适合练特征工程。但它的文件有几百MB本地模式跑起来也不算快而且79列新手容易迷失。我这里换一个思路用Spark自带的样例代码自动生成一份模拟数据集数据量可控能快速验证整个Pipeline。主要是为了让刚上手的人先吃透流程等流程通了再上真实数据集。构造一份包含3个特征、1个标签的数据import random from pyspark.sql import SparkSession spark SparkSession.builder.appName(SparkMLDemo).getOrCreate() # 生成10000条模拟数据 rows [] for i in range(10000): size random.uniform(50, 200) # 房屋面积 rooms random.randint(1, 6) # 房间数 age random.randint(0, 50) # 房龄 price 3000 * size 20000 * rooms - 800 * age random.uniform(-10000, 10000) rows.append((size, rooms, age, price)) df spark.createDataFrame(rows, [size, rooms, age, price]) train_df, test_df df.randomSplit([0.8, 0.2], seed42)这里price是从三个特征线性组合再加噪声生成的所以模型能学到比较完美的规律。用这种模拟数据的好处是如果模型效果很差那肯定是代码写错了而不是数据本身太复杂。4.2 搭建Pipeline完整代码训练部分的代码思路如下from pyspark.ml.feature import VectorAssembler, StandardScaler from pyspark.ml.regression import LinearRegression from pyspark.ml.evaluation import RegressionEvaluator # 1. 特征拼装 assembler VectorAssembler(inputCols[size, rooms, age], outputColraw_features) # 2. 标准化 scaler StandardScaler(inputColraw_features, outputColfeatures) # 3. 线性回归 lr LinearRegression(featuresColfeatures, labelColprice) # 4. 串成Pipeline pipeline Pipeline(stages[assembler, scaler, lr]) # 5. 训练 lr_model pipeline.fit(train_df) # 6. 预测 predictions lr_model.transform(test_df) predictions.select(price, prediction).show(10) # 7. 评估 evaluator RegressionEvaluator(labelColprice, predictionColprediction, metricNamermse) rmse evaluator.evaluate(predictions) print(fRoot Mean Squared Error: {rmse})逻辑很简单但有几个细节必须说明。StandardScaler的作用是把量纲不同的特征统一到类似的尺度上。size的数值是几十到二百rooms是1到6age是0到50。如果不做标准化线性回归的梯度下降收敛会很慢size这个特征会主导更新方向。虽然在这份模拟数据里影响不大但在真实业务里特征量级差距可能达到几十万倍标准化是所有模型的准入门槛。输出大概是这样---------------------------------- | price| prediction| ---------------------------------- | 321269.64237511| 321211.014662157| | 348319.33933678| 348381.968999354| | 264703.44017008| 264611.853467603| ... Root Mean Squared Error: 8524.412RMSE是8524考虑到价格本身是几十万量级这个误差率不到3%对一个模拟数据管道来说已经足够说明Pipeline跑通了。4.3 从回归模型切换到随机森林线性回归是入门的第一个模型但它假设特征和标签是线性关系。真实业务里这个假设很难成立尤其是遇到用户行为、文本Embedding这类非线性特征时树模型往往效果更好。SparkML切模型非常顺滑。只改一行代码from pyspark.ml.regression import RandomForestRegressor rf RandomForestRegressor(featuresColfeatures, labelColprice, numTrees50, maxDepth10) pipeline Pipeline(stages[assembler, scaler, rf]) rf_model pipeline.fit(train_df)numTrees和maxDepth是随机森林最重要的两个超参数。numTrees越大、方差越小但训练时间线性增长一般50~200够用。maxDepth控制单棵树的复杂度太大容易过拟合太小欠拟合在调优时可以用ParamGridBuilder配合CrossValidator自动搜索。4.4 超参数搜索的正确姿势新手调参都是手动改参数反复跑既费时间又容易漏掉组合。SparkML自带的CrossValidator和ParamGridBuilder就是干这个的。from pyspark.ml.tuning import CrossValidator, ParamGridBuilder from pyspark.ml.evaluation import RegressionEvaluator param_grid ParamGridBuilder() \ .addGrid(rf.numTrees, [20, 50, 100]) \ .addGrid(rf.maxDepth, [5, 10, 15]) \ .build() evaluator RegressionEvaluator(labelColprice, predictionColprediction, metricNamermse) crossval CrossValidator( estimatorpipeline, estimatorParamMapsparam_grid, evaluatorevaluator, numFolds3, seed42 ) cv_model crossval.fit(train_df)CrossValidator会把训练集切成3份轮流把1份当验证集、2份当训练集每个超参数组合都跑3次。3个参数组合就是9次模型训练在小数据量上还行数据一多就很耗时。所以一般先用小范围粗网格搜索找到合适区间后再细调。在模拟数据上这样跑基本秒级完成你可以试试。5. 常见问题与排查技巧实录5.1 DataFrame列类型导致模型训练报错这个坑我踩过好多次。CSV文件里有一列叫income看起来是数字但Spark读取时把它推断成了字符串类型可能因为某些行里混了逗号或者空值。结果VectorAssembler在做类型检查时就报错了提示Data type StringType is not supported。排查思路不能瞎猜先用printSchema()看每一列的类型。如果确实推断了错误的类型用withColumn加cast来做转换from pyspark.sql.functions import col df df.withColumn(income, col(income).cast(double))5.2 数据倾斜导致训练慢如蜗牛数据倾斜是分布式计算绕不开的难题。数据里某个Key的样本特别多比如做用户分群时未知用户这个类别可能占了总量的70%。Spark会把这个类别对应的分区计算量搞得特别大其他Executor早算完了干等着整个任务就卡在那里。最简单的排查方法是用Spark UI看Stage的耗时柱状图如果某几个Task的Shuffle Read和计算时间明显长于其他Task基本就是倾斜了。处理倾斜的思路有三种一是过滤掉异常多的类别二是把倾斜的Key加上随机后缀打散后重分区这个办法有点脏但是能用三是用repartition重新调整分区数量和分布让数据尽量均匀。5.3 Pipeline保存后transform阶段报列找不到这个问题在模型上线时特别常见。你本地开发时DataFrame列名叫size到了生产环境上游给的列名叫house_size虽然数据内容一样但PipelineModel加载后会直接报错。对策是在预测入口处加一个数据对齐步骤先select出模型需要的列并改名。我习惯在训练之前就把所有的列名和类型约束固定好用一个字典维护列名映射避免预测阶段到处找名字。5.4 面试高频考点读取CSV与缓存策略热搜词里有绝密100个spark面试题这里先提前堵几个高频问题。第一个是CSV读取时的小文件问题。如果你直接spark.read.csv读几千个小文件会造成大量task调度开销远远大于计算开销。建议读入后写一次Parquet格式落地再后续分析性能提升非常明显。Parquet是列式存储读列很快还自带Schema和压缩是生产环境的首选。第二个是cache与persist的使用时机。如果一个DataFrame会被复用很多次再cache一下能省去重复计算。但不是所有DataFrame都该cache。如果只用一次cache反而白白占用内存。而且cache之后如果数据源原文件更新了你的缓存还是旧数据坑不吭。常见做法是在迭代训练或多次采样时cache训练集用完马上unpersist()。6. 关于面试、期末和更远的SparkML之路话题往回拉一下热搜词里出现了机器学习期末、吴恩达机器学习、机器学习 应用流程这些词说明很多同学是被课程或者面试推着来学的。这很正常我自己当年也是被面试题逼着把Spark源码翻了一遍。但学SparkML不能只背面试题得建立自己的知识网络。很多人学机器学习只学模型不知道如何把模型落到生产环境这就好比学做饭却不会生火。SparkML恰好提供了一个完整的工程视角数据读取、特征工程、模型训练、调优、保存、部署、预测一条链路下来你才算真的懂了一个模型的生命周期。给后来者三个建议。第一先把Spark基础数据结构吃透。DataFrame、Transformations、Actions、宽窄依赖这些概念不牢玩SparkML就像瘸着腿跑步。DataFrame的惰性求值机制会让你在不经意间写出重复计算的烂代码。第二多刷官方文档的示例程序。Spark官方仓库的examples/src/main/python/ml目录下全是算法实例。我当年就是把这些示例逐个跑了一遍理解才扎实。看一百篇博客不如手敲一遍代码。第三算法原理和分布式实现要结合起来学。你在吴恩达的课上学了逻辑回归的损失函数和梯度下降公式SparkML里LogisticRegression封装了这些但你要了解它是通过迭代多次MapReduce式的梯度聚合来完成训练的这直接影响到你对训练时间、资源占用、收敛速度的判断。回到热搜词里的dgx spark部署和spark 数据分析案例这些属于更加垂直的场景。DGX这类GPU服务器跑SparkML主要用于深度学习与特征工程的混合场景涉及GPU调度和Spark的集成复杂度更高也更吃硬件。这类我后面在深度学习的篇章里再展开讲。关于spark etl脚本和spark dbt 转换区别这属于Spark在数据工程侧的应用。ETL和机器学习并不是割裂的在真实的机器学习项目中ETL占的时间往往比训练模型还多。能用好DataFrame做清洗、聚合、宽表拼接才是做好SparkML的前提。dbt和Spark的关系一句话概括dbt负责在数仓里做建模和转换Spark负责更底层的分布式数据加工两者任务边界不同但经常配合使用。最后一句话收尾也是我做这一系列内容最想强调的学SparkML别贪多先把DataFrame、Pipeline、特征工程、一个回归模型完整跑通再逐步扩展。这一篇的内容你跟着做完SparkML的地基就打好了。下一篇我会接着写分类模型和模型调优的实战部分到时你想不踩坑地进阶跟着走就行。我自己的体会是技术这东西入门时最怕的就是一次接触太多概念。这篇把环境、数据结构、Pipeline和回归实战这几个主线掰开了讲就是希望你把骨架立起来。骨架稳了后面再怎么加肉都不慌。