1. 项目概述：当Django遇见Hadoop的化学反应
三年前接手公司短视频平台数据分析需求时，我面临一个典型的大数据困境：MySQL里的用户行为数据已经膨胀到每天500GB，传统的统计查询需要跑15分钟以上。这就是为什么我们需要将Django的敏捷开发能力与Hadoop的分布式计算能力相结合——前者提供优雅的业务逻辑封装和可视化界面，后者解决海量数据的存储与计算瓶颈。
这个架构的核心价值在于：通过Django REST Framework构建的数据API层，将Hadoop集群的计算结果以毫秒级响应呈现给前端。我曾用这套架构处理过单日20亿条播放记录的分析需求，在8台Worker节点的Hadoop集群上，Spark作业能在8分钟内完成全量数据的清洗和指标计算，而Django后台只需从Redis缓存中读取预计算好的JSON数据即可。
2. 技术栈选型背后的血泪史
2.1 为什么是Django + Hadoop组合？
2019年我们最初尝试用纯Spark做全栈开发，但很快发现两个致命问题：一是Thrift Server的JDBC接口在复杂业务查询时性能急剧下降；二是缺乏成熟的模板渲染方案导致前端开发效率低下。后来改用Django作为中台层，主要基于以下考量：
ORM的折衷方案：Django的Model层既能兼容MySQL这类关系型数据库（存储计算结果），又能通过自定义Manager接入HBase等NoSQL（原始日志存储）。我们扩展了Django的数据库路由机制，使读操作自动路由到Hive镜像库，写操作走MySQL主库。
DRF的API生产力：相比Spring Boot，Django REST Framework的序列化器能减少30%的接口代码量。特别是在处理嵌套的推荐结果时，用
SerializerMethodField可以灵活组合来自不同数据源的结果。Admin的隐藏价值：内置的Admin后台经过定制后，成为数据质量监控的利器。我们开发了自定义Action，能直接触发Spark作业的重新计算。
2.2 Hadoop生态组件的精准打击
在数据层我们采用组合战术：
- HDFS：存储原始日志文件，采用冷热数据分层策略。热数据（最近7天）保留3副本，冷数据（历史数据）降为2副本并启用压缩。
- Spark SQL：主力计算引擎，比MapReduce快10倍的关键在于：
# 启用动态分区优化 spark.conf.set("hive.exec.dynamic.partition", "true") spark.conf.set("hive.exec.dynamic.partition.mode", "nonstrict") # 使用DataFrame API而非RDD df = spark.read.parquet("hdfs://logs/daily") .selectExpr("user_id", "video_id", "CAST(play_time AS DOUBLE)") .filter("event_date = '2023-07-15'") - Hive：数据仓库层使用ORC格式存储，压缩比达到5:1。通过
STORED AS ORC和TBLPROPERTIES ("orc.compress"="SNAPPY")声明。 - Kafka：消息队列选用0.11以上版本，关键配置：
# 生产者端 compression.type=snappy linger.ms=20 batch.size=65536 # 消费者组 isolation.level=read_committed enable.auto.commit=false
3. 数据管道的实战细节
3.1 日志采集的五个陷阱
用Flume收集Nginx日志时，我们踩过这些坑：
时间戳陷阱：不同服务器时区不一致导致的事件乱序。解决方案是在Flume拦截器中强制转UTC：
event.getHeaders().put("timestamp", Instant.now().atZone(ZoneOffset.UTC).format(DateTimeFormatter.ISO_INSTANT));反压问题：Kafka集群故障时Flume内存堆积。需要调整channel参数：
agent.channels.memory.type = memory agent.channels.memory.capacity = 50000 agent.channels.memory.transactionCapacity = 5000字段污染：用户输入的非法字符破坏Hive表结构。必须用正则过滤器清洗：
from pyspark.sql.functions import regexp_replace df = df.withColumn("comment", regexp_replace(col("comment"), "[\u0000-\u001f]", ""))
3.2 Hive表设计的艺术
用户行为日志表采用分层分区策略：
CREATE EXTERNAL TABLE user_events ( user_id BIGINT, video_id STRING, event_type STRING, play_time DOUBLE, client_ip STRING ) PARTITIONED BY ( dt STRING COMMENT '日期分区yyyy-MM-dd', hour STRING COMMENT '小时分区HH' ) STORED AS ORC LOCATION '/data/events';每日通过Spark动态添加分区：
spark.sql(f""" ALTER TABLE user_events ADD PARTITION (dt='{date}', hour='{hour}') LOCATION '/data/events/dt={date}/hour={hour}' """)4. 推荐算法的工程化落地
4.1 混合推荐架构
我们融合了两种算法：
ItemCF：基于物品的协同过滤，计算余弦相似度
from pyspark.mllib.recommendation import ALS model = ALS.train(ratings, rank=10, iterations=10)随机森林：处理用户特征
from pyspark.ml.classification import RandomForestClassifier rf = RandomForestClassifier(featuresCol="features", labelCol="label")
4.2 实时推荐实现
Django视图层的关键代码：
class RecommendView(APIView): def get(self, request): user_id = request.user.id # 从Redis获取预计算结果 cache_key = f"rec:{user_id}" data = cache.get(cache_key) if not data: # 触发实时计算 data = calculate_realtime_rec(user_id) cache.set(cache_key, data, timeout=3600) return Response(data)5. 性能优化的七种武器
Redis多级缓存：
# 第一层：本地内存缓存 @cache_page(60 * 15) @method_decorator(cache_control(private=True), name='dispatch') class VideoListView(ListView): pass # 第二层：Redis缓存 CACHES = { 'default': { 'BACKEND': 'django_redis.cache.RedisCache', 'LOCATION': 'redis://:password@redis-host:6379/1', 'OPTIONS': { 'CLIENT_CLASS': 'django_redis.client.DefaultClient', 'COMPRESSOR': 'django_redis.compressors.lzma.LzmaCompressor', } } }Celery任务拆分：
@shared_task(bind=True, rate_limit='100/m') def process_batch(self, batch_ids): try: data = fetch_from_hadoop(batch_ids) store_to_mysql(data) except Exception as e: self.retry(exc=e, countdown=60)
6. 监控体系的建设
用Prometheus+Grafana搭建的监控看板需要关注这些指标：
| 指标名称 | 报警阈值 | 采集方式 |
|---|---|---|
| Spark任务失败率 | >5% (15分钟) | YARN API |
| Django请求延迟(P99) | >800ms | Prometheus客户端 |
| HDFS存储空间使用率 | >85% | JMX导出器 |
| Kafka消费延迟 | >1000消息 | Consumer Lag监控 |
7. 从实验室到生产环境的教训
数据倾斜处理：当某个网红视频的播放量占总量30%时，Spark作业会卡在最后一个Reducer。解决方案：
# 添加随机前缀打散热点 df = df.withColumn("video_id", when(col("video_id") == "hot_video", concat(lit("prefix_"), floor(rand()*10)), col("video_id")))Django连接池配置：
DATABASES = { 'default': { 'ENGINE': 'django.db.backends.mysql', 'CONN_MAX_AGE': 300, 'OPTIONS': { 'connect_timeout': 3, 'read_timeout': 5, 'write_timeout': 5, 'pool_size': 20, 'max_overflow': 10, } } }
这套架构经过三年迭代，目前支撑着日均1.2亿活跃用户的短视频平台。最大的体会是：大数据系统不是组件的简单堆砌，而是要让每个层级发挥其不可替代的价值——Hadoop负责"海量"，Django专注"精确"，而工程师要做的是在两者之间找到最佳平衡点。
