前阵子做了一个图书类别自动标注系统正好把 Hadoop、Spark、Django 和机器学习串成了一整条链路标题看着像课程设计但真做下来涉及的坑一点都不少。从大数据环境搭建到数据清洗再到模型训练和可视化大屏每一步都值得单独拎出来讲讲。这篇就把整个项目的设计思路、核心模块、实际调试过程和踩过的坑一次性整理清楚给打算做类似大数据机器学习 Web 项目的朋友一个可以直接参考的版本。1. 系统整体设计与技术选型思路1.1 为什么选 Hadoop Spark Django 这套组合很多人在拿到“图书类别自动标注”这个需求时第一反应是直接写个 Python 脚本把数据集读进来用 sklearn 训练一个分类模型再接个 Flask 接口就完事。这样做确实能跑通 Demo但离真实业务场景还差得远。真实环境里图书数据往往量级很大——几十万本、上百万本书的元数据包含书名、简介、目录、出版社、用户标签等字段单机内存根本存不下训练和推理的效率也会成为瓶颈。Hadoop 在这里承担的是分布式存储和资源管理的角色。图书原始数据以文本文件或 JSON 格式上传到 HDFS利用 HDFS 的多副本机制保证数据不丢同时让后续 Spark 任务能够就近读取数据而不是每次从本地磁盘加载大量文件。Spark 负责分布式计算无论是数据清洗、特征提取还是模型训练都可以转换成 Spark 上的分布式任务来跑配合 MLlib 自带的 TF-IDF、Word2Vec 和分类器能直接把特征工程和模型训练放到一个 Pipeline 里完成。Django 则负责把训练好的模型包装成 Web 服务。一个系统光有模型不行得有管理后台、API 接口、前端展示Django 自带 Admin 和 ORM开发效率很高配合 Django REST Framework 能快速提供 RESTful 接口。可视化大屏的数据也由 Django 统一输出图表所需的统计结果可以回存到 MySQL由 Django 读取后返回给前端。这套组合的关键在于Hadoop 解决数据存储和集群资源问题Spark 解决计算效率问题Django 解决业务落地问题。三个环节各司其职整个系统的扩展性也比单体脚本好得多。1.2 系统模块划分与整体流程我在设计时把系统拆成了四大模块数据采集与存储模块负责接入图书基础数据包括批量导入、格式校验、写入 HDFS数据处理与模型训练模块基于 Spark 完成数据清洗、分词、TF-IDF 特征构建、模型训练和评估业务服务模块基于 Django 提供图书管理、自动标注、人工审核、类别统计等接口可视化大屏模块基于 ECharts 展示图书类别分布、标注准确率、数据量趋势等核心指标。数据流向是原始图书数据通过 Django 管理台上传一方面写入 MySQL 做业务备份另一方面同步写入 HDFSSpark 定期从 HDFS 拉取全量数据进行特征提取和模型增量训练训练完成的模型序列化保存到共享目录或 HDFSDjango 启动时加载模型当用户提交一本书时先提取该书特征再调用模型预测类别结果写回 MySQL 并展示到大屏上。这个设计最核心的取舍是训练链路和推理链路分开。训练跑在 Spark 集群上推理跑在 Django 进程里因为单条数据预测用 Spark 反而更重本地加载模型直接预测的延迟只有几毫秒更符合 Web 接口的要求。2. 基于 Hadoop 与 Spark 的大数据环境搭建2.1 Hadoop 伪分布式集群搭建要点如果机器资源有限可以先从 Hadoop 伪分布式模式开始也就是用一个 Java 进程模拟分布式环境但在配置和路径上要做到和真实集群一致后面换多台机器时直接复用配置。我用的版本是 Hadoop 3.3.5部署在 Ubuntu 20.04 上。安装步骤大致如下配置 Java 环境建议 JDK 8Hadoop 3.x 对 JDK 11 兼容性也还行但很多踩坑案例都集中在 JDK 版本不匹配上稳妥起见选 JDK 8设置 SSH 免密登录这是启动守护进程的前提修改core-site.xml、hdfs-site.xml、yarn-site.xml、mapred-site.xml四个配置文件格式化 NameNode格式化命令是hdfs namenode -format执行start-dfs.sh和start-yarn.sh启动服务。配置文件中容易忽略的地方有两个。一个是core-site.xml里必须设置fs.defaultFS为hdfs://localhost:9000否则 HDFS 的地址无法识别。另一个是hdfs-site.xml里dfs.replication的取值伪分布式模式下虽然副本数设置成 1 就够了但为了后面迁移到真实集群我直接设成 3这样分布式部署后无需再改。实际启动后可以通过jps命令查看进程是否齐全正常应该看到 NameNode、DataNode、SecondaryNameNode、ResourceManager、NodeManager 五个进程。2.2 Spark 环境部署与运行模式切换Spark 我用的版本是 3.4.0与 Hadoop 3.x 兼容。部署完成后建议优先用 local 模式跑通数据清洗脚本因为 local 模式不需要额外维护集群适合在开发阶段快速验证。等脚本稳定后再通过spark-submit --master yarn指定提交到 YARN 上这时候数据读取走 HDFS计算任务由 YARN 分配资源。这里有一个容易踩坑的点Spark 读取 HDFS 数据时本地模式也可以读但如果你在 Windows 本机跑需要额外配置 HADOOP_HOME 和 winutils.exe否则会报NullPointerException或Failed to locate the winutils binary。我建议开发机直接用 Linux 虚拟机或者云主机省掉这些兼容性问题。Spark 启动前还要注意内存配置。默认的spark.driver.memory只有 1gspark.executor.memory也偏小处理几万条样本时可能没问题但到了几十万条Executor 会直接 OOM。我在启动脚本里做了调整spark-submit \ --master yarn \ --deploy-mode client \ --driver-memory 4g \ --executor-memory 4g \ --executor-cores 2 \ --num-executors 3 \ train_book_classifier.py如果只是想验证一个简单逻辑不提交到 YARN可以直接用spark-submit --master local[4]local[4]表示本地用 4 个线程并发执行任务很适合在单机调试时模拟分布式效果。2.3 ZooKeeper 的整合与元数据协调作用标题里提到了 Hadoop 和 ZooKeeper 整合这在多节点集群里主要用于 NameNode HA。但即便是伪分布式或单机环境Spark 的一些运行模式也可能依赖 ZooKeeper 做协调服务比如 Spark Standalone 模式下的 Master 高可用以及使用 Hive Metastore 时的元数据共享。当前项目里 ZooKeeper 的作用主要有两个如果后面想把 Hadoop NameNode 做成自动故障切换需要用 ZooKeeper 管理两个 NameNode 的选举状态如果 Spark 开启了 Hive 集成Metastore 服务可以通过 ZooKeeper 实现多实例协调。ZooKeeper 的部署很简单解压后复制conf/zoo_sample.cfg为conf/zoo.cfg设置dataDir和clientPort2181然后启动即可。需要注意的是ZooKeeper 集群一般要求奇数个节点单机环境就部署一个但也必须保持服务常驻否则依赖它的组件会启动失败。注意如果只是用 Spark 的 YARN 模式ZooKeeper 不是必须的。一开始不需要急着装先把 Hadoop 和 Spark 的核心流程跑通等需要 HA 或 Hive 集成时再引入这样排障范围会更小。3. 机器学习图书分类模型训练与标注实现3.1 图书数据预处理与标签体系设计训练一个可靠的图书分类器前提是数据干净、标签体系合理。我用的数据集包含大约 8 万条图书元数据字段包括书名、作者、出版社、简介、分类。原始数据里存在不少问题简介字段大量为空分类字段有的细到三级类目、有的只到一级类目还有一些标签明显是错的。我的处理思路是先把类目统一映射到一级分类比如“计算机科学—程序设计—Java编程”全部归并到“计算机”这样能保证每个类别样本量足够模型学习起来也更稳定。最终定下的类别有文学、历史、计算机、经济管理、哲学心理、艺术设计、教育、科学百科、社会科学等十几个一级分类。对于缺失简介的图书用书名加作者名作为文本输入必要时通过其他元数据字段拼接补充。这一步在 Spark 里非常顺手因为用 DataFrame 的withColumn和fillna操作就能分布式完成几千行代码就能搞定。3.2 特征提取方案与分类模型选型文本分类的特征提取我选择了 TF-IDF 加 Word2Vec 做对比实验。TF-IDF 适合词表规模适中的场景实现简单训练快在 8 万条数据上效果稳定。Word2Vec 需要通过训练语料生成词向量再把文档中所有词的向量取平均作为文档向量优点是能保留部分语义信息但训练时间和调参成本更高。分类模型方面对比了逻辑回归、朴素贝叶斯和支持向量机。实际效果是逻辑回归在 TF-IDF 特征上表现最稳定准确率大约在 88% 左右朴素贝叶斯训练速度快但对类别不均衡更敏感SVM 在小样本上表现好但数据量增大后训练时间显著上升。考虑到这个项目要求有“机器学习”的核心训练流程同时最终要上线提供实时预测我最终选了带 TF-IDF 的 LogisticRegression。分词处理用的是 jieba这里有一个分布式上的坑Spark 的mapPartitions中调用 jieba 时每个 Executor 需要独立加载词典如果词典较大会在每个节点重复占用内存。解决办法是把 jieba 的自定义词典放到共享路径并在初始化分区时统一加载。3.3 Spark MLlib 训练流程与模型持久化MLlib 的 Pipeline 方式和 sklearn 非常像定义好各个 Stage 后直接调用fit()就能完成训练。from pyspark.ml.feature import HashingTF, IDF, Tokenizer from pyspark.ml.classification import LogisticRegression from pyspark.ml import Pipeline from pyspark.ml.evaluation import MulticlassClassificationEvaluator tokenizer Tokenizer(inputColtext, outputColwords) hashing_tf HashingTF(inputColwords, outputColrawFeatures, numFeatures10000) idf IDF(inputColrawFeatures, outputColfeatures) lr LogisticRegression(maxIter50, regParam0.01) pipeline Pipeline(stages[tokenizer, hashing_tf, idf, lr]) train_df, test_df data.randomSplit([0.8, 0.2], seed42) model pipeline.fit(train_df) predictions model.transform(test_df) evaluator MulticlassClassificationEvaluator(labelCollabel, predictionColprediction, metricNameaccuracy) accuracy evaluator.evaluate(predictions) print(fAccuracy: {accuracy})训练完成后的模型保存有两种方式。一种是用 MLlib 自带的model.save()保存为 Pipeline 模型目录但 Web 端读取时依赖 PySpark 环境起一个 SparkContext 才能加载对于轻量级 API 服务来说偏重。另一种是把向量化和分类模型的核心参数抽取出来转成标准格式后落盘再把特征提取逻辑在 Django 里用同等 Python 库实现一遍。我实际采用的是第二种方案用 MLlib 训练和评估模型确认指标没问题后将 TF 的词典和 IDF 权重导出成 JSON 文件同时把逻辑回归的权重和偏置导出Django 端加载这些参数结合 jieba 分词和哈希映射实现预测。这样 API 服务完全不需要 Spark 环境只有离线训练才使用 Spark部署成本降低一个量级。4. Django 后端 API 与业务逻辑实现4.1 Django 工程初始化与 App 划分Django 工程我取名book_platform创建了三个 Appbooks、annotation、dashboard。职责划分如下books图书数据的 CRUD、导入导出与 MySQL 表book_info对应annotation自动标注、人工审核、模型调用dashboard大屏统计接口负责汇总类别分布、标注趋势、模型准确率等数据。创建 App 的命令很简单django-admin startproject book_platform python manage.py startapp books python manage.py startapp annotation python manage.py startapp dashboard创建好后需要在settings.py中注册到INSTALLED_APPS同时配置 MySQL 连接。数据库我用的是 MySQL 8.0Django 默认的 driver 是 MySQLdb通常需要安装mysqlclient。如果安装遇到编译错误可以先安装系统依赖再重试。4.2 核心数据表设计与 RESTful API 构建图书详表book_info包含id、book_name、author、publisher、description、manual_category、auto_category、status、created_at等字段。重点字段是manual_category和auto_category其中manual_category是标准答案auto_category是模型预测结果两者对比可以计算准确率。API 层我基于 Django REST Framework 实现用ModelViewSet提供标准的增删改查接口。比如标注接口class AnnotationViewSet(viewsets.ModelViewSet): queryset BookInfo.objects.all() serializer_class BookInfoSerializer action(detailFalse, methods[post]) def auto_label(self, request): book_id request.data.get(book_id) book BookInfo.objects.get(idbook_id) category predictor.predict(book.description or book.book_name) book.auto_category category book.save() return Response({category: category})Django 执行查询和删除对象时有几个细节必须注意。用get查询不存在的记录会抛出DoesNotExist异常需要在视图里用try-except或get_object_or_404处理。批量删除时如果有关联数据要留意on_delete级联策略避免误删。还有一点是 ORM 的save()默认会更新全部字段在高并发场景下要显式指定update_fields否则可能覆盖其他字段的变更。4.3 自动标注服务与异步任务设计刚做完时我发现一个问题如果每来一本书都同步调用模型预测接口响应时间在 30-50 毫秒左右单本书没问题但如果需要给存量 8 万本书批量打标签同步接口就会超时。于是我把批量标注任务放到了 Celery 里执行Django 只负责接收请求、创建任务、返回任务 ID后台 Celery Worker 异步处理处理结果通过 WebSocket 主动推送到前端。Celery 的 Broker 我用的 Redis配置也很直接CELERY_BROKER_URL redis://localhost:6379/0 CELERY_RESULT_BACKEND redis://localhost:6379/1创建任务的代码shared_task def batch_auto_label(book_ids): predicted 0 for book_id in book_ids: book BookInfo.objects.get(idbook_id) category predictor.predict(book.description or book.book_name) book.auto_category category book.save(update_fields[auto_category]) predicted 1 return predictedDjango 视图里只需要batch_auto_label.delay(book_ids)即可。这样设计的好处是耗时任务不阻塞 Web 请求大屏端能看到任务进度体验比一直转圈强得多。5. 可视化大屏数据展示与实时推送5.1 大屏布局与前端组件选型可视化大屏是这个项目里最出效果的部分。大屏设计采用 1920x1080 的固定分辨率左右两侧放图表中间顶部放核心指标整体色调以深蓝和青色为主。前端技术栈是 Vue 3 ECharts 5通过 Ajax 从 Django 获取数据。大屏包含的图表模块有图书类别分布饼图各类别数量排行柱状图近 30 天新增图书趋势折线图自动标注与人工标注准确率对比图核心指标卡片包括总图书数、今日新增、平均标注置信度、模型准确率。ECharts 图表只需要准备好option对象设置series数据和xAxis、yAxis即可。数据格式方面我让 Django 接口统一返回{data: [...]}的结构方便前端直接使用。5.2 大屏图表数据接口设计与性能优化仪表盘接口和业务接口要分开大屏数据要求一次返回完整汇总数据避免多次请求导致加载闪屏。我在dashboard/views.py里写了一个聚合接口def overview(request): total_books BookInfo.objects.count() category_dist (BookInfo.objects.values(auto_category) .annotate(countCount(id)) .order_by(-count)) trend (BookInfo.objects .filter(created_at__gtetimezone.now() - timedelta(days30)) .annotate(dayTruncDate(created_at)) .values(day) .annotate(countCount(id))) return JsonResponse({ total_books: total_books, category_dist: list(category_dist), trend: list(trend), accuracy: accuracy_service.get_recent_accuracy(), })这里有个性能问题每刷新一次页面就执行 4-5 条聚合 SQL虽然数据量不大时没问题但大屏通常固定刷新频率比如 60 秒一次没必要每次都查数据库。我加了 Redis 缓存缓存时间为 30 秒命中缓存时直接返回数据库压力明显下降。5.3 Django WebSocket 实现后台数据主动推送大屏上有一个“最新标注动态”模块要求后台完成图书自动标注后前端不用手动刷新就能看到新的记录。实现方案是 Django Channels 的 WebSocket。在settings.py中注册channels配置ASGI_APPLICATION然后编写消费者class DashboardConsumer(AsyncWebsocketConsumer): async def connect(self): await self.accept() await self.channel_layer.group_add(dashboard, self.channel_name) async def disconnect(self, close_code): await self.channel_layer.group_discard(dashboard, self.channel_name) async def book_update(self, event): await self.send(text_datajson.dumps(event[data]))后端在自动标注任务里每处理完一批数据就通过channel_layer.group_send推送一条消息async_to_sync(channel_layer.group_send)( dashboard, { type: book_update, data: { book_name: book.book_name, category: category, time: datetime.now().strftime(%Y-%m-%d %H:%M:%S), } } )前端只要建立 WebSocket 连接收到消息后把新的记录插入表格、同时刷新统计图表的某个 series 即可。这里比较关键的一点是Django 默认是同步框架和 Channels 的异步消费者结合时要注意同步代码和异步代码的切换比如在同步视图里使用async_to_sync不然会碰到线程冲突的报错。6. 调试过程与经典踩坑记录6.1 Hadoop 启动失败与端口冲突排查做 Hadoop 相关项目NameNode 进程起不来或者一会就挂是最常见的问题。我遇到过的情况是dfs.namenode.http.address端口被占用导致 Web UI 无法访问。排查方法是netstat -tlnp | grep 9870先看端口是否被占再用jps确认进程是否真的存在。还有一个非常容易中招的坑NameNode 已经格式化过一次之后因为配置修改又格式化了第二次结果 DataNode 的clusterID和 NameNode 不一致DataNode 一直起不来。解决办法是停止 HDFS 后把data和name目录下的数据清空重新格式化。这个过程会丢 HDFS 里的数据所以只在伪分布式环境或者数据可以重建时才这样做。YARN 任务提交失败通常和资源有关。伪分布式模式下本机内存如果偏小ResourceManager 分配给任务的yarn.nodemanager.resource.memory-mb默认可能只有 1gSpark 任务请求 4g 内存就直接被拒绝。调试时要先看 YARN 的调度日志确认是否因为资源不足导致 App 被 kill。6.2 Spark 执行日志查看与内存溢出处理Spark 任务出问题第一反应是去 YARN 的日志或者 Spark Web UI 里看 Executor 的状态。我习惯在spark-submit任务跑完后打开 Spark History Server查看每个 Stage 的 Shuffle Read/Write 大小和 Executor 的 GC 时间定位数据倾斜或内存溢出的具体 Task。OOM 的常见原因是数据分区不均匀。8 万条数据看起来不多但如果某个类别占据绝大多数样本按类别分组后继续处理时个别 Executor 会扛下大量数据。我采用了两个缓解措施给 DataFrame 做重分区repartition(partitions)让数据分散到更多分区调整 Executor 内存的同时设置spark.memory.offHeap.enabledtrue和spark.memory.offHeap.size2g给 JVM 堆外内存留出空间。另外用cache()缓存中间结果时要注意如果后续不再复用就及时unpersist()否则内存里堆的 RDD 太多也会触发 GC 频繁和 OOM。6.3 Django 数据库查询、跨域与序列化问题Django 后端联调时遇到的第一个问题是前端访问接口报跨域错误。DRF 默认不允许跨域需要安装django-cors-headers在settings.py中配置INSTALLED_APPS [corsheaders, ...] MIDDLEWARE [corsheaders.middleware.CorsMiddleware, ...] CORS_ALLOW_ALL_ORIGINS True开发阶段可以全部放行上线前再改成白名单。另一个坑是 Django 的DateTimeField默认不支持直接 JSON 序列化。返回时间字段时DRF 的DateTimeField会转成 ISO 格式字符串但如果自己手写JsonResponse忘记转换就会抛TypeError: Object of type datetime is not JSON serializable。解决方法是统一用 DRF 序列化器或手写转换函数from datetime import datetime def to_jsonable(obj): if isinstance(obj, datetime): return obj.strftime(%Y-%m-%d %H:%M:%S) return obj6.4 模型预测效率优化与缓存策略刚开始模型预测接口平均耗时 50 毫秒多个并发请求下 Django 线程阻塞明显。后来做了几个优化耗时降到 10 毫秒左右分词结束后只保留列表不保留中间字符串对象减少内存分配和 GC将 TF-IDF 特征映射表预加载为字典预测时直接用字典查找替代大数组遍历对预测概率最高的类别设置置信度阈值如果置信度低于 0.6 就返回“待人工确认”避免模型把不确定的样本强行归类。系统上线后每天都会有新图书数据进来所以我还加了一个简单的增量训练策略每周日凌晨用过去一周的人工标注数据重新训练一次模型训练完成后用新的参数文件替换线上模型替换动作通过版本号控制模型文件不直接覆盖方便随时回滚。7. 项目完整运行流程与复现说明7.1 从原始数据到模型上线的全链路如果要从零复现这个项目建议按下面顺序操作启动 Hadoop HDFS将原始图书数据文件上传到 HDFS 的/data/books/目录在开发环境跑train_book_classifier.py用 Spark 读取 HDFS 数据完成数据清洗、特征工程、模型训练和评估评估指标达到预期后导出模型参数到指定的 JSON 文件启动 Django 服务和 Celery Worker确认管理后台能正常上传、查询图书数据启动前端大屏项目确认接口连通WebSocket 推送正常进入管理后台选择一批未标注图书触发批量自动标注任务观察大屏上的最新动态和类别分布是否实时更新。每一步都有对应的验证方法。Hadoop 环境用hdfs dfs -ls /data/books验证数据是否到位Spark 训练完打印 accuracyDjango 接口用 Postman 传入一本新书确认返回类别大屏刷新看图表是否出现新数据。7.2 启动脚本与常见启动顺序为了让运维和本地调试都方便我写了一个启动脚本按依赖顺序拉起服务# 1. 启动 Hadoop start-dfs.sh start-yarn.sh # 2. 启动 ZooKeeper如果启用 HA 或 Hive Metastore zkServer.sh start # 3. 启动 Spark HistoryServer可选便于查看任务日志 $SPARK_HOME/sbin/start-history-server.sh # 4. 启动 Django python manage.py runserver 0.0.0.0:8000 # 5. 启动 Celery Worker celery -A book_platform worker -l info # 6. 启动前端大屏 npm run serve启动顺序是有讲究的。先启动依赖底层服务Hadoop、ZooKeeper再启动上层应用Django、Celery不然 Django 启动时如果去连 HDFS 或 Redis服务还没起好就会报错。我实际跑的时候发现Celery Worker 如果先于 Redis 启动也会反复重连虽然最终能连上但日志会很乱。统一先启动基础设施流程清晰很多。最后分享一点实际操作中的体会这个项目做完我最大的感受是“技术栈多”不等于“复杂度高”真正麻烦的是模块之间的衔接。Hadoop、Spark、Django 各自都能跑但串在一起时数据格式、对象序列化、资源消耗的差异就会被放大。最开始我在 Spark 里处理完的数据直接用自带的save()保存结果 Django 那边为了加载这个模型被迫引入整套 PySpark部署体积直接膨胀后来狠下心重构成了参数导出方案才让 API 服务真正变轻。还有一个建议做这类综合项目一定先把数据流画清楚再动手。我前期直接在代码里东补一块西补一块改到最后发现book_info表的字段结构和模型特征长度对不上又回头改表结构浪费了不少时间。如果一开始就确定“原始数据进 HDFS → Spark 清洗训练 → 参数落盘 → Django 加载预测 → MySQL 存储结果 → 大屏展示”这条链路每个环节的输入输出都很明确后面的开发会顺畅得多。如果你想在现有系统上继续扩展可以从这几个方向入手把单级分类改成多标签分类让一本书同时属于多个类别给标注结果加入置信度排序和人工复审工作流或者把 Spark 的训练任务改成定时增量训练让模型随着标注数据的积累持续迭代。这几点做到位整个系统就从一个课程设计级别的东西变成了一个接近生产可用的小平台。