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 BootDjango 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.compressSNAPPY)声明。Kafka消息队列选用0.11以上版本关键配置# 生产者端 compression.typesnappy linger.ms20 batch.size65536 # 消费者组 isolation.levelread_committed enable.auto.commitfalse3. 数据管道的实战细节3.1 日志采集的五个陷阱用Flume收集Nginx日志时我们踩过这些坑时间戳陷阱不同服务器时区不一致导致的事件乱序。解决方案是在Flume拦截器中强制转UTCevent.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, rank10, iterations10)随机森林处理用户特征from pyspark.ml.classification import RandomForestClassifier rf RandomForestClassifier(featuresColfeatures, labelCollabel)4.2 实时推荐实现Django视图层的关键代码class RecommendView(APIView): def get(self, request): user_id request.user.id # 从Redis获取预计算结果 cache_key frec:{user_id} data cache.get(cache_key) if not data: # 触发实时计算 data calculate_realtime_rec(user_id) cache.set(cache_key, data, timeout3600) return Response(data)5. 性能优化的七种武器Redis多级缓存# 第一层本地内存缓存 cache_page(60 * 15) method_decorator(cache_control(privateTrue), namedispatch) class VideoListView(ListView): pass # 第二层Redis缓存 CACHES { default: { BACKEND: django_redis.cache.RedisCache, LOCATION: redis://:passwordredis-host:6379/1, OPTIONS: { CLIENT_CLASS: django_redis.client.DefaultClient, COMPRESSOR: django_redis.compressors.lzma.LzmaCompressor, } } }Celery任务拆分shared_task(bindTrue, rate_limit100/m) def process_batch(self, batch_ids): try: data fetch_from_hadoop(batch_ids) store_to_mysql(data) except Exception as e: self.retry(exce, countdown60)6. 监控体系的建设用PrometheusGrafana搭建的监控看板需要关注这些指标指标名称报警阈值采集方式Spark任务失败率5% (15分钟)YARN APIDjango请求延迟(P99)800msPrometheus客户端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专注精确而工程师要做的是在两者之间找到最佳平衡点。