Hadoop实战：全国酒店数据离线数仓清洗与Hive分析

简介这是一份面向大数据初学者与Hadoop实践者的完整项目资料围绕全国各省市酒店数据的分析与处理展开帮助读者掌握分布式存储与MapReduce编程的核心流程。资源包共79个文件约758KB以Java源码与编译后的class文件为主辅以XML配置、properties参数文件、csv数据集及part-r-00000结果文件覆盖从代码编写到作业运行的完整链路。项目以hotel.csv为数据源通过HDFS分布式存储再用Java编写的Map与Reduce程序统计酒店总数、省市分布、平均房价等指标说明文档则记录了数据清洗、任务实现与结果解读的细节。目前已有2096人学习下载适合希望以真实案例入门Hadoop生态、理解MapReduce执行机制并积累大数据处理经验的读者参考。1. 全国酒店数据上 Hadoop从一堆 CSV 到能查能算的离线数仓手里拿到一份「全国各省市酒店数据」的 CSV几十万到几百万行不等字段无非是酒店名、省市、地址、评分、评论数、价格、星级、开业时间这几类。用 Excel 打开直接卡死用 pandas 单机跑一遍 group by 还能忍但一旦要按省、市、星级、价格区间做多维交叉统计再叠加评论数排序、评分分布单机内存就开始告急。这时候 Hadoop 的价值就出来了HDFS 负责把大文件切块存下来MapReduce 或 Hive 负责把聚合逻辑分发到多台机器上跑。这篇笔记讲的就是这条链路怎么落地数据怎么清洗、怎么传到 HDFS、怎么用 Hive 建外部表、怎么写 SQL 做省市维度的分析、中间会遇到哪些编码和分隔符的坑。适合正在做 Hadoop 课程设计、或者第一次把业务数据往集群上搬的工程师。整套流程在伪分布式和真集群上都能跑区别只是mapreduce.framework.name和资源参数。2. 先想清楚数据长什么样字段清洗与 HDFS 落地2.1 酒店数据集的典型字段与脏数据形态全国酒店数据这类数据集来源通常是爬虫或者公开数据平台导出字段结构大同小异。我一般先拿head -5和wc -l摸一遍底# 看前 5 行确认表头和分隔符 head -5 hotels.csv # 统计总行数估算数据规模 wc -l hotels.csv # 看是否有 Windows 换行符\r\n这是后面 Hive 查询出错的常见原因 file hotels.csv常见的脏数据有这几类一是省市字段不统一有的写「广东省」有的写「广东」有的干脆是「广东 深圳」挤在一个字段里二是价格字段混了「¥」符号或者「起」字三是评分字段有空值或者「暂无评分」这种文本四是 CSV 里酒店名带英文逗号导致列错位。这些问题不处理后面 Hive 建表查出来的结果就是一团乱麻。处理思路是分两步能脚本化的用 Python 或 awk 批量清洗清洗完再上传 HDFS。不要指望在 Hive 里用 SQL 把所有脏数据都洗干净那样 SQL 会写得又长又难维护。2.2 用 Python 做字段标准化与格式统一下面这段脚本做四件事统一省市名称、剥离价格里的非数字字符、把评分空值填成 -1、把清洗后的结果写成制表符分隔的文件避免酒店名里的逗号干扰。import csv import re # 省市名称映射表把各种写法归一到标准省名 PROVINCE_MAP { 广东: 广东省, 广东省: 广东省, 浙江: 浙江省, 浙江省: 浙江省, 江苏: 江苏省, 江苏省: 江苏省, # 实际项目里这张表要按数据里出现的所有写法补全 } def clean_price(raw): 从 ¥388起 这类字符串里提取数字提取不到返回 -1 if not raw: return -1 m re.search(r(\d), raw) return int(m.group(1)) if m else -1 def clean_score(raw): 评分字段空值或暂无评分统一填 -1 if not raw or 暂无 in raw: return -1.0 try: return float(raw) except ValueError: return -1.0 with open(hotels.csv, encodingutf-8) as fin, \ open(hotels_clean.tsv, w, encodingutf-8, newline) as fout: reader csv.DictReader(fin) writer csv.writer(fout, delimiter\t) # 写出表头字段名后面 Hive 建表要用 writer.writerow([hotel_name, province, city, price, score, star, comment_cnt]) for row in reader: province PROVINCE_MAP.get(row[province].strip(), row[province].strip()) writer.writerow([ row[hotel_name].strip(), province, row[city].strip(), clean_price(row.get(price, )), clean_score(row.get(score, )), row.get(star, ).strip(), row.get(comment_cnt, 0).strip() or 0, ]) print(清洗完成输出 hotels_clean.tsv)逻辑说明PROVINCE_MAP是归一化的核心实际项目里这张表要根据cut -f2 hotels.csv | sort -u的结果来补不要凭感觉写。clean_price用正则提取第一个数字串兼容「¥388」「388元」「388起」几种写法。输出用制表符分隔而不是逗号是因为酒店名里带逗号的情况太常见用逗号做分隔符迟早翻车。参数说明encodingutf-8要跟源文件编码一致如果源文件是 GBK这里要改成gbk否则第一行就报UnicodeDecodeError。newline是 csv 模块的推荐写法避免 Windows 下多出空行。2.3 上传 HDFS 并确认块分布清洗完的文件要传到 HDFS。伪分布式环境下 NameNode 默认在localhost:9000Hadoop 3.x 常见配置真集群换成实际地址。# 在 HDFS 上建目录按业务分目录是好习惯 hdfs dfs -mkdir -p /warehouse/hotel/ods/hotels # 上传清洗后的文件 hdfs dfs -put hotels_clean.tsv /warehouse/hotel/ods/hotels/ # 确认文件在 HDFS 上的块分布和副本数 hdfs fsck /warehouse/hotel/ods/hotels/hotels_clean.tsv -files -blocks # 看文件大小估算后面 MapReduce 会起几个 map hdfs dfs -du -h /warehouse/hotel/ods/hotels/fsck的输出会告诉你这个文件被切成了几个 block。默认块大小 128MB如果文件只有几十 MB就是一个 block后面 MapReduce 只会起一个 map 任务跑起来看着像单机。这不是 bug是数据量还没到。真要看并行效果要么把文件搞大要么调小dfs.blocksize。提示上传前先hdfs dfs -ls确认目标目录不存在同名文件-put遇到同名文件会直接报错退出不会覆盖。3. 用 Hive 建外部表把 HDFS 上的 TSV 变成能查的表3.1 外部表 vs 管理表为什么酒店数据要用外部表Hive 建表分管理表managed table和外部表external table。管理表删表的时候会把 HDFS 上的数据一起删掉外部表只删元数据数据还在。酒店数据这种原始数据我一般用外部表因为后面可能还要用 Spark 或者 MapReduce 直接读这份数据不想因为 Hive 里 drop 一下表就把数据搞没了。建表语句如下CREATE EXTERNAL TABLE IF NOT EXISTS ods_hotels ( hotel_name STRING COMMENT 酒店名称, province STRING COMMENT 省份, city STRING COMMENT 城市, price INT COMMENT 价格-1表示缺失, score DOUBLE COMMENT 评分-1表示缺失, star STRING COMMENT 星级, comment_cnt INT COMMENT 评论数 ) ROW FORMAT DELIMITED FIELDS TERMINATED BY \t STORED AS TEXTFILE LOCATION /warehouse/hotel/ods/hotels;逻辑说明EXTERNAL关键字决定这是外部表。ROW FORMAT DELIMITED FIELDS TERMINATED BY \t必须跟清洗时用的分隔符一致这里清洗输出的是 TSV所以是\t。LOCATION指向 HDFS 上已经上传好的目录建表的时候 Hive 不会移动数据只是把元数据登记到 metastore。参数说明price和comment_cnt用INTscore用DOUBLE如果源数据里这些字段有非数字内容查询时会返回NULL不会报错但结果会偏。所以清洗阶段把缺失值填成 -1 是有意义的查询时用WHERE price 0就能过滤掉。建完表先验证一下-- 看前 10 行确认字段没串列 SELECT * FROM ods_hotels LIMIT 10; -- 统计总行数跟 wc -l 的结果对一下 SELECT COUNT(*) FROM ods_hotels;如果SELECT *出来的字段明显错位八成是分隔符不对或者源文件里有残留的\r。\r的问题可以在清洗脚本里用line.replace(\r, )处理掉。3.2 分区表改造按省份分区提升查询效率全国酒店数据按省份查询是最常见的场景。如果每次查询都全表扫描数据量大了之后很慢。Hive 分区表可以把数据按省份物理分开存查询时只扫对应分区。改造方式是建一张分区表然后用INSERT ... SELECT从外部表导数据-- 建分区表按省份分区 CREATE TABLE IF NOT EXISTS dw_hotels_partitioned ( hotel_name STRING, city STRING, price INT, score DOUBLE, star STRING, comment_cnt INT ) PARTITIONED BY (province STRING) STORED AS ORC; -- 开启动态分区让 Hive 根据 province 字段自动创建分区 SET hive.exec.dynamic.partition true; SET hive.exec.dynamic.partition.mode nonstrict; -- 从外部表导数据到分区表 INSERT OVERWRITE TABLE dw_hotels_partitioned PARTITION (province) SELECT hotel_name, city, price, score, star, comment_cnt, province FROM ods_hotels WHERE province IS NOT NULL AND province ! ;逻辑说明PARTITIONED BY (province STRING)把省份从普通字段变成分区字段注意SELECT里 province 要放在最后一列跟分区字段的顺序对应。hive.exec.dynamic.partition.mode nonstrict是必须的否则 Hive 要求至少指定一个静态分区动态分区插不进去。参数说明STORED AS ORC比 TEXTFILE 省空间、查询快适合做中间层。如果集群开了 Tez 执行引擎ORC 格式的收益更明显。导完之后可以SHOW PARTITIONS dw_hotels_partitioned看分区列表确认每个省都建出来了。注意动态分区会按 province 的 distinct 值创建分区如果 province 字段有大量脏值比如空字符串、乱码会创建出一堆没用的分区。导数据前先SELECT DISTINCT province FROM ods_hotels看一眼。4. 省市维度分析几个能直接抄的 Hive SQL4.1 按省统计酒店数量、均价、平均评分这是最基础的多维聚合一条 SQL 能出结果SELECT province, COUNT(*) AS hotel_cnt, ROUND(AVG(price), 2) AS avg_price, ROUND(AVG(score), 2) AS avg_score, SUM(comment_cnt) AS total_comments FROM dw_hotels_partitioned WHERE price 0 AND score 0 GROUP BY province ORDER BY hotel_cnt DESC;逻辑说明WHERE price 0 AND score 0把清洗阶段填的 -1 过滤掉避免缺失值拉低均价。ROUND保留两位小数ORDER BY hotel_cnt DESC让酒店最多的省排前面。参数说明如果数据量很大ORDER BY会触发一个 reduce 做全局排序可能比较慢。如果只是看排名可以改成ORDER BY hotel_cnt DESC LIMIT 20减少数据传输。4.2 城市价格区间分布用 CASE WHEN 做分桶想知道每个城市里经济型、中端、高端的酒店各占多少用CASE WHEN分桶SELECT province, city, CASE WHEN price 200 THEN 经济型 WHEN price 200 AND price 500 THEN 中端 WHEN price 500 AND price 1000 THEN 高端 ELSE 豪华 END AS price_level, COUNT(*) AS cnt FROM dw_hotels_partitioned WHERE price 0 GROUP BY province, city, CASE WHEN price 200 THEN 经济型 WHEN price 200 AND price 500 THEN 中端 WHEN price 500 AND price 1000 THEN 高端 ELSE 豪华 END ORDER BY province, city, cnt DESC;逻辑说明CASE WHEN的分桶逻辑要跟业务对齐这里的价格阈值只是示例实际项目里要根据数据分布来定。GROUP BY里必须把CASE WHEN表达式完整重复一遍Hive 不支持按别名分组。参数说明如果分桶规则经常变建议把阈值抽成配置或者干脆在 ETL 阶段就多算一列price_level存到表里查询时直接 group by 这一列SQL 更干净。4.3 用窗口函数排城市内酒店评分 Top N每个城市评分最高的前 5 家酒店用ROW_NUMBER()窗口函数SELECT province, city, hotel_name, score, price FROM ( SELECT province, city, hotel_name, score, price, ROW_NUMBER() OVER (PARTITION BY province, city ORDER BY score DESC, comment_cnt DESC) AS rn FROM dw_hotels_partitioned WHERE score 0 ) t WHERE rn 5 ORDER BY province, city, rn;逻辑说明PARTITION BY province, city把数据按省市分组ORDER BY score DESC, comment_cnt DESC在组内按评分降序、评论数降序排ROW_NUMBER()给每行编个号。外层查询过滤rn 5就拿到每个城市的前 5 名。参数说明ROW_NUMBER()遇到相同评分不会并列如果要并列排名用RANK()或DENSE_RANK()。窗口函数在 Hive 里是支持的但如果集群版本很老0.11 之前需要用UDF或者改写 SQL这种情况现在很少见了。5. 避坑与排查酒店数据上 Hadoop 常见的 5 个翻车点5.1 中文乱码查询结果全是问号现象Hive 查询出来的酒店名和省市全是???或者乱码。原因源文件编码是 GBK清洗脚本按 UTF-8 读或者 Hive 建表时没指定字符集metastore 用了默认的 latin1。解决清洗阶段用chardet检测源文件编码或者直接file -i hotels.csv看。Hive 这边在建表时加TBLPROPERTIES (charsetutf-8)同时确认hive-site.xml里javax.jdo.option.ConnectionURL的字符集参数带了useUnicodetruecharacterEncodingUTF-8。5.2 字段错位酒店名里带逗号导致列偏移现象SELECT *出来发现 city 字段里是酒店名的一部分后面所有列都往右偏了。原因源 CSV 用逗号分隔但酒店名里本身带逗号csv.DictReader能处理带引号的字段但如果源文件里逗号没被引号包起来就会错列。解决清洗阶段不要用split(,)这种土办法用csv模块的DictReader。如果源文件本身格式就不规范先用awk -F, NF!7 hotels.csv把列数不对的行捞出来单独看。输出统一用\t分隔从根上避开逗号问题。5.3 动态分区报错requires at least one static partition现象INSERT OVERWRITE ... PARTITION (province)执行时报FAILED: SemanticException [Error 10096]: Dynamic partition cannot be the parent of a static partition。原因hive.exec.dynamic.partition.mode默认是strict要求至少有一个静态分区。解决执行前SET hive.exec.dynamic.partition.mode nonstrict;。这个设置只在当前会话有效换个会话要重新设。如果用的是 Beeline可以在连接串里加--hiveconf hive.exec.dynamic.partition.modenonstrict。5.4 小文件过多每个分区一堆几百 KB 的文件现象hdfs dfs -ls /warehouse/hotel/dw_hotels_partitioned/province广东省/看到几十个几百 KB 的小文件。原因动态分区插入时每个 map 任务会为每个分区生成一个文件map 数量多、分区多小文件就爆炸了。解决导完数据后跑一次合并-- 合并小文件 ALTER TABLE dw_hotels_partitioned PARTITION (province广东省) CONCATENATE;或者在建表时设置hive.merge.mapfiles true和hive.merge.mapredfiles true让 Hive 在任务结束时自动合并。更彻底的办法是控制 map 数量导数据前SET mapred.reduce.tasks 10;限制 reduce 个数。5.5 MapReduce 任务卡在 reduce 阶段不动现象任务跑到 99% 卡住reduce 阶段一直不结束。原因数据倾斜。某个省的酒店数据量特别大分到同一个 reduce 上其他 reduce 早就跑完了就它还在跑。解决先确认是不是倾斜看 JobTracker 或者 YARN 的 UI哪个 reduce 的输入记录数明显比别人大。如果是倾斜可以在 SQL 里加随机前缀打散-- 给 province 加随机前缀打散到多个 reduce SELECT CONCAT(province, _, CAST(RAND() * 10 AS INT)) AS province_key, COUNT(*) FROM dw_hotels_partitioned GROUP BY CONCAT(province, _, CAST(RAND() * 10 AS INT));这个办法会让结果多出 10 倍的行外层再聚合一次就行。或者直接SET hive.groupby.skewindata true;让 Hive 自动做两阶段聚合。6. 进阶用 MapReduce 直接处理酒店数据与结果验证Hive SQL 写起来快但有些定制化的清洗逻辑或者复杂的评分计算SQL 表达起来别扭这时候直接写 MapReduce 更灵活。下面这个例子统计每个省的平均价格用 Java 写一个最简单的 MapReduce 作业。// Mapper解析 TSV输出 province, price public class HotelPriceMapper extends MapperLongWritable, Text, Text, IntWritable { private Text province new Text(); private IntWritable price new IntWritable(); Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String line value.toString(); // 跳过表头 if (line.startsWith(hotel_name)) return; String[] fields line.split(\t); // 字段数不对或者价格无效的直接跳过 if (fields.length 7) return; try { int p Integer.parseInt(fields[3]); if (p 0) return; province.set(fields[1]); price.set(p); context.write(province, price); } catch (NumberFormatException e) { // 价格字段解析失败跳过这一行 } } }// Reducer对每个省的价格求平均 public class HotelPriceReducer extends ReducerText, IntWritable, Text, DoubleWritable { private DoubleWritable result new DoubleWritable(); Override protected void reduce(Text key, IterableIntWritable values, Context context) throws IOException, InterruptedException { long sum 0; int count 0; for (IntWritable val : values) { sum val.get(); count; } if (count 0) { result.set((double) sum / count); context.write(key, result); } } }逻辑说明Mapper 按\t切分取第 4 个字段下标 3作为价格第 2 个字段下标 1作为省份。line.startsWith(hotel_name)跳过表头fields.length 7过滤掉格式不对的行。Reducer 累加价格和计数最后输出平均值。参数说明split(\t)如果遇到字段里本身带\t的情况会切错但清洗阶段已经保证了输出格式所以这里可以放心用。如果数据里可能有\t改用split(\t, -1)保留尾部空字段。打包提交到集群# 打包 mvn clean package # 提交作业 hadoop jar hotel-analysis.jar com.example.HotelPriceDriver \ /warehouse/hotel/ods/hotels/hotels_clean.tsv \ /warehouse/hotel/output/avg_price # 看结果 hdfs dfs -cat /warehouse/hotel/output/avg_price/part-r-00000 | head -20跑完之后把 MapReduce 的结果跟 Hive SQL 的结果对一下-- Hive 侧的结果 SELECT province, ROUND(AVG(price), 2) AS avg_price FROM dw_hotels_partitioned WHERE price 0 GROUP BY province ORDER BY province;两边结果应该一致。如果不一致优先检查 MapReduce 里有没有漏掉price 0的过滤或者 Hive 侧有没有把空省份算进去。这种交叉验证是我每次做完离线任务都会做的比单纯看一个结果靠谱得多。最后一个习惯所有清洗脚本、Hive SQL、MapReduce 代码都放到 Git 里按etl/、sql/、mr/分目录。酒店数据这种项目字段和口径经常变没有版本管理过两周自己都忘了当时怎么算的。希望帮到你。本文还有配套的精品资源点击获取