简介一套基于Hadoop的股票大数据分析系统采用Flask-Hive技术方案围绕股票数据的处理、分析与可视化展示源码与文档配套完整。适合计算机相关专业学生用于毕业设计、课程设计或项目初期演示也可作为大数据初学者从零理解Hadoop生态的实践样本。代码均通过运行测试作者注明答辩平均分达96分整体可靠性较高。资源包共57个文件主体为27个Python源文件承担后端接口、数据模型与控制逻辑9个JavaScript、4个HTML和3个CSS构成前端展示层另有配置、日志、说明文档等辅助文件压缩包仅433KB。工程采用蓝图模块化设计按数据模型、控制逻辑与前端展示分层组织包含用户认证、数据分页、缓存清理、测试脚本等模块便于二次开发与局部替换。已有456人学习下载获取后可得到可直接运行的完整项目、文档说明及目录结构既能支撑毕设答辩演示也能帮助理解如何将Flask与Hadoop/Hive结合完成实际数据分析任务适合在此基础上继续扩展功能。1. 股票日线数据上亿行时先选 Hadoop 这套技术栈的理由做股票分析系统最常见的翻车点不是算法是数据量。沪深两市五千多只股票每只走十年的日线一千万行以上单机 pandas 跑全市场均线扫描一次任务就是分钟级。这套基于 Hadoop 的股票大数据分析系统把行情 CSV 批量丢进 HDFS分析统计交给 Hive SQL 引擎再通过 Flask 把结果暴露给前端图表走的是典型离线批处理闭环。代码量不大HDFS 存储、Hive 数仓建模、MapReduce 调度、Web 接口一样没少适合做毕设底子也适合想看真实项目文件布局的工程师。它不是实时行情系统所有指标都是盘后批量算完再查这一点先要认清。2. Flask-Hive 分层架构从行情文件到浏览器图表的完整链路拿到这套源代码第一件事不是启动 Flask而是先读目录。它把页面、路由、业务控制和数据访问拆成了四个层次作者明显是按照企业里“控制层与数据层分离”的习惯组织的。这样做的好处很直接Hive 分析语句改动不影响接口定义新增一个股票指标只需要改 control 和 sql_tpl 两个地方不需要重写视图函数。项目里templates管页面web包下三个蓝图管路由control管业务编排dbmodel管 SQL 和连接common管鉴权与统一响应依赖方向清晰可查。2.1 为什么是 Flask Hive而不是单机 pandas MySQL如果数据量停留在一两只股票、几千行日线的量级单机 pandas 反而更顺手引入 Hadoop 确实是过度设计。但这套系统的目标计算场景是全市场扫描算沪深全部股票的 20 日均线pandas 要把每只股票的 close 一次性载入内存再 groupby内存占用随股票数线性增长算完还要拼接。Hive 的做法不同它把任务拆成分布在各数据块上的 MapReduce/Tez 作业计算跟着数据走应用进程只拿到聚合结果。数据量越大这种“移动计算而不是移动数据”的思路收益越明显。MySQL 在这套架构里的位置是结果存储不是分析引擎。Hive 全量重算出结果后先清掉stock_indicator表里对应股票的记录再批量 INSERTFlask 的接口层只查 MySQL。这样交互查询落在毫秒级Hive 只在盘后跑批时承担重量计算。对毕设来说这个设计还能把“数据仓库层”和“应用层”分开讲答辩时更容易说清楚技术分工。2.2 项目骨架里的关键文件先把职责认清楚文件职责设计要点bootstrap.py应用入口注册 page/user/data 三个蓝图加载全局配置control/datacontrol.py行情与指标业务编排决定先查缓存还是触发 Hive 批计算dbmodel/sql_tpl.py集中管理 SQL所有 Hive/MySQL 模板集中避免 SQL 散落在视图common/response.py统一返回体{code, msg, data}结构前端解析逻辑唯一这四个文件串起来的依赖方向是单向的web 路由只认识 control 层control 层只调用 dbmodeldbmodel 不直接接触前端。你在新增接口时照data_blueprint.py里已有路由复制一个再在datacontrol.py添加同名方法基本不会破坏现有功能。错误处理也统一走common/response.py比如 Hive 查询超时调用方拿到的是相同的错误结构前端弹窗逻辑不用为每个接口单独写。2.3 Hive 外部表建模与 MySQL 结果表设计行情数据要先变成 Hive 能查的表这里用外部表而不是内部表CREATE EXTERNAL TABLE IF NOT EXISTS stock_daily ( ts_code STRING COMMENT 股票代码, trade_date STRING COMMENT 交易日期 YYYYMMDD, open DOUBLE COMMENT 开盘价, high DOUBLE COMMENT 最高价, low DOUBLE COMMENT 最低价, close DOUBLE COMMENT 收盘价, volume BIGINT COMMENT 成交量(股) ) PARTITIONED BY (market STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY , STORED AS TEXTFILE LOCATION /data/stock;选外部表是因为可以反复 DROP 表而不删 HDFS 文件行情 CSV 有变动时重建表结构成本很低。分区选market而不是trade_date原因很实际查询通常是先给股票代码再给时间范围按市场分区可以把沪深股票天然隔开如果按日分区每天一批 CSV 会产生大量小文件NameNode 元数据压力大分析任务启动也慢。字段类型上close用 DOUBLE 而不是 DECIMALHive 里 DECIMAL 运算会多一层类型转换股票价格精度对毕设场景够用。结果表放在 MySQLFlask 接口直接读它CREATE TABLE IF NOT EXISTS stock_indicator ( id INT AUTO_INCREMENT PRIMARY KEY, ts_code VARCHAR(16) NOT NULL, trade_date VARCHAR(16) NOT NULL, ma5 DOUBLE, ma20 DOUBLE, dif DOUBLE, dea DOUBLE, macd DOUBLE, KEY idx_code_date (ts_code, trade_date) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;索引建在(ts_code, trade_date)上正好匹配接口层最常见的查询模式给一个股票代码拉最近 N 个交易日的指标。写入时用INSERT ... ON DUPLICATE KEY UPDATE或者当天重算前先按 ts_code 删除旧记录再批量插入避免重复数据把前端折线图拉出毛刺。3. Hadoop 伪分布式、Hive 与 Zookeeper 整合的安装与配置落地如果只想把系统跑通Hadoop 装成伪分布式就够但如果后面要接两个以上数据节点最好提前把 Hive 与 Zookeeper 的服务发现机制配好省得到时候每台机器上的连接串都要改。下面按实际搭过的顺序讲版本选择、核心配置、HiveServer2 与 ZK 整合、行情数据入库。3.1 版本组合与安装顺序常见稳定组合是 JDK 8 Hadoop 3.x Hive 3.x 或 Hive 2.3。窗口函数在 Hive 2.1 之后才完整如果要在 SQL 里写ROWS BETWEEN版本别低于 2.1。安装顺序固定为 JDK → Hadoop → Zookeeper → MySQL → HiveMetastore 要单独起服务不能同时让 Flask 和 beeline 去连 Hive 的内嵌 Derby 库否则会频繁报数据库锁或连接失败。伪分布式下副本数一定要改这是初学者最容易忽略的配置项。3.2 伪分布式核心配置core-site 与 hdfs-site!-- core-site.xml -- property namefs.defaultFS/name valuehdfs://localhost:9000/value /property !-- hdfs-site.xml -- property namedfs.replication/name value1/value /property property namedfs.namenode.name.dir/name valuefile:///data/hadoop/name/value /property property namedfs.datanode.data.dir/name valuefile:///data/hadoop/data/value /propertyfs.defaultFS是所有 HDFS 访问的统一入口Flask 所在节点也靠它找到 NameNode。伪分布式只有一个节点dfs.replication必须设为 1保持默认 3 的话集群会一直处于 under-replicated 状态后台日志不断刷块复制警告还会拖慢 put 文件的速度。元数据目录不要放在/tmp下Linux 重启会清空导致 NameNode 起不来或数据块丢失。参数推荐值说明fs.defaultFShdfs://localhost:9000HDFS 入口地址客户端访问统一走这里dfs.replication1伪分布式单节点必须为 1dfs.namenode.name.dirfile:///data/hadoop/name元数据持久化目录避免用 /tmphadoop.tmp.dir/data/hadoop/tmpHadoop 基础临时目录同样避开系统 /tmp3.3 Hive 与 Zookeeper 整合以及 HiveServer2 的服务注册Hadoop 本身在 HA 模式下依赖 Zookeeper 做 JournalNode 选主在伪分布式下Zookeeper 更常见的整合点是 HiveServer2 的动态发现。配置后beeline 客户端不再需要写明 HiveServer2 所在 IP 和端口而是从 ZK 拿地址服务重启或迁移节点时客户端无感。hive-site.xml 里加下面几项property namehive.server2.support.dynamic.service.discovery/name valuetrue/value /property property namehive.server2.zookeeper.namespace/name valuehiveserver2/value /property property namehive.zookeeper.quorum/name valuelocalhost:2181/value /property开启了动态发现以后hive.server2.thrift.bind.host要配成节点实际 IP不能是 0.0.0.0否则注册到 ZK 的地址外部客户端连不上。项目里 Python 用 pyhive 连接时连接串要写成jdbc:hive2://localhost:2181/;serviceDiscoveryModezooKeeper;zooKeeperNamespacehiveserver2端口从 10000 变成 2181第一次对接的人容易在这里卡住。Metastore 也建议独立配置到 MySQL避免 Derby 单连接限制property namejavax.jdo.option.ConnectionURL/name valuejdbc:mysql://localhost:3306/hive_metastore?createDatabaseIfNotExisttrue/value /property property namejavax.jdo.option.ConnectionDriverName/name valuecom.mysql.cj.jdbc.Driver/value /property这一步解决的是并发问题Flask 接口、beeline 客户端、定时批处理脚本可能同时访问元数据Derby 内嵌模式只支持单进程换成 MySQL 之后所有客户端共享一套元数据视图表的可见性才一致。3.4 启动顺序与行情数据入库配置完成后启动顺序有讲究我先启 HDFS 再启 YARN最后单独拉起 metastore 和 HiveServer2start-dfs.sh start-yarn.sh nohup hive --service metastore /data/hadoop/logs/metastore.log 21 nohup hive --service hiveserver2 /data/hadoop/logs/hiveserver2.log 21 数据导入路径也固定下来CSV 文件直接放进 HDFS 外部表对应的 LOCATION 目录然后修复分区hdfs dfs -mkdir -p /data/stock hdfs dfs -put stock_daily_2024.csv /data/stock/ hive -e MSCK REPAIR TABLE stock_daily;把 CSV 直接put到外部表的 location比在 Hive 里执行LOAD DATA更直观也避免了 LOAD 之后 HDFS 文件被移动到表目录以外的地方。MSCK REPAIR TABLE负责扫描目录结构把新增分区同步到 metastore如果你在put之后直接查库发现没数据九成是漏了这一步。4. Hive 窗口函数算均线与 MACD再封装成 Flask 查询接口分析接口是整套系统的门面。这一章先解决“哪些指标能用 Hive 一次算完哪些得回 Python”再给出均线、金叉两段可复用 SQL最后说明 Flask 蓝图里接口和 control 层怎么配合。接口实现的思路是能下推到 Hive 的统计不下推到 Python减少数据传输量必须逐行迭代的指标留在 Python 算避免写出又长又慢的 Hive UDF。4.1 先分清哪些指标能交给 Hive哪些要回 PythonHive 擅长的是“按窗口聚合”这一类指标移动平均、成交量加权均价、区间涨幅、全市场涨跌家数统计一条窗口函数就能出结果。MACD 这类依赖前一条结果做迭代的指标比较特殊EMA 的计算需要引用上一条 EMA 的值Hive 窗口函数不支持这种递归引用。常见的做法是Hive 只负责把某只股票的 close、trade_date 明细查出来Python 端用 pandas 的ewm完成 EMA/MACD 计算再批量写回 MySQL。这样分工清晰调试也方便。4.2 用 Hive SQL 算 MA5 与 MA20 金叉移动平均是最典型的下推场景SELECT ts_code, trade_date, close, AVG(close) OVER ( PARTITION BY ts_code ORDER BY trade_date ROWS BETWEEN 4 PRECEDING AND CURRENT ROW ) AS ma5, AVG(close) OVER ( PARTITION BY ts_code ORDER BY trade_date ROWS BETWEEN 19 PRECEDING AND CURRENT ROW ) AS ma20 FROM stock_daily WHERE ts_code 600519 AND trade_date 20240101 ORDER BY trade_date;PARTITION BY ts_code把不同股票的窗口隔开否则所有股票的日期会混在一起排序ORDER BY trade_date决定窗口的前进方向ROWS BETWEEN 4 PRECEDING AND CURRENT ROW表示取当前行及前 4 行求平均。这里有个小坑如果只写ORDER BY trade_date不写ROWS BETWEENHive 默认用 RANGE 窗口同一天有两条记录时会把它们同时拉进窗口导致 MA5 的样本数漂移加上 ROWS 从句后行为完全确定。金叉统计可以在此基础上用LAG拿到上一交易日的数据WITH ma_data AS ( SELECT ts_code, trade_date, AVG(close) OVER (PARTITION BY ts_code ORDER BY trade_date ROWS BETWEEN 4 PRECEDING AND CURRENT ROW) AS ma5, AVG(close) OVER (PARTITION BY ts_code ORDER BY trade_date ROWS BETWEEN 19 PRECEDING AND CURRENT ROW) AS ma20 FROM stock_daily WHERE trade_date 20240101 ) SELECT ts_code, trade_date, ma5, ma20 FROM ( SELECT ts_code, trade_date, ma5, ma20, LAG(ma5, 1) OVER (PARTITION BY ts_code ORDER BY trade_date) AS prev_ma5, LAG(ma20, 1) OVER (PARTITION BY ts_code ORDER BY trade_date) AS prev_ma20 FROM ma_data ) t WHERE prev_ma5 prev_ma20 AND ma5 ma20;内层子查询先把 MA5/MA20 算出来LAG(ma5, 1)取同一只股票上一交易日的 MA5外层过滤条件prev_ma5 prev_ma20 AND ma5 ma20表达的语义是昨天 MA5 还在 MA20 下方今天向上穿越这就是金叉。首条记录因为prev_ma5为 NULLNULL prev_ma20结果为 NULL不会被当成金叉不需要额外写 IS NOT NULL。4.3 MACD 的递归计算问题与 Python 兜底EMA 是递归公式Hive 里没有现成的窗口函数能一步算完硬写 UDF 维护状态又增加了排查成本。我一般直接让 Python 接手import pandas as pd df pd.read_sql( SELECT trade_date, close FROM stock_daily WHERE ts_code 600519 ORDER BY trade_date, hive_conn ) ema12 df[close].ewm(span12, adjustFalse).mean() ema26 df[close].ewm(span26, adjustFalse).mean() dif ema12 - ema26 dea dif.ewm(span9, adjustFalse).mean() macd (dif - dea) * 2 df[dif] dif df[dea] dea df[macd] macdewm(span12, adjustFalse)表示指数加权窗口为 12并且从第一条数据就开始计算第一条 EMA 直接取第一条 close。adjustFalse很关键默认的 adjustTrue 会对初始值做归一化算出来的 MACD 前几个交易日会明显偏离常规行情软件的结果。计算完成后按 ts_code 和 trade_date 写回 MySQL写之前先 delete 掉同股票在目标日期范围内的旧记录保证接口幂等。4.4 Flask 蓝图封装接口、控制层与 SQL 模板的配合接口层要写得足够薄下面的代码是data_blueprint.py里的典型路由from flask import Blueprint, request, jsonify from control.datacontrol import DataControl data_bp Blueprint(data, __name__) data_bp.route(/api/stock/ma, methods[GET]) def stock_ma(): code request.args.get(code, 600519) days int(request.args.get(days, 250)) result DataControl().calc_ma(code, days) return jsonify({code: 0, msg: success, data: result})视图函数不直接碰 Hive 连接也不拼 SQLDataControl().calc_ma(code, days)内部会决定先查 MySQL 缓存缓存没有命中再走 SQL 模板查 Hive 明细并计算指标。days参数在这里的作用是限制返回最近 N 个交易日不要把 Hive 全量结果一次性塞给前端。接口完成后参数含义可以列成固定表参数类型默认值说明codestr600519股票代码对应 stock_daily.ts_codedaysint250返回最近多少个交易日的指标数据startstr20240101起始日期可配合全市场回刷使用endstr空结束日期空则取最新交易日有两个隐蔽的连接问题要提前避开一是 HiveServer2 的默认传输模式是二进制Flask 和 HiveServer2 不在同一台机器时连接串要显式加transportModehttp;httpPathcliservice二是不要在每个请求里新建 Hive 连接再随手关闭常见做法是启动时建立一个连接池指标重算时排队复用连接否则高并发下 HiveServer2 的会话数会迅速打满。5. 上线前的结果校验、Hive 性能参数与指标缓存清理技巧系统能跑通和算得对是两回事最后一部分给三条可落地的经验第一随机抽股票做结果对账第二改三组 Hive 执行参数降低批计算成本第三把缓存清理逻辑独立成脚本避免运行时的数据快照不一致。5.1 随机抽股票做结果对账对账的原理是拿 Hive 原始计算结果和已经落库的 MySQL 指标做差值比较允许的误差范围可以按 0.001 控制import pandas as pd from pyhive import hive import pymysql hive_conn hive.Connection(hostlocalhost, port10000, usernameroot) mysql_conn pymysql.connect( hostlocalhost, userroot, password123456, databasestock_analysis ) hive_df pd.read_sql( SELECT ts_code, trade_date, AVG(close) OVER (PARTITION BY ts_code ORDER BY trade_date ROWS BETWEEN 4 PRECEDING AND CURRENT ROW) AS ma5 FROM stock_daily WHERE ts_code IN (600519, 000001, 000858) , hive_conn) mysql_df pd.read_sql( SELECT ts_code, trade_date, ma5 FROM stock_indicator WHERE ts_code IN (600519, 000001, 000858), mysql_conn ) merged pd.merge( hive_df, mysql_df, on[ts_code, trade_date], suffixes(_hive, _mysql) ) merged[diff] (merged[ma5_hive] - merged[ma5_mysql]).abs() print(merged[merged[diff] 0.001])抽样的股票要覆盖不同价格带和停牌情况比如高价股、低价股、长期停牌复牌股避免只验算几种正常行情。误差超过 0.001 时先检查 Hive 里 trade_date 字段是不是有脏数据再检查写入 MySQL 时的覆盖逻辑是否清干净了旧记录。5.2 三组值得改的 Hive 执行参数参数默认值建议值说明hive.fetch.task.conversionminimalmore小查询走 fetch task不提交 MR 作业显著降低冷启动hive.exec.parallelfalsetrue多个无依赖的 MapReduce 阶段并行执行hive.merge.mapfilesfalsetrue合并 Map 端输出的小文件降低后续读取开销hive.fetch.task.conversionmore对接口联调阶段帮助最大像SELECT * FROM stock_daily LIMIT 10这类语句不会真的去申请 YARN 资源响应速度从秒级降到毫秒级。全市场批量计算时再把hive.exec.parallel打开多只股票的分析任务能并行跑。合并小文件参数要配合hive.merge.size.per.task使用否则全部文件合并成一个大文件后续并发度反而会掉。5.3 缓存清理与批计算写入指标重算的常见节奏是每个交易日收盘后清空当日缓存再触发一次全量或增量计算。项目里runcleanmongocache.py这类脚本名字带 cache本质上是清理临时结果和失效连接缓存我习惯把清理逻辑独立出来保证接口请求时不需要动态删数据避免读到一半被清空的中间状态。调度可以用 cron 固定时间执行0 18 * * 1-5 cd /opt/stock_analysis python script/runcleanmongocache.py /var/log/stock_analysis/clean.log 21 20 18 * * 1-5 python /opt/stock_analysis/script/compute_daily.py /var/log/stock_analysis/compute.log 21清理脚本先跑给批计算腾出干净的结果表compute_daily.py 计算完成后再更新一个last_update标记表Flask 接口启动时检查这个标记判断库存指标是否为最新。当天如果遇到补数需求只要重放对应 CSV 并重跑 compute_daily.py指标表自动覆盖无需再手动改数据库里的任何记录。本文还有配套的精品资源点击获取