★ FEATURED ARTICLE 我最早接到这个题目时第一反应是“客流预测到底能做到什么程度”。毕竟地铁客流受天气、节假日、通勤潮汐的影响太大了而我又不想一上来就上深度学习。最终落地的这套方案就是用 Flask 做可视化服务、Hive 存历史数据、Spark 做清洗和特征工程、线性回归做预测基线。整个链路跑通后发现线性回归虽然朴素但胜在可解释、可部署、可迭代非常适合作为客流预测的第一版模型。这套系统最终完成的事情很简单统计某城市地铁全网和各站点的客流趋势预测未来几天同一时段车站的进出站量并把结果通过 Web 页面展示出来。无论你是剛入門大数据的学生还是一线数据工程师这篇文章里关于数仓建模、Spark 特征工程、模型调优和 Flask 联调的经验都可以直接抄走。1. 需求拆解与架构选型为什么是 HiveSparkFlask 这套组合1.1 客流分析业务上的几个刚需点在动手写代码之前我先把需求拆成了三层。第一层是“看历史”运营方需要知道昨天、上周、去年同期全网和各个站点的客流分别是多少峰值出现在几点哪个站涨幅最明显。第二层是“找规律”工作日早高峰和晚高峰的潮汐特性如何周末客流集中在哪些商圈站、景点站天气突变对客流的影响有多大。第三层是“做预测”基于历史规律预判明天某个时段某站会有多少人进站、出站方便调度、安保和商业部门提前准备。这三个需求对应的数据类型差异很大。历史统计是纯粹的聚合查询适合放在数仓里规律分析需要多维度对比考验的是数据模型的灵活度预测则要落到特征工程和模型训练。如果只用关系型数据库前两个需求勉强能做第三个需求的特征拼接会非常痛苦。如果直接上流式计算引擎面对十几亿条历史刷卡记录又杀鸡用牛刀。所以这时候Hive 负责存储和离线数仓Spark 负责清洗和建模Flask 负责对外输出恰好各司其职。1.2 技术选型的取舍逻辑选 Hive 而不是直接把 CSV 丢给 Pandas核心原因是数据量。以某城市地铁为例一天单是 AFC 闸机刷卡记录就有几千万条一年累计几十亿条。这种体量下 Pandas 连加载都费劲更不要提做跨年对比。Hive 的好处是把复杂的 MapReduce 作业转换成 SQL做海量数据的过滤、聚合、分区裁剪非常稳定而且我们后续的特征宽表也依赖数仓的持久化能力。选 Spark 而不是纯 Hive SQL 做特征工程则是因为线性回归需要生成大量时序特征比如“前一天同一时段客流”“上周同一天同一时段客流”“最近七天滚动均值”。这类特征涉及多表关联、窗口函数、时间偏移用 Hive SQL 写出来又臭又长调试一次要等很久。Spark 的 DataFrame API 写这类逻辑非常顺手还能顺便完成数据质量校验和异常值过滤。选 Flask 的原因更简单它足够轻。这个项目的可视化端只需要提供十几个 JSON 接口没有用户体系没有复杂权限。Flask 的路由设计直观和 ECharts 这类前端库配合也简单。如果换 Django 或者 Spring Boot反而要把大量的精力花在框架本身的配置上。1.3 整体数据流转设计整个系统的数据流转我是这样设计的原始刷卡记录通过离线同步任务进入 Hive 的原始层Spark 定期从原始层读取数据清洗后进行站点、时段聚合生成客流事实表随后 Spark 再基于事实表计算特征宽表训练好的线性回归模型每晚会把预测结果写回结果表结果表和聚合表再同步到 MySQL最后 Flask 从 MySQL 读数据暴露成 API 给前端图表使用。这里有个关键决定Flask 不直接查 Hive。原因后面会详细说但在架构上这一层隔离非常重要。下面是简化后的链路示意AFC原始记录 - Hive原始层 - Spark清洗聚合 - Hive聚合层 | Spark特征工程 线性回归预测 | 结果同步到 MySQL | Flask JSON API - ECharts 可视化这套架构没有用消息队列也没有用实时流因为需求是“看历史预测明天”离线链路完全能满足。如果有人跟你说地铁客流必须上 Flink 实时计算你就要先问清楚业务到底要的是秒级报警还是日级预测别被技术堆叠绑架。2. 数仓层落地Hive 表结构与分区策略的关键细节2.1 原始刷卡数据表怎么建数仓层我分了三个层次原始层、聚合层、特征层。原始层最简单的做法是一张宽表每条记录对应一次闸机进出站动作。由于不同城市的 ACC 系统字段命名略有差异我先以最常见的字段为例卡号、交易时间、线路 ID、站点 ID、进出站标志、闸机编号、票卡类型。建表语句如下CREATE TABLE ods_subway_afc ( card_id STRING COMMENT 脱敏后卡号, trans_time STRING COMMENT 交易时间 yyyy-MM-dd HH:mm:ss, line_id STRING COMMENT 线路编号, station_id STRING COMMENT 站点编号, in_out_flag TINYINT COMMENT 1进站 2出站, gate_no INT COMMENT 闸机编号, card_type STRING COMMENT 票卡类型 ) PARTITIONED BY (dt STRING COMMENT 交易日期分区) STORED AS PARQUET TBLPROPERTIES (parquet.compressionsnappy);这张表有两点值得注意。第一卡号必须脱敏这个不用多说涉及个人隐私的字段在进数仓之前就要做哈希处理后面所有分析都不再用原始卡号。第二分区字段用dt而不是直接用trans_time里的日期截断因为这样分区裁剪效率最高每天跑增量任务时只需要加载当天的分区。存储格式选 Parquet 的考量是列式存储对聚合类查询友好而且压缩率高。同样一份数据存成文本大概 2TBParquet Snappy 压缩之后通常只有 400GB 左右对后续 Spark 读取的 I/O 压力小很多。没选 ORC 是因为我们的下游主要是 SparkParquet 在 Spark SQL 里的兼容性更顺滑。2.2 清洗层的分区与聚合表设计原始表不会直接拿去做分析因为里面有大量脏数据重复刷卡、超时记录、测试闸机产生的虚拟数据、末班车后的进站无匹配出站记录等。清洗动作在 Spark 里做清洗后的数据落到聚合层。我建了两张核心表一张是站点小时级客流事实表一张是站点进出站汇总表。站点小时级客流事实表的设计要照顾到后续特征工程。最细粒度是“站点 线路 小时 进出站方向”字段包含客流量、小时编号、是否工作日、日期类型。小时编号用hour_id表示 0 到 23这样在按小时做对齐特征时非常方便。建表语句如下CREATE TABLE dws_subway_station_hourly ( station_id STRING, line_id STRING, hour_id INT, in_cnt BIGINT COMMENT 进站量, out_cnt BIGINT COMMENT 出站量, total_cnt BIGINT COMMENT 进出站总量, is_workday TINYINT COMMENT 是否工作日, date_type STRING COMMENT 工作日/周末/节假日 ) PARTITIONED BY (dt STRING) STORED AS PARQUET;为什么最细粒度只到小时因为线性回归的预测目标是“下一小时某站进出站量”这个粒度已经满足运营的大多数场景。如果未来要做 15 分钟级别的精细预测可以再单独建一张表但在项目初期没必要把所有粒度都做进一张表里。2.3 Hive 预聚合表的设计思路上面的小时级事实表虽然已经聚合过了但如果前端要展示“全网一天客流曲线”每次查询都要扫描所有站点再求和效率依然不高。所以我还建了一张预聚合表dws_subway_network_daily粒度为全网天级小时级字段只有日期、小时、总进站量、总出站量。这张表可以支撑首页大屏的主曲线数据量很小MySQL 都装得下。做这张表的核心思路是把高频查询提前算好别让 Web 层做聚合运算。可视化页面最怕的就是接口响应慢。预聚合表就是把计算前移的典型手段宁可多花一点数仓存储也要保证前端查询毫秒级返回。3. Spark 任务从清洗到特征工程的完整动作3.1 数据清洗里的几个隐蔽坑很多教程写数据清洗无非是去重、填充空值、过滤异常。但真实客流数据里最隐蔽的问题是“时间字段里藏着时区偏移”和“站点换乘虚拟记录”。时区问题某些闸机上报的时间是 UTC 时间同步到 Hive 后如果不加 8 小时你会发现每天的高峰期整体偏移到下午。这个坑我排查了整整半天最后把几条原始记录和线下刷卡时间做对比才发现。所以清洗的第一步就是把所有时间字段统一转成东八区并且写进注释里。换乘虚拟记录地铁换乘站在 AFC 系统里会产生“只进不出”或“只出不进”的特殊记录用于内部清分不算真实客流。如果直接 group by 站点去统计换乘大站的进出站量会明显虚高。处理方式是先过滤掉标志位异常的记录再根据线路和站点关系表排除换乘虚拟闸机。我习惯把清洗逻辑拆成几个独立的方法clean_duplicate,filter_test_record,normalize_time,exclude_transfer, 每个方法返回一个 DataFrame最后按顺序链式调用。这样做的好处是每步都可以单独验证出了问题能快速定位。3.2 时序特征如何构造特征工程是这套系统里最花时间的部分。线性回归的特征可以归为四类时间属性、历史客流、环境外因、站点属性。时间属性包括小时、是否工作日、星期几、是否节假日、是否高峰期。这里有个细节节假日不能只填一个“是否节假日”最好还要有一个“节前倒数第几天”的字段因为节前一天和节日当天、节后一天的客流模式完全不同。这个字段我用日历表去关联比在代码里写死日期要灵活。历史客流特征是最核心的预测因子包括前一天同一小时客流量、上周同一天同一小时客流量、最近七天同一小时平均客流量、当天截至当前小时的累计客流量。这些特征全部用窗口函数计算。以“前一天同一小时”为例在 Spark 里的写法是from pyspark.sql import functions as F from pyspark.sql.window import Window w Window.partitionBy(station_id, hour_id).orderBy(dt) feature_df df.withColumn( prev_day_same_hour, F.lag(total_cnt, 1).over(w) ).withColumn( prev_week_same_hour, F.lag(total_cnt, 7).over(w) )这里最容易被忽视的是partitionBy的顺序。我一开始只按station_id分区结果同一站不同小时的客流被交叉取到模型完全没法收敛。正确做法必须是station_id hour_id一起分区这样每一行取到的才是真正同一小时的历史值。3.3 特征宽表的生成方式所有特征算完之后需要拼成一张宽表每一行是一个“站点在某日期某小时”的样本列是特征最后一列是total_cnt作为标签。宽表落在 Hive 的dws_subway_features表里同时挂一个时间分区。这里有一个经验不要把所有站点的历史数据一次性全部加载到内存里做特征。按日期分区加载比如只处理最近 180 天的数据既控制了训练集大小也不会因为数据太旧导致特征分布偏移。我实际跑下来180 天的数据做小时级预测样本量足够模型表现稳定。4. 线性回归在客流预测里的真实定位4.1 为什么先选线性回归项目上线时间紧张业务方要的是一个“能解释、能改、能快速出数”的版本而不是一个黑盒模型。线性回归的优势正好体现在这里每个特征的系数可以直接打印出来给业务方解释“前一天同一时段客流每增加 1 万人今天此时段客流平均增加 0.7 万人”这种解释力是树模型和神经网络给不了的。另外客流预测本质上是一个强周期性的回归问题。周一到周五的早高峰规律非常稳定特征和标签之间近似线性关系。我对样本做了一次简单的相关性分析发现“前 7 天同一小时均值”和当前客流的 Pearson 相关系数高达 0.85 以上。这说明线性模型在这个场景下有足够的上限空间。我用的是 Spark MLlib 里的LinearRegression训练代码大致如下from pyspark.ml.regression import LinearRegression from pyspark.ml.feature import VectorAssembler, StandardScaler feature_cols [hour_id, is_workday, prev_day_same_hour, prev_week_same_hour, seven_day_avg_same_hour, is_peak, is_holiday, station_avg_flow] assembler VectorAssembler(inputColsfeature_cols, outputColfeatures_vec) scaler StandardScaler(inputColfeatures_vec, outputColscaled_features, withStdTrue, withMeanTrue) lr LinearRegression(featuresColscaled_features, labelColtotal_cnt, maxIter100, regParam0.01, elasticNetParam0.5) pipeline Pipeline(stages[assembler, scaler, lr])标准的特征缩放这个问题要特别说明。小时编号、是否工作日这种特征和客流绝对值的尺度差了好几个数量级如果不缩放正则项惩罚会失衡。StandardScaler会把所有特征统一到均值为 0、方差为 1 的分布上训练过程才会稳定。注意这里用的是 Spark MLlib 的 Pipeline API在 PySpark 3.x 版本下运行正常。如果你用的是 Spark 2.xStandardScaler的 withMean 参数在大规模稀疏数据上会报错需要检查版本兼容性。4.2 特征筛选与多重共线性处理训练之前我先看了一遍相关系数矩阵发现“prev_day_same_hour”和“seven_day_avg_same_hour”的相关系数高达 0.9。这两个特征高度共线同时放进模型会导致系数估计不稳定。解决方式是保留prev_day_same_hour把seven_day_avg_same_hour改成残差化的“七天均值偏离度”也就是(当天前七天均值 - 前一天值) / 前一天值这样既保留了周期性信息又避免和滞后特征打架。多重共线性在线性回归里最直观的表现是训练集 R² 很高但测试集一预测就崩。如果你发现某个特征的系数正负号不符合业务直觉比如“是否工作日”的系数居然是负的那大概率就是共线性或者特征泄漏的问题。排查方式很简单每次只加一个特征看系数方向是否稳定。4.3 模型评估指标怎么定客流预测的评估指标我用了三个RMSE、MAE、MAPE。RMSE 对大的预测误差特别敏感适合评估晚高峰这种绝对量大的时段MAE 更直观单位是人次MAPE 则用来衡量不同站点的预测精度差异因为大站和小站的客流基数差太多只看绝对误差没法评价小站预测得好不好。我按站点分组做了一次评估结果是大站点的 RMSE 高但 MAPE 反而低普遍在 10% 到 15%小站点的 RMSE 低但 MAPE 容易超过 30%。后来我把站点按日均客流量分桶分别训练了三个模型效果比全网一个模型好了不少。这个结论很重要客流预测不要试图用一个模型覆盖所有站点流量规模不同的站点规律差别很大。4.4 预测结果落库模型每天凌晨跑一次预测未来 48 小时每个站点的小时级客流结果写回 Hive 的dws_subway_predict表同时同步到 MySQL。这里有一个和训练时一致的原则预测时使用的滞后特征必须来自真实历史值不能拿昨天的预测值来预测今天否则误差会一路累积放大。写完预测后我还会额外算一个“预测置信区间”用训练残差的标准差乘一个系数得到区间上下限。这个区间对业务的参考价值很大调度人员不会只看一个单点预测值他们更想知道客流会不会突破某个阈值。5. Flask 服务层接口设计、实时查询与缓存策略5.1 为什么不让 Flask 直接查 Hive在最开始的原型阶段我以为可以直接让 Flask 连 Hive一个/api/trend接口即查即返回。结果第一次测试就让人崩溃一个查询两个站、30 天、24 小时的聚合请求Hive 跑了几十秒才出结果而且频繁查询还会把 YARN 队列的资源占满连早上跑数仓任务的正常作业都受影响。后来我把架构改成“Hive 计算 结果同步 MySQL Flask 查询 MySQL”。MySQL 里只存两张表一张是站点小时级聚合结果一张是预测结果。因为预聚合过数据量不大查询都是毫秒级。这个改动是整个项目性能提升最大的一步比调 Spark 参数还立竿见影。5.2 核心 API 定义可视化端我定义了五个核心接口全部走 JSONGET /api/overview 全网当日总客流、进站量、出站量、同比环比 GET /api/trend 某个站点/多个站点在某个时间段的客流曲线 GET /api/station_heat 当日各站点客流热力数据 GET /api/predict 某个站点未来 48 小时预测曲线 GET /api/model_metric 模型 RMSE、MAE、MAPE 指标以/api/trend为例Flask 的实现大概是from flask import Flask, jsonify, request import pymysql app Flask(__name__) def get_db(): return pymysql.connect(host10.0.0.7, userviz_user, password***, databasesubway_viz, charsetutf8mb4) app.route(/api/trend, methods[GET]) def trend(): station_id request.args.get(station_id, ) start_date request.args.get(start_date, ) end_date request.args.get(end_date, ) sql SELECT dt, hour_id, in_cnt, out_cnt, total_cnt FROM station_hourly_result WHERE station_id %s AND dt BETWEEN %s AND %s ORDER BY dt, hour_id with get_db() as conn: with conn.cursor() as cursor: cursor.execute(sql, (station_id, start_date, end_date)) rows cursor.fetchall() result [ {date: r[0], hour: r[1], in: r[2], out: r[3], total: r[4]} for r in rows ] return jsonify({code: 0, data: result})注意station_id和日期条件必须用参数化查询不能拼接字符串原因不用多说了防注入是基本功。另外凡是接口的入参都要加默认值和基础校验否则前端少传一个参数就会 500。5.3 缓存策略与性能兜底即使从 MySQL 查询高频接口仍然需要缓存。我的做法是给 Flask 加了简单的 TTL 缓存对于/api/overview这种首页接口缓存 5 分钟对于/api/trend这种按条件查询的接口用station_id start_date end_date拼成 key缓存 1 分钟。实现方式没有用 Redis直接用进程内字典加过期时间因为部署是单实例没有必要引入额外的中间件。如果后期要水平扩展再换 Redis 也不迟。cache {} def cached(key, ttl_seconds): def decorator(func): def wrapper(*args, **kwargs): now time.time() if key in cache and now - cache[key][time] ttl_seconds: return cache[key][data] data func(*args, **kwargs) cache[key] {data: data, time: now} return data return wrapper return decorator做到这一步可视化页面的接口响应时间基本稳定在 200ms 以内。如果某个接口还是慢不要先怀疑框架去查 MySQL 索引和查询计划90% 的慢查询都是索引缺失导致的。6. 可视化端实现细节与前后端联动6.1 页面布局与图表选型可视化端我用的是 ECharts原因是对大数据量折线图和地图支持比较成熟社区资源多。页面分成四块顶部总览卡片、中部全网客流趋势折线图、左下站点热力图、右下预测对比图。首页的总览卡片放最核心的四个数字今日总客流、进站量、出站量、预测明日客流。这些数字一个接口搞定。趋势折线图展示近 30 天全网每日客流再加一条去年同期曲线作为对比。热力图用地图组件展示站点客流分布颜色越深代表客流越大。预测对比图则把真实值和预测值画在同一条时间轴上方便业务方直观评估模型效果。6.2 前后端数据格式约定前后端数据格式的约定一定要前置。我走过弯路先让后端返回嵌套对象前端拿到手发现 ECharts 期望的是数组又回头改后端来回折腾。现在我的约定是所有时序接口统一返回扁平结构每条记录包含日期、小时、数值字段前端拿到后自行组装。比如预测接口返回格式如下前端直接映射到 series 数据即可{ code: 0, data: [ {date: 2025-05-20, hour: 8, actual: 12500, predict: 11800}, {date: 2025-05-20, hour: 9, actual: 15200, predict: 14700} ] }6.3 大屏模式注意点这个项目最后还接了一个大屏展示场景。大屏和 Web 页面最大的区别是分辨率固定、长时间不刷新、数据要自动更新。我在实现时把大屏页面设计成 1920x1080 定宽内部用百分比布局所有图表在窗口变化时调用resize方法。自动更新用了一个定时器每 5 分钟请求一次/api/overview和/api/trend而不是整体刷新页面避免屏幕闪烁。大屏上线前最好连续开机跑一天观察内存是否有泄漏。ECharts 在频繁 setOption 时如果不做清理内存会一路涨上去。我的做法是每次更新前调用chart.clear()并在页面卸载时dispose()掉所有图表实例。7. 部署上线与 Hadoop 环境下的调优经验7.1 环境依赖与版本踩坑这套系统涉及的组件很多版本不一致是最折磨人的。我的开发环境用的是 Spark 3.2 Hive 2.3 Python 3.8编译和运行时遇到过两个问题。一个是 PySpark 默认会加载自带的 Scala 版本如果集群上的 Scala 版本不同提交任务会报java.lang.NoSuchMethodError。另一个是 Hive 的 metastore 连接依赖mysql-connector-java的 jar 包如果打包时漏掉Spark 写 Hive 表会一直卡在连接 metastore 上。建议是所有组件版本尽量固定在项目根目录放一个requirements.txt和env.sh把版本号、JAVA_HOME、SPARK_HOME 全写清楚。不要凭记忆部署写文档的过程本身就是排雷的过程。7.2 Spark 任务的资源参数调优客流清洗和特征工程的任务涉及大量 Shuffle。我最开始直接用默认配置跑 30 天数据都要 40 分钟。后来逐步调整参数核心是三个spark.sql.shuffle.partitions、spark.executor.memory、spark.executor.cores。我的经验是 shuffle 分区数不能拍脑袋定 200而要按输入数据量来估。比如小时级聚合表有 500 个分区文件shuffle 分区数设为 500 到 1000每个分区处理的数据量控制在 100MB 到 200MB。执行器内存则要看单个任务的最大输入我习惯把spark.executor.memory设为 8Gspark.executor.cores设为 4每个执行器跑 2 个任务留足内存给 Shuffle 缓冲和 Java 堆外空间。还有一个高效的优化在最终聚合前做一遍coalesce把分区数降到目标文件数。比如写 Hive 聚合表时希望每张表有 100 个左右的文件就直接repartition(100)再写否则会产生大量小文件拖慢后续读取。7.3 Flask 服务的部署方式Flask 自带的 dev server 只能用于本地调试线上部署我用 gunicorn配合多 worker 和 gevent 协程模式。原因很简单可视化接口绝大多数是 I/O 等待MySQL 查询和缓存读取都很快真正耗时的场景是多个用户同时请求热力图接口这时协程能有效提高吞吐。部署命令大致是gunicorn -w 4 -k gevent -b 0.0.0.0:8080 app:app4 个 worker 对单机 16G 内存来说足够。如果并发量再大优先考虑在 Nginx 层做缓存而不是继续加 worker。可视化接口的数据本身就是分钟级一致的Nginx 缓存 30 秒完全不影响体验。8. 模型失效监测与线上评估机制8.1 预测偏差怎么及时发现模型上线后最大的风险不是预测不准而是明明不准了却没人发现。我的做法是每天用真实数据对比预测结果计算当天的 MAPE如果连续三天超过预警阈值就触发告警邮件通知到项目成员。这个监控脚本用 Spark 跑每天凌晨对比一次产出报表。阈值怎么定全网客流的 MAPE 通常能控制在 8% 到 12%站点级可能要放宽到 20% 到 30%。我分别设置了全网和站点两个维度的阈值全网告警线 15%站点告警线 35%避免误报。8.2 节假日和突发事件的特殊处理线性回归在常规日子的预测表现很好但遇到节假日、重大活动或者天气突变时误差会明显放大。比如地铁延时运营、临时封站、演唱会散场这些事件几乎没有历史规律可循纯时序模型无能为力。针对节假日我在特征里加入了“节前一天”“节日当天”“节后一天”的标志位虽然样本量少但至少能让模型学会节假日的流量整体偏移。针对突发封站唯一能做的就是人工干预在预测接口里提供一个“叠加调整参数”字段运营人员可以根据经验手动上调或者下调某个站的预测值。这个功能虽然简单但业务方非常买单因为它给了人参与调整的空间而不是完全依赖一个黑盒。8.3 模型迭代节奏线性回归模型不是训练一次就完事的。我的迭代节奏是每周重新训练一次训练窗口滚动为最近 180 天。同时每周检查一次特征重要性排序看看哪些特征的系数发生了明显变化。比如冬天天黑得早晚高峰提前小时特征的系数会相应变化如果模型不更新预测就会滞后。重新训练由一条 shell 任务驱动调用spark-submit提交训练脚本跑完自动把新模型指标和旧模型对比只有当新模型在验证集上的 MAPE 更低时才替换线上模型。这个自动对比机制避免了“辛辛苦苦重新训练结果模型越换越差”的尴尬。我个人的体会是这个项目最耗时的环节并不是模型调参而是特征工程和数据清洗的反复打磨。线性回归本身很简单但想让它的预测稳定你必须在数仓设计和特征构造上花足功夫。如果你也准备做类似的项目我的建议是先定义清楚“预测误差能接受多少”把评估机制搭好再开始优化模型。否则你会发现每个版本的模型看起来指标都行一上线就被业务方追问“为什么昨天预测差那么多”。先把数据链路打通把监控做好这个系统就已经成功了一半。
阅读完成 · 觉得有帮助？
