拓冰网站定制 
简介面向大数据开发初学者这套以酒店度假为业务场景的实战项目结合Spark内存计算、MySQL数据存储与Echarts可视化展示完整覆盖了从数据清洗、聚合统计到图表呈现的全链路实现能帮助读者建立大数据分析项目的整体认知也可作为课程设计、毕业设计的参考案例。压缩包共24个文件涵盖Scala业务源码、HTML/CSS/JS前端页面、字体与图片资源、数据库相关文件以及实训报告和项目总结PPT包体约9.27MB目录结构清晰便于按模块学习定位。源码中涉及酒店均价计算、城市数量统计、类型TopN等典型分析任务Scala部分可练习Spark常用算子前端通过Echarts完成图表配置并与MySQL数据联动基本还原了真实项目中存储、计算、展示的分层实现。已有96人学习既可直接运行查看效果也可在源码基础上扩展更多分析维度是大数据入门实战的不错参考。1. 酒店度假数据可视化难的是数据链路不是图表用 Spark MySQL Echarts 搭建一套酒店度假数据可视化系统很多人以为是三个组件拼一拼Spark 算数MySQL 存数Echarts 画图。真正动手就会发现数据从哪来、清洗规则怎么定、指标口径怎么统一、图表如何不出错才是这个大数据项目实战的核心。下面按实训和毕业设计最常见的方式展开单机 Spark、PySpark 编码、MySQL 存聚合结果、Flask 出接口最后用 Echarts 做经营看板。适合想要一份能完整跑通源码、数据库脚本、实训报告和 PPT 的读者也适合小团队快速验证内部数据。2. 搭建前先定版本Spark 集群、MySQL 授权、Echarts 离线包这类项目最容易翻车的不是业务逻辑而是环境版本互相不兼容。比较稳的组合是 Spark 3.3Python 3.8MySQL 8.0Echarts 5.xPySpark 和 MySQL JDBC 驱动的版本要对得上否则会出现No suitable driver或UnsupportedColumnTypeException这类让人摸不着头脑的报错。下面按“能本地复现”的思路把三方环境定下来。2.1 为什么实训项目推荐 Spark local 模式起步Spark 的运行模式有 local、standalone、yarn、kubernetes。很多人一听说大数据项目就直接搭集群结果时间全部耗在 SSH 免密和 HDFS 启动上。课程设计或企业看板这种规模一台开发机上用local[*]就够了Spark 会在本地用多线程模拟 executor行为上仍然遵守任务调度和 shuffle 逻辑。只有你要专门考察集群部署能力时才需要把节点扩展到三台去验证“spark 集群搭建”的过程。初始化 SparkSession 时下面这段是每个 Spark 应用的第一步from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(hotel_demo) \ .master(local[*]) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate()这里要求写清楚两个点master(local[*])中的*表示使用机器全部核心也可以写成local[4]固定四核方便反复调试spark.sql.shuffle.partitions默认是 200在小数据集上会产生大量空任务本地调低到 8 或 12运行结束时间能肉眼可见变短。如果以后要上集群只需要把master换成 YARN 集群入口或者通过spark-submit --master yarn提交代码本身不需要改这也是 Spark 代码值得单独拆成一个项目目录的原因。模式适用规模配置成本典型场景local[*]单机百万行内最低实训、毕设、原型验证standalone3-20 台机器中等独立 Spark 集群不依赖 Hadoopyarn与 Hadoop 共用较高企业已有 HDFS/Hive 环境2.2 MySQL 8.0 建库与用户授权避免写入卡在权限上Spark 通过 JDBC 写 MySQL 时需要一个具备建表、插入、删除权限的账号。不建议直接用 root而是新建一个最小权限用户防止调试的数据被误清掉。在 MySQL 里执行CREATE DATABASE IF NOT EXISTS hotel_dw DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER spark_app% IDENTIFIED BY spark2024; GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, INDEX, ALTER ON hotel_dw.* TO spark_app%; FLUSH PRIVILEGES;字符集必须显式指定utf8mb4否则城市名、酒店名写入后可能变成???。spark_app%表示允许任何 IP 连接因为 PySpark 通过 IP 访问 MySQL不像命令行默认走 socket写localhost会在远程连接时报Access denied。JDBC 连接串里同样要带上时区和编码jdbc:mysql://localhost:3306/hotel_dw?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8serverTimezoneAsia/Shanghai解决 MySQL 8.0 与 Spark 默认时区不一致导致的日期偏移characterEncodingutf8是防止中文乱码的最后一道保险。随后把聚合结果表的建表语句准备好主键用(city, ym)这样 Spark 多次重跑数据不会产生重复行CREATE TABLE IF NOT EXISTS city_month_agg ( city VARCHAR(20) NOT NULL, ym CHAR(7) NOT NULL, room_nights INT, revenue DECIMAL(14,2), adr DECIMAL(10,2), PRIMARY KEY (city, ym) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;2.3 Python 造一份酒店订单数据字段口径先于代码定下来没有真实业务数据时自己生成一份合理的 CSV 是第一步。行业里常见的字段包括订单号、酒店名、城市、入住日期、连住晚数、单晚房价、房间数和渠道。用 Python 生成 20000 条订单时间覆盖一整年便于后面分析淡旺季import csv import random from datetime import datetime, timedelta random.seed(42) cities [北京, 上海, 广州, 深圳, 杭州, 成都, 三亚, 厦门, 重庆, 西安] hotels {city: [f{city}酒店{i} for i in range(1, 6)] for city in cities} channels [自家小程序, 携程, 美团, 飞猪, 线下] start datetime.strptime(2024-01-01, %Y-%m-%d) rows [] for order_id in range(1, 20001): city random.choice(cities) hotel random.choice(hotels[city]) days random.randint(1, 7) checkin start timedelta(daysrandom.randint(0, 364)) price round(random.uniform(200, 1200), 2) rooms random.randint(1, 3) rows.append([order_id, hotel, city, checkin.strftime(%Y-%m-%d), days, price, rooms, random.choice(channels)]) with open(data/hotel_orders.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([order_id, hotel_name, city, checkin_date, nights, room_price, rooms, channel]) writer.writerows(rows)random.seed(42)保证每次生成的数据一致答辩时两次运行的图表不会自己变样。数据覆盖 2024 年全年价格范围 200 到 1200 元渠道同时包含私域和 OTA。这里的字段口径决定了 Spark 阶段怎么算指标revenue room_price * nights * roomsroom_nights nights * rooms后面所有图表都基于这两个衍生值。如果你想让地图出现“三亚冬季热、重庆暑期热”的规律可以按城市月份微调权重否则完全随机的结果很难看出业务意义。3. Spark 清洗与聚合订单明细变酒店经营指标的完整过程3.1 读取 CSV 后的第一道检查schema 和空值Spark 读取 CSV 时最常用的两个参数是headertrue和inferSchematrue前者把第一行当表头后者让 Spark 自动推断字段类型。但inferSchema不是万能的日期列可能被推断成字符串价格列可能因为某个脏值变成字符串。稳妥做法是读进来后先打印 schema 再决定要不要做类型转换。df spark.read \ .option(header, true) \ .option(inferSchema, true) \ .option(dateFormat, yyyy-MM-dd) \ .csv(data/hotel_orders.csv) df.printSchema() df.show(5, truncateFalse)printSchema()能看到每个字段的 Spark SQL 类型show(5)看前五行内容重点确认中文列没有乱码。数据量不大时直接让 Spark 推断没问题但需要留一手如果某天上游字段顺序变了下游聚合不会报错只会静默出错。更好的做法是在读数据时手动指定 schema不过对于 2 万行的小数据集成本收益不高。清洗逻辑按顺序做四步from pyspark.sql import functions as F df_clean df.dropDuplicates([order_id]) \ .filter(F.col(checkin_date).isNotNull()) \ .withColumn(room_price, F.round(F.col(room_price), 2)) \ .withColumn(revenue, F.col(room_price) * F.col(nights) * F.col(rooms)) \ .withColumn(ym, F.date_format(F.to_date(checkin_date, yyyy-MM-dd), yyyy-MM))这里有个很容易忽略的细节dropDuplicates([order_id])只保留每个订单的第一条记录。CSV 是我们自己生成的不会有重复但真实业务中同一订单可能因为重试写入产生多条放在这里能积累一层防御。revnue和ym是后面所有图表的基础字段ym统一为yyyy-MM格式避免前端还要处理日期字符串。3.2 用 groupBy 和窗口函数计算间夜、ADR、城市热度酒店行业最常用的指标有三个间夜数room_nights、收入revenue、平均每日房价ADR。ADR 的计算口径是revenue / room_nights而不是avg(room_price)因为一个订单可能订多间房多晚后者算出来的数字没有业务意义。这一步的差异在实训报告里经常被面试官追问。city_metric df_clean.groupBy(city, ym).agg( F.sum(nights).alias(room_nights), F.sum(rooms).alias(room_counts), F.sum(revenue).alias(revenue), F.avg(room_price).alias(avg_price) ).withColumn(adr, F.round(F.col(revenue) / F.col(room_nights), 2)) city_metric.cache() city_metric.orderBy(ym, F.desc(room_nights)).show(10, truncateFalse)cache()会把中间结果缓存到内存后面如果反复查询这份数据能省掉重复计算。聚合结果先写show()验证数据量和格式确认没问题再落库。如果还想给答辩加一个进阶点可以用窗口函数计算累计间夜数解释“每个城市 2024 年累计间夜进度”from pyspark.sql.window import Window windowSpec Window.partitionBy(city).orderBy(ym).rowsBetween(Window.unboundedPreceding, Window.currentRow) city_metric.withColumn(cum_room_nights, F.sum(room_nights).over(windowSpec)).show(10)窗口函数的rowsBetween表示从分区第一行到当前行这是累计值的标准写法。它需要在 shuffle 时把同一城市的数据拉到同一个 task所以分区数不能太少coalesce(1)写文件之前再考虑不要在最前面用。3.3 写回 MySQL 前必须调好的 JDBC 参数Spark 写 MySQL 最直接的 API 是df.write.jdbc()但很多人第一次都会卡在驱动上。PySpark 不会自带 MySQL 驱动需要先把mysql-connector-java-8.0.33.jar下载到项目的libs/目录然后通过spark-submit --jars或pyspark --jars传进去。base_url jdbc:mysql://localhost:3306/hotel_dw props { user: spark_app, password: spark2024, driver: com.mysql.cj.jdbc.Driver, truncate: true, batchsize: 2000 } city_metric.write.mode(overwrite) \ .option(createTableOptions, ENGINEInnoDB DEFAULT CHARSETutf8mb4) \ .jdbc(base_url ?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8, city_month_agg, propertiesprops)truncatetrue在overwrite模式下会先TRUNCATE TABLE再插入比逐行DELETE快得多batchsize2000控制单个批次最多写入 2000 行调大通常能减少网络往返但会占用更多驱动内存。createTableOptions会在表不存在时自动建表的 DDL 后面追加建表选项保证 InnoDB 和 utf8mb4 生效。如果表已经存在这个参数只影响 Spark 自动建表分支。JDBC 参数建议值作用truncatetrue重跑时先清空再插入避免主键冲突batchsize2000单批写入行数过大增加内存压力rewriteBatchedStatementstrueMySQL JDBC 优化批量写入需要在 URL 中追加useSSLfalse本机连接避免 SSL 握手耗时提交到集群时本地脚本里的master配置会被spark-submit的参数覆盖驱动 JAR 用--jars指定同时--driver-class-path把它加到 driver 端否则会一直报ClassNotFound。验证写入是否成功直接在 MySQL 里执行最简单SELECT city, ym, room_nights, revenue FROM city_month_agg ORDER BY ym, room_nights DESC LIMIT 10;4. MySQL 查询接口与 Echarts 图表映射让数据可视化不只在图表上好看4.1 Flask 只读接口前端不接触数据库连接串前端直接连接 MySQL 会把账号密码暴露在浏览器里企业级数据可视化项目基本不会这么干。常见做法是后端提供一个只读 JSON 接口Flask 因为轻量适合这种聚合结果已经算好、只需要查表的场景。安装依赖后写一个最小可运行的服务from flask import Flask, jsonify import pymysql import pymysql.cursors app Flask(__name__) DB { host: localhost, port: 3306, user: spark_app, password: spark2024, database: hotel_dw, charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor } app.route(/api/city_month) def city_month(): conn pymysql.connect(**DB) with conn.cursor() as cur: cur.execute(SELECT city, ym, room_nights, revenue, adr FROM city_month_agg ORDER BY ym, room_nights DESC) rows cur.fetchall() conn.close() return jsonify(rows) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)DictCursor让查询结果直接变成字典列表jsonify序列化后前端不用再手动拼字段。host0.0.0.0允许局域网访问演示时手机也能打开这种细节在答辩现场很加分。上面的 SQL 没有参数如果是可变的日期筛选务必改成参数化查询例如cur.execute(SELECT ... WHERE ym %s, (ym,))避免 SQL 注入。4.2 柱状图和折线图Echarts 的最简启动配置Echarts 5.x 的引入方式很简单在生产环境建议把echarts.min.js下载到static/目录下避免演示时现场网络加载超时。页面结构是一个div加一段scriptdiv idchart1 stylewidth: 100%; height: 480px;/div script src/static/echarts.min.js/script script async function loadChart() { const resp await fetch(/api/city_month); const rows await resp.json(); const months [...new Set(rows.map(r r.ym))].sort(); const cityMap {}; rows.forEach(r { if (!cityMap[r.city]) cityMap[r.city] []; cityMap[r.city].push({ ym: r.ym, nights: r.room_nights }); }); const chart echarts.init(document.getElementById(chart1)); chart.setOption({ tooltip: { trigger: axis }, legend: { top: 10 }, xAxis: { type: category, data: months }, yAxis: { type: value, name: 间夜数 }, series: Object.entries(cityMap).map(([city, arr]) ({ name: city, type: line, data: months.map(m arr.find(d d.ym m)?.nights ?? 0) })) }); } loadChart(); /script这个写法解决了两个常见问题第一series 是根据接口数据动态生成的而不是前端写死十个城市第二months.map(m ... ?? 0)会把没有数据的月份补成 0否则折线在 11 月突然断掉看起来像丢数据。tooltip.triggeraxis让鼠标悬浮时纵向对比所有城市legend自动显示城市名。如果你想要柱状图和折线图同时出现在一张图里把两个城市分别设置type: bar和type: line、yAxisIndex: 0/1即可。企业级数据可视化看板一般会用“柱状图展示间夜数、折线图展示 ADR”轴单位不一致时再加yAxisIndex区分。4.3 中国地图热力展示GeoJSON 与 Echarts 的坑搜索“echarts 中国地图”的人经常踩同一个坑Echarts 从 4.x 开始不再内置 GeoJSON 地图数据直接series.type: map, map: china只会得到空白或一个长方形边框。正确的做法是先注册地图再使用。把china.json放在static/目录前端这样加载fetch(/static/china.json) .then(res res.json()) .then(geo { echarts.registerMap(china, geo); chart.setOption({ series: [{ type: map, map: china, roam: true, data: heatData }] }); });registerMap的第一个参数是自定义地图名第二个参数是 GeoJSON 对象heatData的每一项必须是{ name: 三亚, value: 12345 }这种结构且name必须和 GeoJSON 里的行政区域名称一致。一个高频报错是控制台没有错误但地图区域不填充颜色原因通常是 GeoJSON 里的名字叫“海南”而前端传的是“海南省”需要先console.log(geo.features.map(f f.properties.name))检查真实名称。Echarts 系列适合指标动态数据注意事项line趋势、环比月份缺失时补零bar城市间夜数对比短的柱子可能显示 0需设置barMinHeightpie渠道占比数据传对象否则仍然显示旧图map城市地域热度name 必须匹配 GeoJSON如果图形只显示一个长方形排除法看两步一是 GeoJSON 是否成功 fetch二是map名称是否和注册名完全一致。另外roam: true允许图表缩放和拖拽用于演示时聚焦到某个热门区域这是让地图看起来有“企业级”质感的关键参数。5. 从实训报告到 PPT把源码、数据库脚本和图表打包成可复现的大数据项目5.1 一键重跑脚本数据更新与校验一起做项目交付给老师或同事时对方不会关心你怎么写代码只会尝试执行一个命令把整个流程跑通。把数据生成、Spark 聚合、MySQL 校验串成一个build.sh能省掉大量口述时间#!/bin/bash set -e echo 重新生成订单数据 python3 data/generate_data.py echo Spark 清洗聚合 spark-submit --jars libs/mysql-connector-java-8.0.33.jar \ --driver-class-path libs/mysql-connector-java-8.0.33.jar \ scripts/build_marts.py echo 校验结果表 mysql -h 127.0.0.1 -u spark_app -pspark2024 hotel_dw \ -e SELECT COUNT(*) AS row_count, ROUND(SUM(revenue),2) AS total_revenue FROM city_month_agg;set -e让脚本在任一命令失败时立即退出不会出现“Spark 失败了但 MySQL 校验还在继续”的假成功。密码写进脚本只适用于实训演示提交到 Git 前应改用环境变量或配置文件如果无法在答辩现场连接公网所有依赖 JAR 都放在libs/目录中这是保证可复现的最低条件。5.2 实训报告和 PPT 的叙事顺序实训报告不需要写“我用了哪些技术”这种流水账最能说明问题的是数据链路图加指标口径。建议按五页 PPT 递进第一页抛出问题“酒店度假数据分散在多个渠道如何呈现城市经营差异”第二页画架构图Spark 清洗→MySQL 聚合→Flask 接口→Echarts 展示第三页放附录中 SQL 建表和 Spark 代码片段第四页放置 3 张图表分别在评论区写出一个结论例如“三亚 ADR 在 11 月至次年 2 月明显高于其他城市”第五页列出验证过程包括随机抽样订单手算间夜数、与 MySQL 查询结果对比。最后一个可用的验证命令是用接口返回 JSON 做整体检查curl -s http://127.0.0.1:5000/api/city_month | python -c import json,sys; datajson.load(sys.stdin); assert all(r[adr]0 for r in data), ADR 异常; print(check passed)这条命令在启动 Flask 后执行会检查所有 ADR 是否大于 0任何一条负数都会让程序抛出异常并返回非零退出码也是后面接入定时调度和告警脚本的雏形。本文还有配套的精品资源点击获取 Spark+MySQL+Echarts构建酒店数据可视化系统实战
简介面向大数据开发初学者这套以酒店度假为业务场景的实战项目结合Spark内存计算、MySQL数据存储与Echarts可视化展示完整覆盖了从数据清洗、聚合统计到图表呈现的全链路实现能帮助读者建立大数据分析项目的整体认知也可作为课程设计、毕业设计的参考案例。压缩包共24个文件涵盖Scala业务源码、HTML/CSS/JS前端页面、字体与图片资源、数据库相关文件以及实训报告和项目总结PPT包体约9.27MB目录结构清晰便于按模块学习定位。源码中涉及酒店均价计算、城市数量统计、类型TopN等典型分析任务Scala部分可练习Spark常用算子前端通过Echarts完成图表配置并与MySQL数据联动基本还原了真实项目中存储、计算、展示的分层实现。已有96人学习既可直接运行查看效果也可在源码基础上扩展更多分析维度是大数据入门实战的不错参考。1. 酒店度假数据可视化难的是数据链路不是图表用 Spark MySQL Echarts 搭建一套酒店度假数据可视化系统很多人以为是三个组件拼一拼Spark 算数MySQL 存数Echarts 画图。真正动手就会发现数据从哪来、清洗规则怎么定、指标口径怎么统一、图表如何不出错才是这个大数据项目实战的核心。下面按实训和毕业设计最常见的方式展开单机 Spark、PySpark 编码、MySQL 存聚合结果、Flask 出接口最后用 Echarts 做经营看板。适合想要一份能完整跑通源码、数据库脚本、实训报告和 PPT 的读者也适合小团队快速验证内部数据。2. 搭建前先定版本Spark 集群、MySQL 授权、Echarts 离线包这类项目最容易翻车的不是业务逻辑而是环境版本互相不兼容。比较稳的组合是 Spark 3.3Python 3.8MySQL 8.0Echarts 5.xPySpark 和 MySQL JDBC 驱动的版本要对得上否则会出现No suitable driver或UnsupportedColumnTypeException这类让人摸不着头脑的报错。下面按“能本地复现”的思路把三方环境定下来。2.1 为什么实训项目推荐 Spark local 模式起步Spark 的运行模式有 local、standalone、yarn、kubernetes。很多人一听说大数据项目就直接搭集群结果时间全部耗在 SSH 免密和 HDFS 启动上。课程设计或企业看板这种规模一台开发机上用local[*]就够了Spark 会在本地用多线程模拟 executor行为上仍然遵守任务调度和 shuffle 逻辑。只有你要专门考察集群部署能力时才需要把节点扩展到三台去验证“spark 集群搭建”的过程。初始化 SparkSession 时下面这段是每个 Spark 应用的第一步from pyspark.sql import SparkSession spark SparkSession.builder \ .appName(hotel_demo) \ .master(local[*]) \ .config(spark.sql.shuffle.partitions, 8) \ .getOrCreate()这里要求写清楚两个点master(local[*])中的*表示使用机器全部核心也可以写成local[4]固定四核方便反复调试spark.sql.shuffle.partitions默认是 200在小数据集上会产生大量空任务本地调低到 8 或 12运行结束时间能肉眼可见变短。如果以后要上集群只需要把master换成 YARN 集群入口或者通过spark-submit --master yarn提交代码本身不需要改这也是 Spark 代码值得单独拆成一个项目目录的原因。模式适用规模配置成本典型场景local[*]单机百万行内最低实训、毕设、原型验证standalone3-20 台机器中等独立 Spark 集群不依赖 Hadoopyarn与 Hadoop 共用较高企业已有 HDFS/Hive 环境2.2 MySQL 8.0 建库与用户授权避免写入卡在权限上Spark 通过 JDBC 写 MySQL 时需要一个具备建表、插入、删除权限的账号。不建议直接用 root而是新建一个最小权限用户防止调试的数据被误清掉。在 MySQL 里执行CREATE DATABASE IF NOT EXISTS hotel_dw DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; CREATE USER spark_app% IDENTIFIED BY spark2024; GRANT SELECT, INSERT, UPDATE, DELETE, CREATE, DROP, INDEX, ALTER ON hotel_dw.* TO spark_app%; FLUSH PRIVILEGES;字符集必须显式指定utf8mb4否则城市名、酒店名写入后可能变成???。spark_app%表示允许任何 IP 连接因为 PySpark 通过 IP 访问 MySQL不像命令行默认走 socket写localhost会在远程连接时报Access denied。JDBC 连接串里同样要带上时区和编码jdbc:mysql://localhost:3306/hotel_dw?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8serverTimezoneAsia/Shanghai解决 MySQL 8.0 与 Spark 默认时区不一致导致的日期偏移characterEncodingutf8是防止中文乱码的最后一道保险。随后把聚合结果表的建表语句准备好主键用(city, ym)这样 Spark 多次重跑数据不会产生重复行CREATE TABLE IF NOT EXISTS city_month_agg ( city VARCHAR(20) NOT NULL, ym CHAR(7) NOT NULL, room_nights INT, revenue DECIMAL(14,2), adr DECIMAL(10,2), PRIMARY KEY (city, ym) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;2.3 Python 造一份酒店订单数据字段口径先于代码定下来没有真实业务数据时自己生成一份合理的 CSV 是第一步。行业里常见的字段包括订单号、酒店名、城市、入住日期、连住晚数、单晚房价、房间数和渠道。用 Python 生成 20000 条订单时间覆盖一整年便于后面分析淡旺季import csv import random from datetime import datetime, timedelta random.seed(42) cities [北京, 上海, 广州, 深圳, 杭州, 成都, 三亚, 厦门, 重庆, 西安] hotels {city: [f{city}酒店{i} for i in range(1, 6)] for city in cities} channels [自家小程序, 携程, 美团, 飞猪, 线下] start datetime.strptime(2024-01-01, %Y-%m-%d) rows [] for order_id in range(1, 20001): city random.choice(cities) hotel random.choice(hotels[city]) days random.randint(1, 7) checkin start timedelta(daysrandom.randint(0, 364)) price round(random.uniform(200, 1200), 2) rooms random.randint(1, 3) rows.append([order_id, hotel, city, checkin.strftime(%Y-%m-%d), days, price, rooms, random.choice(channels)]) with open(data/hotel_orders.csv, w, newline, encodingutf-8) as f: writer csv.writer(f) writer.writerow([order_id, hotel_name, city, checkin_date, nights, room_price, rooms, channel]) writer.writerows(rows)random.seed(42)保证每次生成的数据一致答辩时两次运行的图表不会自己变样。数据覆盖 2024 年全年价格范围 200 到 1200 元渠道同时包含私域和 OTA。这里的字段口径决定了 Spark 阶段怎么算指标revenue room_price * nights * roomsroom_nights nights * rooms后面所有图表都基于这两个衍生值。如果你想让地图出现“三亚冬季热、重庆暑期热”的规律可以按城市月份微调权重否则完全随机的结果很难看出业务意义。3. Spark 清洗与聚合订单明细变酒店经营指标的完整过程3.1 读取 CSV 后的第一道检查schema 和空值Spark 读取 CSV 时最常用的两个参数是headertrue和inferSchematrue前者把第一行当表头后者让 Spark 自动推断字段类型。但inferSchema不是万能的日期列可能被推断成字符串价格列可能因为某个脏值变成字符串。稳妥做法是读进来后先打印 schema 再决定要不要做类型转换。df spark.read \ .option(header, true) \ .option(inferSchema, true) \ .option(dateFormat, yyyy-MM-dd) \ .csv(data/hotel_orders.csv) df.printSchema() df.show(5, truncateFalse)printSchema()能看到每个字段的 Spark SQL 类型show(5)看前五行内容重点确认中文列没有乱码。数据量不大时直接让 Spark 推断没问题但需要留一手如果某天上游字段顺序变了下游聚合不会报错只会静默出错。更好的做法是在读数据时手动指定 schema不过对于 2 万行的小数据集成本收益不高。清洗逻辑按顺序做四步from pyspark.sql import functions as F df_clean df.dropDuplicates([order_id]) \ .filter(F.col(checkin_date).isNotNull()) \ .withColumn(room_price, F.round(F.col(room_price), 2)) \ .withColumn(revenue, F.col(room_price) * F.col(nights) * F.col(rooms)) \ .withColumn(ym, F.date_format(F.to_date(checkin_date, yyyy-MM-dd), yyyy-MM))这里有个很容易忽略的细节dropDuplicates([order_id])只保留每个订单的第一条记录。CSV 是我们自己生成的不会有重复但真实业务中同一订单可能因为重试写入产生多条放在这里能积累一层防御。revnue和ym是后面所有图表的基础字段ym统一为yyyy-MM格式避免前端还要处理日期字符串。3.2 用 groupBy 和窗口函数计算间夜、ADR、城市热度酒店行业最常用的指标有三个间夜数room_nights、收入revenue、平均每日房价ADR。ADR 的计算口径是revenue / room_nights而不是avg(room_price)因为一个订单可能订多间房多晚后者算出来的数字没有业务意义。这一步的差异在实训报告里经常被面试官追问。city_metric df_clean.groupBy(city, ym).agg( F.sum(nights).alias(room_nights), F.sum(rooms).alias(room_counts), F.sum(revenue).alias(revenue), F.avg(room_price).alias(avg_price) ).withColumn(adr, F.round(F.col(revenue) / F.col(room_nights), 2)) city_metric.cache() city_metric.orderBy(ym, F.desc(room_nights)).show(10, truncateFalse)cache()会把中间结果缓存到内存后面如果反复查询这份数据能省掉重复计算。聚合结果先写show()验证数据量和格式确认没问题再落库。如果还想给答辩加一个进阶点可以用窗口函数计算累计间夜数解释“每个城市 2024 年累计间夜进度”from pyspark.sql.window import Window windowSpec Window.partitionBy(city).orderBy(ym).rowsBetween(Window.unboundedPreceding, Window.currentRow) city_metric.withColumn(cum_room_nights, F.sum(room_nights).over(windowSpec)).show(10)窗口函数的rowsBetween表示从分区第一行到当前行这是累计值的标准写法。它需要在 shuffle 时把同一城市的数据拉到同一个 task所以分区数不能太少coalesce(1)写文件之前再考虑不要在最前面用。3.3 写回 MySQL 前必须调好的 JDBC 参数Spark 写 MySQL 最直接的 API 是df.write.jdbc()但很多人第一次都会卡在驱动上。PySpark 不会自带 MySQL 驱动需要先把mysql-connector-java-8.0.33.jar下载到项目的libs/目录然后通过spark-submit --jars或pyspark --jars传进去。base_url jdbc:mysql://localhost:3306/hotel_dw props { user: spark_app, password: spark2024, driver: com.mysql.cj.jdbc.Driver, truncate: true, batchsize: 2000 } city_metric.write.mode(overwrite) \ .option(createTableOptions, ENGINEInnoDB DEFAULT CHARSETutf8mb4) \ .jdbc(base_url ?useSSLfalseserverTimezoneAsia/ShanghaicharacterEncodingutf8, city_month_agg, propertiesprops)truncatetrue在overwrite模式下会先TRUNCATE TABLE再插入比逐行DELETE快得多batchsize2000控制单个批次最多写入 2000 行调大通常能减少网络往返但会占用更多驱动内存。createTableOptions会在表不存在时自动建表的 DDL 后面追加建表选项保证 InnoDB 和 utf8mb4 生效。如果表已经存在这个参数只影响 Spark 自动建表分支。JDBC 参数建议值作用truncatetrue重跑时先清空再插入避免主键冲突batchsize2000单批写入行数过大增加内存压力rewriteBatchedStatementstrueMySQL JDBC 优化批量写入需要在 URL 中追加useSSLfalse本机连接避免 SSL 握手耗时提交到集群时本地脚本里的master配置会被spark-submit的参数覆盖驱动 JAR 用--jars指定同时--driver-class-path把它加到 driver 端否则会一直报ClassNotFound。验证写入是否成功直接在 MySQL 里执行最简单SELECT city, ym, room_nights, revenue FROM city_month_agg ORDER BY ym, room_nights DESC LIMIT 10;4. MySQL 查询接口与 Echarts 图表映射让数据可视化不只在图表上好看4.1 Flask 只读接口前端不接触数据库连接串前端直接连接 MySQL 会把账号密码暴露在浏览器里企业级数据可视化项目基本不会这么干。常见做法是后端提供一个只读 JSON 接口Flask 因为轻量适合这种聚合结果已经算好、只需要查表的场景。安装依赖后写一个最小可运行的服务from flask import Flask, jsonify import pymysql import pymysql.cursors app Flask(__name__) DB { host: localhost, port: 3306, user: spark_app, password: spark2024, database: hotel_dw, charset: utf8mb4, cursorclass: pymysql.cursors.DictCursor } app.route(/api/city_month) def city_month(): conn pymysql.connect(**DB) with conn.cursor() as cur: cur.execute(SELECT city, ym, room_nights, revenue, adr FROM city_month_agg ORDER BY ym, room_nights DESC) rows cur.fetchall() conn.close() return jsonify(rows) if __name__ __main__: app.run(host0.0.0.0, port5000, debugFalse)DictCursor让查询结果直接变成字典列表jsonify序列化后前端不用再手动拼字段。host0.0.0.0允许局域网访问演示时手机也能打开这种细节在答辩现场很加分。上面的 SQL 没有参数如果是可变的日期筛选务必改成参数化查询例如cur.execute(SELECT ... WHERE ym %s, (ym,))避免 SQL 注入。4.2 柱状图和折线图Echarts 的最简启动配置Echarts 5.x 的引入方式很简单在生产环境建议把echarts.min.js下载到static/目录下避免演示时现场网络加载超时。页面结构是一个div加一段scriptdiv idchart1 stylewidth: 100%; height: 480px;/div script src/static/echarts.min.js/script script async function loadChart() { const resp await fetch(/api/city_month); const rows await resp.json(); const months [...new Set(rows.map(r r.ym))].sort(); const cityMap {}; rows.forEach(r { if (!cityMap[r.city]) cityMap[r.city] []; cityMap[r.city].push({ ym: r.ym, nights: r.room_nights }); }); const chart echarts.init(document.getElementById(chart1)); chart.setOption({ tooltip: { trigger: axis }, legend: { top: 10 }, xAxis: { type: category, data: months }, yAxis: { type: value, name: 间夜数 }, series: Object.entries(cityMap).map(([city, arr]) ({ name: city, type: line, data: months.map(m arr.find(d d.ym m)?.nights ?? 0) })) }); } loadChart(); /script这个写法解决了两个常见问题第一series 是根据接口数据动态生成的而不是前端写死十个城市第二months.map(m ... ?? 0)会把没有数据的月份补成 0否则折线在 11 月突然断掉看起来像丢数据。tooltip.triggeraxis让鼠标悬浮时纵向对比所有城市legend自动显示城市名。如果你想要柱状图和折线图同时出现在一张图里把两个城市分别设置type: bar和type: line、yAxisIndex: 0/1即可。企业级数据可视化看板一般会用“柱状图展示间夜数、折线图展示 ADR”轴单位不一致时再加yAxisIndex区分。4.3 中国地图热力展示GeoJSON 与 Echarts 的坑搜索“echarts 中国地图”的人经常踩同一个坑Echarts 从 4.x 开始不再内置 GeoJSON 地图数据直接series.type: map, map: china只会得到空白或一个长方形边框。正确的做法是先注册地图再使用。把china.json放在static/目录前端这样加载fetch(/static/china.json) .then(res res.json()) .then(geo { echarts.registerMap(china, geo); chart.setOption({ series: [{ type: map, map: china, roam: true, data: heatData }] }); });registerMap的第一个参数是自定义地图名第二个参数是 GeoJSON 对象heatData的每一项必须是{ name: 三亚, value: 12345 }这种结构且name必须和 GeoJSON 里的行政区域名称一致。一个高频报错是控制台没有错误但地图区域不填充颜色原因通常是 GeoJSON 里的名字叫“海南”而前端传的是“海南省”需要先console.log(geo.features.map(f f.properties.name))检查真实名称。Echarts 系列适合指标动态数据注意事项line趋势、环比月份缺失时补零bar城市间夜数对比短的柱子可能显示 0需设置barMinHeightpie渠道占比数据传对象否则仍然显示旧图map城市地域热度name 必须匹配 GeoJSON如果图形只显示一个长方形排除法看两步一是 GeoJSON 是否成功 fetch二是map名称是否和注册名完全一致。另外roam: true允许图表缩放和拖拽用于演示时聚焦到某个热门区域这是让地图看起来有“企业级”质感的关键参数。5. 从实训报告到 PPT把源码、数据库脚本和图表打包成可复现的大数据项目5.1 一键重跑脚本数据更新与校验一起做项目交付给老师或同事时对方不会关心你怎么写代码只会尝试执行一个命令把整个流程跑通。把数据生成、Spark 聚合、MySQL 校验串成一个build.sh能省掉大量口述时间#!/bin/bash set -e echo 重新生成订单数据 python3 data/generate_data.py echo Spark 清洗聚合 spark-submit --jars libs/mysql-connector-java-8.0.33.jar \ --driver-class-path libs/mysql-connector-java-8.0.33.jar \ scripts/build_marts.py echo 校验结果表 mysql -h 127.0.0.1 -u spark_app -pspark2024 hotel_dw \ -e SELECT COUNT(*) AS row_count, ROUND(SUM(revenue),2) AS total_revenue FROM city_month_agg;set -e让脚本在任一命令失败时立即退出不会出现“Spark 失败了但 MySQL 校验还在继续”的假成功。密码写进脚本只适用于实训演示提交到 Git 前应改用环境变量或配置文件如果无法在答辩现场连接公网所有依赖 JAR 都放在libs/目录中这是保证可复现的最低条件。5.2 实训报告和 PPT 的叙事顺序实训报告不需要写“我用了哪些技术”这种流水账最能说明问题的是数据链路图加指标口径。建议按五页 PPT 递进第一页抛出问题“酒店度假数据分散在多个渠道如何呈现城市经营差异”第二页画架构图Spark 清洗→MySQL 聚合→Flask 接口→Echarts 展示第三页放附录中 SQL 建表和 Spark 代码片段第四页放置 3 张图表分别在评论区写出一个结论例如“三亚 ADR 在 11 月至次年 2 月明显高于其他城市”第五页列出验证过程包括随机抽样订单手算间夜数、与 MySQL 查询结果对比。最后一个可用的验证命令是用接口返回 JSON 做整体检查curl -s http://127.0.0.1:5000/api/city_month | python -c import json,sys; datajson.load(sys.stdin); assert all(r[adr]0 for r in data), ADR 异常; print(check passed)这条命令在启动 Flask 后执行会检查所有 ADR 是否大于 0任何一条负数都会让程序抛出异常并返回非零退出码也是后面接入定时调度和告警脚本的雏形。本文还有配套的精品资源点击获取