网站流量日志数据分析系统听起来就是一个把 Nginx 日志接进来、算出 PV/UV、出报表的离线任务。可真动手做一轮你会发现这套链路里藏了太多“看起来简单、做起来想骂人”的细节。我去年从头到尾搭建过一套这样的系统今天把它拆开聊透适合准备入门 Hadoop 离线数仓、或者想把公司日志分析体系补全的朋友做参考。整套东西不依赖实时计算HDFS Hive 调度 导出就是全部家当但怎么把每个环节做扎实才是这套系统真正的价值所在。1. 项目到底要做什么需求拆解与技术选型1.1 网站流量指标的业务口径从哪来我们当时接的是一个内部业务平台的访问日志日增量大概在 3 亿条左右压缩落地后一百多 GB。业务方张嘴就要 PV、UV、热门页面、访问时长、跳出率、来源分析后面还要按省份、按浏览器拆维度。听起来都是常见指标但真要对齐口径就会发现每个指标都需要业务方确认定义PV 是“页面浏览量”那静态资源算不算jpg、css、js 必须滤掉。UV 是“独立访客”按什么识别IP、cookie、还是登录用户 ID如果按 IP局域网代理会导致 UV 被严重低估。跳出率指的是“只访问了一个页面就离开的会话占比”那就必须先定义什么是“一个会话”30 分钟无新动作算一次会话过期每次点击算新会话这些口径没定清楚后面所有 HQL 都会白写。我的做法是拉上业务方一起列了一张指标口径清单把每个指标的名称、定义、统计周期、维度字段写清楚然后再开始设计表结构。技术侧的需求其实不复杂每天凌晨跑一次数据处理上午 9 点前把结果推到报表库给运营和产品看昨天的数据。它天然就是一个 T1 的离线场景对实时性几乎没有任何要求。数据量虽然到了百 GB 级但离线批处理完全扛得住。1.2 为什么是离线 Hadoop 而不是 Spark/Flink这个项目选型时不是没有纠结过。团队里有人提出用 Spark 会不会更快用 Flink 是不是能把时效升级到分钟级。我当时的判断是不要为了一点点性能优势把整套系统的复杂度抬上去。时效要求业务方只要“昨天全天数据”晚上 12 点后开始跑完全来得及。离线 T1 是性价比最高的方案。技术栈匹配团队对 Hive SQL 的熟悉程度远高于 Scala 和 JavaSpark 虽快但需要额外排查内存问题Flink 更不必说状态管理、精确一次语义、窗口机制学习成本高。基于 HDFS 的离线链路生态成熟Flume 采集、Hive 计算、Sqoop 导出、调度平台跑任务每一个环节都有大量公开实践可以抄作业。数据量级也没到非用 Spark 不可的程度。3 亿条日访问日志Hive 跑聚合任务MapReduce 引擎下大概 40 到 60 分钟能跑完如果用 Tez 引擎还能再快一些。为了快这半小时去迁移 Spark完全不值得。所以最终方案用了 Hadoop 离线全家桶Flume 负责日志采集HDFS 做存储Hive 做清洗和指标计算Oozie 负责任务编排Sqoop 把结果导出到 MySQL最后接报表平台。链路看起来长但每个环节都有现成的轮子可以复用。1.3 组件选型和链路设计整体链路我用文字梳理一遍流程如下Nginx 将访问日志写到本地磁盘日志格式配置成统一的 JSON 或 Key-Value 结构。每台业务服务器部署 Flume Agent用 taildir source 监控日志文件将新写入的数据发送到采集集群的 Avro source。Flume 汇聚层将数据下沉到 HDFS按日期和小时拆分目录写入时直接压缩成 gzip 文件。次日由调度平台触发 Hive 脚本先加载原始日志到 ODS 层外部表再清洗过滤得到 DWD 层明细表然后按维度汇总到 DWS 层最后生成 ADS 层业务指标表。Sqoop 将 ADS 层结果表导出到 MySQL。报表平台或邮件报表直接查询 MySQL给业务方展示。这套链路里最容易被忽略的是 Flume 的 sink 端小文件控制。Flume 默认写 HDFS 会按你设置的滚动时间生成文件如果设成 60 秒滚动一次一晚上下来会出现几万个小文件后面跑 Hive 时 NameNode 压力巨大。这个问题后面第 5 节细讲。选型上还有一个细节我们最初用 Hive 默认的 MapReduce 引擎跑 3 亿条日志的 UV 去重要 50 分钟。后来把执行引擎切成 Tez相同任务压到了 30 分钟左右改动成本很低。所以如果你在用 Hive 3.x强烈建议检查一下hive.execution.engine配置别死守 MapReduce。2. 日志采集与预处理数据是地基2.1 日志格式要注意的坑Web 服务器日志最常见的坑就是“字段不完整”和“格式五花八门”。我们接手时发现 Nginx 日志里有几条明显缺字段-代表空值、URL 里带着无法解码的乱码、user_agent 字段里混着换行符直接切分会导致字段错位。所以第一步是让 Nginx 输出为统一 JSON 格式。我们在 Nginx 配置里用log_format将remote_addr、time_local、request、status、body_bytes_sent、http_referer、http_user_agent、request_time等字段拼成 JSON。这样做比用默认的combined格式好在解析时只需要 JSON 解析不用做空格切分字段缺失时也能保留空字符串而不是直接错位。但 JSON 也有新坑业务方在 URL 上带一些特殊字符例如引号、反斜杠会导致生成的 JSON 不合法需要在 Nginx 侧把特殊字符做转义。我们线上就遇到过用户搜索词里带双引号直接把整行日志干成脏数据的情况。除了格式字段补充也要在源头考虑。原始日志没有省份信息只能拿到 IP所以 ETL 阶段要用 IP 库把 IP 解析成省份、城市user_agent 里有浏览器和系统信息需要用解析库拆出browser、os、device_type字段。这些操作放在清洗阶段做不放在采集端因为 Flume 不擅长复杂计算。2.2 Flume 采集配置实战采集端用的是 Flume 1.9核心诉求是不丢数据、不产生大量小文件、能断点续采。这里我给出一份常用的采集 Agent 配置大家可以直接参考# 业务服务器上的 Flume Agenttaildir avro sink a1.sources r1 a1.channels c1 a1.sinks k1 a1.sources.r1.type TAILDIR a1.sources.r1.positionFile /data/flume/taildir_position.json a1.sources.r1.filegroups f1 a1.sources.r1.filegroups.f1 /data/nginx/logs/access.log a1.sources.r1.fileHeader true a1.channels.c1.type file a1.channels.c1.checkpointDir /data/flume/checkpoint a1.channels.c1.dataDirs /data/flume/data a1.sinks.k1.type avro a1.sinks.k1.hostname flume-agg-01 a1.sinks.k1.port 4141 a1.sources.r1.channels c1 a1.sinks.k1.channel c1汇聚层 Agent 的 sink 写 HDFS关键配置如下# 汇聚层 Flume Agentavro source hdfs sink a2.sources r1 a2.channels c1 a2.sinks k1 a2.sources.r1.type avro a2.sources.r1.bind 0.0.0.0 a2.sources.r1.port 4141 a2.channels.c1.type file a2.sinks.k1.type hdfs a2.sinks.k1.hdfs.path /warehouse/ods/raw_log/date%Y-%m-%d/hour%H a2.sinks.k1.hdfs.filePrefix log a2.sinks.k1.hdfs.fileType CompressedStream a2.sinks.k1.hdfs.codec gzip a2.sinks.k1.hdfs.rollInterval 300 a2.sinks.k1.hdfs.rollSize 268435456 a2.sinks.k1.hdfs.rollCount 0 a2.sinks.k1.hdfs.batchSize 2000这段配置里有几个点很容易踩坑rollCount0表示不按事件条数滚动文件只按时间和大小。rollSize是每个文件达到 256MB 就滚动。这个值要配合后续 Hive 读取不是越大越好我们的经验是 128MB 到 512MB 都行。batchSize控制批量写 HDFS 的事务大小太大会增加内存压力2000 是比较稳妥的值。Flume 埋点重启后为什么不会丢数据依赖 taildir source 维护的 position 文件它把每个文件读取的偏移量记录下来了Agent 重启后从偏移量位置继续读。但要注意 positionFile 所在目录的权限否则 Flume 起不来数据会延迟采集到早上的调度时点还没采完。2.3 清洗与富化ETL 该做哪些事原始日志落 HDFS 后接下来是以 Hive SQL 为主做清洗和富化。这个阶段目标很明确过滤无效数据去掉状态码在 400 以上的请求去掉以.js、.css、.png、.jpg、.ico结尾的静态资源请求去掉明显是爬虫的 user_agent。字段标准化把时间字段从字符串转成 timestamp把 URL 解析出 path 和 query把 IP 解析成省份。会话标识生成为每条访问记录生成一个可聚合的session_id后面算跳出率和平均访问时长就靠它。清洗阶段的 HQL 是整套系统里最长最繁琐的因为你要处理各种脏数据。比如 IP 字段出现-时间字段出现nulluser_agent 里出现半个中文字符。我的建议是清洗分两层第一层做硬过滤直接丢脏数据第二层做字段弥补能解析的解析不能解析的填unknown不要因为一条脏数据让整个任务失败。这一层 HQL 示例大致长这样INSERT OVERWRITE TABLE dwd_access_log PARTITION (dt ${dt}) SELECT remote_addr, parse_url(request_url, PATH) AS request_path, parse_url(request_url, QUERY) AS request_query, status, body_bytes_sent, from_unixtime(unix_timestamp(time_local, yyyy-MM-dd HH:mm:ss)) AS access_time, ip_to_province(remote_addr) AS province, parse_user_agent(user_agent, browser) AS browser, parse_user_agent(user_agent, os) AS os FROM ods_raw_log WHERE dt ${dt} AND status 400 AND request_path NOT RLIKE \\.(js|css|png|jpg|ico|svg|woff|woff2)$ AND remote_addr ! -;有人会问为什么不直接对 ODS 表做聚合非要清洗一层明细如果直接把原始日志全量保存每天重复跑聚合时耗时会越来越长而且明细层可以反复复用后面加一个新指标只需要从明细层重新聚合一次不需要再解析一遍原始字符串。砍掉哪一层的干净明细后面新增需求时哭的就是你。3. Hive 数仓建模与指标计算整套系统的重头3.1 数仓分层别一张表打天下很多入门同学做日志分析时只建一张表所有字段全塞进去然后直接在原始表上跑count(distinct ...)。小数据量感觉没问题一到亿级数据就原形毕露跑一次要一小时而且团队里每个人都写自己的临时 SQL口径完全对不上。这套系统我按标准数仓四层模型来搭建ODS 层ods_raw_log只做原始日志落地与日志源字段一一对应保留原始数据用于回溯和排查。DWD 层dwd_access_log经过清洗、过滤、维度补全、会话标识生成的明细事实表粒度是一条访问记录。DWS 层dws_flow_stats按业务维度预先聚合比如按日期 省份 浏览器统计 PV/UV/会话数。ADS 层ads_flow_report面向业务方的最终结果表按天存储核心指标供报表查询。这样分层最大的好处是基础数据只清洗一次所有人都从 DWD 层读取指标口径收敛在 DWS 层每次加指标只影响新增字段不动老逻辑。我把“分层”理解为“把做饭和吃饭分开”ODS 是菜市场买回来的菜DWD 是洗好切好的菜DWS 是半成品ADS 是直接上桌的成品。3.2 核心表结构与分区策略ODS 表我建成了 Hive 外部表数据由 Flume 直接写到表目录下表本身没做复杂压缩转换CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log ( remote_addr STRING, time_local STRING, request_url STRING, status INT, body_bytes_sent BIGINT, http_referer STRING, user_agent STRING, request_time FLOAT ) PARTITIONED BY (dt STRING, hour STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY \t STORED AS TEXTFILE LOCATION /warehouse/ods/raw_log;注意外部表这个细节。如果 Hive 表是内部表一个误操作DROP TABLE会把 HDFS 上的原始日志全删掉找不回来外部表删表只删元数据数据文件还在 HDFS 上这是日志数据最基本的保命手段。DWD 层建议用列式存储加压缩CREATE TABLE IF NOT EXISTS dwd_access_log ( remote_addr STRING, province STRING, browser STRING, os STRING, request_path STRING, request_query STRING, status INT, body_bytes_sent BIGINT, access_time TIMESTAMP, session_id STRING ) PARTITIONED BY (dt STRING) STORED AS PARQUET TBLPROPERTIES (parquet.compressionSNAPPY);使用 Parquet 后的体积只有原始文本的 1/4 左右而且列式存储在按省份、浏览器做聚合时更快。为什么分区只保留dt而不是dt hour因为清洗后是按天分析小时分区只会导致大量小文件没必要保留小时粒度。3.3 PV/UV/跳出率等指标的计算口径与 SQL指标计算是整个系统的核心也是口径最容易打架的地方。我贴几个关键指标的 HQL 片段和边界条件说明。PV是最简单的SELECT dt, COUNT(*) AS pv FROM dwd_access_log WHERE dt ${dt} GROUP BY dt;UV如果直接用COUNT(DISTINCT remote_addr)在亿级数据量下很容易因为数据倾斜变慢。优化思路是先做一层子查询去重再 countSELECT dt, COUNT(*) AS uv FROM ( SELECT remote_addr FROM dwd_access_log WHERE dt ${dt} GROUP BY remote_addr ) t;这里要注意“UV 按 IP 口径”和“UV 按设备口径”是有本质差别的。如果业务方要求按 cookie 或用户 ID 去重DWD 层必须保留对应字段HQL 改 group by 字段即可。会话划分是跳出率和平均访问时长的基础。我们用的标准是同一个remote_addr或 cookie下相邻两条访问记录时间差超过 30 分钟算作一个新的会话。在 Hive 里可以用窗口函数LAG实现SELECT remote_addr, access_time, session_id FROM ( SELECT remote_addr, access_time, CASE WHEN LAG(access_time) OVER ( PARTITION BY remote_addr ORDER BY access_time ) IS NULL OR UNIX_TIMESTAMP(access_time) - UNIX_TIMESTAMP( LAG(access_time) OVER ( PARTITION BY remote_addr ORDER BY access_time ) ) 1800 THEN CONCAT(remote_addr, _, ROW_NUMBER() OVER ( PARTITION BY remote_addr ORDER BY access_time )) ELSE NULL END AS session_id FROM dwd_access_log WHERE dt ${dt} ) t;这段 SQL 生成的session_id会在下一次 CASE 判断时通过上一个值带下去所以必须再做一次LAST_VALUE或“向前填充”补齐 session_id。更标准的分会话方案是把数据导出到 Spark 或写 UDF 去做但在不引新组件的前提下Hive 窗口函数也能完成只是脚本会稍长。跳出率的定义是“跳出会话数 / 总会话数”。一个会话内如果只有一次页面浏览就算跳出SELECT dt, SUM(IF(session_cnt 1, 1, 0)) / COUNT(session_id) AS bounce_rate FROM ( SELECT session_id, COUNT(*) AS session_cnt FROM dwd_access_log WHERE dt ${dt} GROUP BY session_id ) t;平均访问时长同理先算出每个会话的MAX(access_time) - MIN(access_time)再求均值。需要特别说的是指标 SQL 写出来不是终点跑出来的数字要和业务方预期核对。我记得第一版跳出率脚本跑出来 68%业务方说“不可能这么高”后来一查是测试环境的一批爬虫只抓首页就离开把跳出率拉高了。过滤爬虫口 HQL 加了一个白名单判断再跑数值才回到 35% 左右。指标异常很多时候不是代码错而是口径和现实没对齐。4. 调度、导出与报表展示把离线链路串起来4.1 为什么不用 crontab 硬顶离线链路有三个任务阶段Flume 落 HDFS、Hive 清洗聚合、Sqoop 导出。如果只用 crontab每个任务之间的依赖只能靠“时间估算”去硬等例如设定 2 点跑清洗、6 点跑导出。集群一忙计算延迟 20 分钟清洗任务还没结束导出任务已经把空结果推到 MySQL 了。所以调度这么做用一个工作流引擎把三个环节编排成 DAG每个节点执行具体的脚本节点之间做依赖检查上游成功才触发下游失败自动重试最终任务全跑完可以发一条日报“数据已更新”。我们用过 Oozie后来切到了 Azkaban。简单对比一下能力OozieAzkaban配置方式XML复杂properties直观任务依赖支持 DAG支持 DAG失败重跑支持且灵活支持指定节点重跑使用体验学习成本高上手快如果你不是为了兼容 Hadoop 老版本生态我更推荐 Azkaban。它用纯 properties 就能定义任务还能在 Web UI 上直接点击失败节点重跑省去了频繁登录服务器敲命令的麻烦。调度脚本的核心是一个 Shell 包装传dt参数给 HQL并保证每个任务幂等。幂等是什么意思同一份数据重跑多次结果一致不会因为重试导致数据翻倍。最简单的方式是每个 HQL 开头都执行INSERT OVERWRITE指定dt分区这样重跑就是覆盖写不会留脏数据。4.2 Sqoop 导出细节指标算完后要推给业务看Hive 表直接查询不现实所以用 Sqoop 把 ADS 层表导出到 MySQLsqoop export \ --connect jdbc:mysql://mysql-host:3306/analytics \ --username reader --password xxx \ --table ads_flow_report \ --export-dir /warehouse/ads/ads_flow_report/dt${dt} \ --input-fields-terminated-by \001 \ --update-mode allowinsert \ --update-key dt \ --num-mappers 4这里有两个关键点并行度num-mappers控制并行导出线程数不要盲目调到 30。MySQL 写入能力有限并行太高会直接打爆数据库连接我们一般设 4 到 8 个。幂等update-mode allowinsert配合update-key dt让 Sqoop 对主键冲突的行执行 update不冲突则 insert这样重跑同样日期的导出不会导致报表出现两行相同日期数据。还有一个容易忽略的问题Hive 导出到 MySQL 时字段分隔符要统一。Hive 表如果设置了自定义分隔符比如\001Sqoop 必须要加上--input-fields-terminated-by \001否则 sqoop 默认按逗号切分所有字段都会错位报表里数字全变成 0。4.3 报表侧优化与数据校验指标表导出到 MySQL 后报表平台直接查询这张表。但 MySQL 单表数据量过大时查询会变慢所以要在报表表上建索引按dt province browser建联合索引这样业务方按天筛选时能快速出数。同时报表侧不是只看数字我们对每天的结果做数据质量校验当日 PV 和前一天对比波动超过 30% 自动告警。当日 UV 和前一天对比波动超过 30% 自动告警。检查 MySQL 报表表当天是否有数据、是否有重复行。这些校验用 Shell 脚本 定时任务就能完成不用引入复杂的监控系统。数据团队接到告警后第一时间检查 Nginx 日志是否有断流、Flume 是否有积压、Hive 任务是否有重试。5. 离线系统常见问题与排查实录5.1 小文件堆积离线任务的第一杀手我们系统刚上线时跑了一周之后发现 Hive 任务越跑越慢NameNode 的 RPC 请求经常超时。一查 HDFS 目录Flume 按小时分区每个分区下有几百个几 KB 到几十 MB 的小文件。因为配置里rollInterval300每 5 分钟就滚动一次一个小时内 12 个文件一天 288 个文件一周 2000 多个。小文件的危害不只是占 NameNode 内存Hive 读取时要同时打开几千个文件Map Task 数量暴涨集群调度和计算成本跟着翻倍。解决思路有两层第一层在源头控制把 Flume 的rollSize调大到 256MBrollInterval调大到 30 分钟让文件尽量大、数量尽量少。第二层在周期上兜底每天调度一个文件合并任务对当天的分区做INSERT OVERWRITE重新写成少量大文件INSERT OVERWRITE TABLE ods_raw_log PARTITION (dt ${dt}) SELECT * FROM ods_raw_log WHERE dt ${dt};注意这一步也会触发一次全分区重读所以只对时间超过一天的历史分区做定期合并比如一周合并一次。千万不要对当前正在写入的分区执行否则新采到的日志会被覆盖掉。5.2 数据倾斜按天聚合时最容易踩COUNT(DISTINCT remote_addr)最大的问题是会把相同 IP 的数据分到同一个 Reduce热门 IP 可能携带了上百万条记录单 Reduce 跑到内存溢出。我们专门做过一次数据倾斜排查发现某个爬虫 IP 在一天内产生了近 200 万条请求导致 UV 任务卡了 40 分钟没跑完。解决方式最常用的是“两阶段聚合”先打随机前缀分两次 group by。第一层把相同 key 打散到不同 Reduce第二层去掉前缀再聚合。在 Hive 里可以写成嵌套子查询SELECT dt, COUNT(*) AS uv FROM ( SELECT tmp.remote_addr FROM ( SELECT CONCAT(CAST(RAND() * 10 AS INT), _, remote_addr) AS diy_key, remote_addr FROM dwd_access_log WHERE dt ${dt} GROUP BY CONCAT(CAST(RAND() * 10 AS INT), _, remote_addr), remote_addr ) tmp GROUP BY tmp.remote_addr ) t;另外Hive 3.x 自带hive.groupby.skewindatatrue配置可以在不做手动加盐的前提下自动分桶聚合但手动加盐对倾斜场景更可控。加了这层优化之后UV 任务从卡死状态降到 15 分钟内跑完。5.3 重复数据与回溯重跑离线系统最怕的就是“跑错了重跑一遍数据不对”。我们遇到过一次调度平台因为网络闪断重跑了当天的清洗任务但 Sqoop 导出节点没有一起重跑导致 MySQL 里缺当天数据后来又手动补跑导出结果因为update-mode allowinsert配置没加插入了一行重复数据报表里出现两个相同 dt 的记录。这个问题的根治思路是让整条链路的每个节点都具备幂等性。Hive 脚本一律用INSERT OVERWRITESqoop 导出必须带--update-key。这样无论失败重跑多少遍结果都不会叠加。数据回溯时也要注意如果业务方要求重新计算过去 7 天数据调度脚本要把dt参数改成循环传入逐天执行不要一口气写一个跑 7 天数据的 SQL。逐天跑的好处是失败后只需要重跑失败那天的数据而不是整个区间推倒重来。5.4 任务超时与资源分配Hive 离线任务在集群资源紧张时很容易出现“任务排队半小时Map 阶段跑 40 分钟Reduce 阶段再跑 20 分钟”的情况。尤其是多张业务表同时跑日任务时资源队列抢占非常明显。我们当时的处理是两个方面给不同优先级任务分配独立 YARN 队列核心日报任务调度到高优先级队列临时分析任务到低优先级队列避免相互抢占。设置合理的 Hive 容器参数例如set hive.tez.container.size4096; set hive.tez.java.opts-Xmx3072m; set mapreduce.map.memory.mb4096; set mapreduce.reduce.memory.mb4096;注意容器内存不能超过 YARN 单节点可用内存不然任务会直接失败。先跑一次小分区数据用yarn logs查看实际使用的 Container 内存再微调参数比随便拍脑袋设一个大值靠谱得多。5.5 数据质量校验没有监控就不算完成系统上线前我花了不少时间写数据质量校验。最基础的三道检查数据完整性ODS 和 DWD 层当天分区是否存在记录数是否为 0。波动监控核心指标 PV/UV 的环比波动是否超过 30%。超过就触发告警人工排查是业务活动、流量异常还是采集故障。一致性校验DWS 和 ADS 的 PV 相加是否等于 ODS 明细总量避免聚合时漏数据。这套脚本放在每日任务最后如果校验失败直接给相关人员发报警并阻止报表导出。没有监控的离线系统就是走钢丝数据错了业务方已经看了一上午报表再回去解释就非常被动。最后再分享一个小技巧整个项目踩坑最多的地方其实集中在数据口径和幂等设计。很多同学搭离线系统时觉得“能跑就行”但真正上线跑一段时间就会明白离线链路的核心不是某个组件用得有多熟而是把每个环节都做成可重跑、可校验、不重复。哪怕你只是给一个小站点搭日志分析也建议从第一天就做好分层和幂等。我个人实际操作中的习惯是每个 Hive 脚本顶部都有一段统一的“参数校验”脚本判断传入的dt是否合法、要计算的分区是否存在出现问题直接失败退出避免在错误参数下跑出全表脏数据。这套系统后来还接入了更多业务表加了主题宽表也在尝试往实时方向扩展。但说起来离线 Hadoop 这套 T1 链路永远是数据分析的地基把它吃透后面学什么新框架都不慌。希望这篇复盘能帮你少踩几个坑。 网站流量日志数据分析系统听起来就是一个把 Nginx 日志接进来、算出 PV/UV、出报表的离线任务。可真动手做一轮你会发现这套链路里藏了太多“看起来简单、做起来想骂人”的细节。我去年从头到尾搭建过一套这样的系统今天把它拆开聊透适合准备入门 Hadoop 离线数仓、或者想把公司日志分析体系补全的朋友做参考。整套东西不依赖实时计算HDFS Hive 调度 导出就是全部家当但怎么把每个环节做扎实才是这套系统真正的价值所在。1. 项目到底要做什么需求拆解与技术选型1.1 网站流量指标的业务口径从哪来我们当时接的是一个内部业务平台的访问日志日增量大概在 3 亿条左右压缩落地后一百多 GB。业务方张嘴就要 PV、UV、热门页面、访问时长、跳出率、来源分析后面还要按省份、按浏览器拆维度。听起来都是常见指标但真要对齐口径就会发现每个指标都需要业务方确认定义PV 是“页面浏览量”那静态资源算不算jpg、css、js 必须滤掉。UV 是“独立访客”按什么识别IP、cookie、还是登录用户 ID如果按 IP局域网代理会导致 UV 被严重低估。跳出率指的是“只访问了一个页面就离开的会话占比”那就必须先定义什么是“一个会话”30 分钟无新动作算一次会话过期每次点击算新会话这些口径没定清楚后面所有 HQL 都会白写。我的做法是拉上业务方一起列了一张指标口径清单把每个指标的名称、定义、统计周期、维度字段写清楚然后再开始设计表结构。技术侧的需求其实不复杂每天凌晨跑一次数据处理上午 9 点前把结果推到报表库给运营和产品看昨天的数据。它天然就是一个 T1 的离线场景对实时性几乎没有任何要求。数据量虽然到了百 GB 级但离线批处理完全扛得住。1.2 为什么是离线 Hadoop 而不是 Spark/Flink这个项目选型时不是没有纠结过。团队里有人提出用 Spark 会不会更快用 Flink 是不是能把时效升级到分钟级。我当时的判断是不要为了一点点性能优势把整套系统的复杂度抬上去。时效要求业务方只要“昨天全天数据”晚上 12 点后开始跑完全来得及。离线 T1 是性价比最高的方案。技术栈匹配团队对 Hive SQL 的熟悉程度远高于 Scala 和 JavaSpark 虽快但需要额外排查内存问题Flink 更不必说状态管理、精确一次语义、窗口机制学习成本高。基于 HDFS 的离线链路生态成熟Flume 采集、Hive 计算、Sqoop 导出、调度平台跑任务每一个环节都有大量公开实践可以抄作业。数据量级也没到非用 Spark 不可的程度。3 亿条日访问日志Hive 跑聚合任务MapReduce 引擎下大概 40 到 60 分钟能跑完如果用 Tez 引擎还能再快一些。为了快这半小时去迁移 Spark完全不值得。所以最终方案用了 Hadoop 离线全家桶Flume 负责日志采集HDFS 做存储Hive 做清洗和指标计算Oozie 负责任务编排Sqoop 把结果导出到 MySQL最后接报表平台。链路看起来长但每个环节都有现成的轮子可以复用。1.3 组件选型和链路设计整体链路我用文字梳理一遍流程如下Nginx 将访问日志写到本地磁盘日志格式配置成统一的 JSON 或 Key-Value 结构。每台业务服务器部署 Flume Agent用 taildir source 监控日志文件将新写入的数据发送到采集集群的 Avro source。Flume 汇聚层将数据下沉到 HDFS按日期和小时拆分目录写入时直接压缩成 gzip 文件。次日由调度平台触发 Hive 脚本先加载原始日志到 ODS 层外部表再清洗过滤得到 DWD 层明细表然后按维度汇总到 DWS 层最后生成 ADS 层业务指标表。Sqoop 将 ADS 层结果表导出到 MySQL。报表平台或邮件报表直接查询 MySQL给业务方展示。这套链路里最容易被忽略的是 Flume 的 sink 端小文件控制。Flume 默认写 HDFS 会按你设置的滚动时间生成文件如果设成 60 秒滚动一次一晚上下来会出现几万个小文件后面跑 Hive 时 NameNode 压力巨大。这个问题后面第 5 节细讲。选型上还有一个细节我们最初用 Hive 默认的 MapReduce 引擎跑 3 亿条日志的 UV 去重要 50 分钟。后来把执行引擎切成 Tez相同任务压到了 30 分钟左右改动成本很低。所以如果你在用 Hive 3.x强烈建议检查一下hive.execution.engine配置别死守 MapReduce。2. 日志采集与预处理数据是地基2.1 日志格式要注意的坑Web 服务器日志最常见的坑就是“字段不完整”和“格式五花八门”。我们接手时发现 Nginx 日志里有几条明显缺字段-代表空值、URL 里带着无法解码的乱码、user_agent 字段里混着换行符直接切分会导致字段错位。所以第一步是让 Nginx 输出为统一 JSON 格式。我们在 Nginx 配置里用log_format将remote_addr、time_local、request、status、body_bytes_sent、http_referer、http_user_agent、request_time等字段拼成 JSON。这样做比用默认的combined格式好在解析时只需要 JSON 解析不用做空格切分字段缺失时也能保留空字符串而不是直接错位。但 JSON 也有新坑业务方在 URL 上带一些特殊字符例如引号、反斜杠会导致生成的 JSON 不合法需要在 Nginx 侧把特殊字符做转义。我们线上就遇到过用户搜索词里带双引号直接把整行日志干成脏数据的情况。除了格式字段补充也要在源头考虑。原始日志没有省份信息只能拿到 IP所以 ETL 阶段要用 IP 库把 IP 解析成省份、城市user_agent 里有浏览器和系统信息需要用解析库拆出browser、os、device_type字段。这些操作放在清洗阶段做不放在采集端因为 Flume 不擅长复杂计算。2.2 Flume 采集配置实战采集端用的是 Flume 1.9核心诉求是不丢数据、不产生大量小文件、能断点续采。这里我给出一份常用的采集 Agent 配置大家可以直接参考# 业务服务器上的 Flume Agenttaildir avro sink a1.sources r1 a1.channels c1 a1.sinks k1 a1.sources.r1.type TAILDIR a1.sources.r1.positionFile /data/flume/taildir_position.json a1.sources.r1.filegroups f1 a1.sources.r1.filegroups.f1 /data/nginx/logs/access.log a1.sources.r1.fileHeader true a1.channels.c1.type file a1.channels.c1.checkpointDir /data/flume/checkpoint a1.channels.c1.dataDirs /data/flume/data a1.sinks.k1.type avro a1.sinks.k1.hostname flume-agg-01 a1.sinks.k1.port 4141 a1.sources.r1.channels c1 a1.sinks.k1.channel c1汇聚层 Agent 的 sink 写 HDFS关键配置如下# 汇聚层 Flume Agentavro source hdfs sink a2.sources r1 a2.channels c1 a2.sinks k1 a2.sources.r1.type avro a2.sources.r1.bind 0.0.0.0 a2.sources.r1.port 4141 a2.channels.c1.type file a2.sinks.k1.type hdfs a2.sinks.k1.hdfs.path /warehouse/ods/raw_log/date%Y-%m-%d/hour%H a2.sinks.k1.hdfs.filePrefix log a2.sinks.k1.hdfs.fileType CompressedStream a2.sinks.k1.hdfs.codec gzip a2.sinks.k1.hdfs.rollInterval 300 a2.sinks.k1.hdfs.rollSize 268435456 a2.sinks.k1.hdfs.rollCount 0 a2.sinks.k1.hdfs.batchSize 2000这段配置里有几个点很容易踩坑rollCount0表示不按事件条数滚动文件只按时间和大小。rollSize是每个文件达到 256MB 就滚动。这个值要配合后续 Hive 读取不是越大越好我们的经验是 128MB 到 512MB 都行。batchSize控制批量写 HDFS 的事务大小太大会增加内存压力2000 是比较稳妥的值。Flume 埋点重启后为什么不会丢数据依赖 taildir source 维护的 position 文件它把每个文件读取的偏移量记录下来了Agent 重启后从偏移量位置继续读。但要注意 positionFile 所在目录的权限否则 Flume 起不来数据会延迟采集到早上的调度时点还没采完。2.3 清洗与富化ETL 该做哪些事原始日志落 HDFS 后接下来是以 Hive SQL 为主做清洗和富化。这个阶段目标很明确过滤无效数据去掉状态码在 400 以上的请求去掉以.js、.css、.png、.jpg、.ico结尾的静态资源请求去掉明显是爬虫的 user_agent。字段标准化把时间字段从字符串转成 timestamp把 URL 解析出 path 和 query把 IP 解析成省份。会话标识生成为每条访问记录生成一个可聚合的session_id后面算跳出率和平均访问时长就靠它。清洗阶段的 HQL 是整套系统里最长最繁琐的因为你要处理各种脏数据。比如 IP 字段出现-时间字段出现nulluser_agent 里出现半个中文字符。我的建议是清洗分两层第一层做硬过滤直接丢脏数据第二层做字段弥补能解析的解析不能解析的填unknown不要因为一条脏数据让整个任务失败。这一层 HQL 示例大致长这样INSERT OVERWRITE TABLE dwd_access_log PARTITION (dt ${dt}) SELECT remote_addr, parse_url(request_url, PATH) AS request_path, parse_url(request_url, QUERY) AS request_query, status, body_bytes_sent, from_unixtime(unix_timestamp(time_local, yyyy-MM-dd HH:mm:ss)) AS access_time, ip_to_province(remote_addr) AS province, parse_user_agent(user_agent, browser) AS browser, parse_user_agent(user_agent, os) AS os FROM ods_raw_log WHERE dt ${dt} AND status 400 AND request_path NOT RLIKE \\.(js|css|png|jpg|ico|svg|woff|woff2)$ AND remote_addr ! -;有人会问为什么不直接对 ODS 表做聚合非要清洗一层明细如果直接把原始日志全量保存每天重复跑聚合时耗时会越来越长而且明细层可以反复复用后面加一个新指标只需要从明细层重新聚合一次不需要再解析一遍原始字符串。砍掉哪一层的干净明细后面新增需求时哭的就是你。3. Hive 数仓建模与指标计算整套系统的重头3.1 数仓分层别一张表打天下很多入门同学做日志分析时只建一张表所有字段全塞进去然后直接在原始表上跑count(distinct ...)。小数据量感觉没问题一到亿级数据就原形毕露跑一次要一小时而且团队里每个人都写自己的临时 SQL口径完全对不上。这套系统我按标准数仓四层模型来搭建ODS 层ods_raw_log只做原始日志落地与日志源字段一一对应保留原始数据用于回溯和排查。DWD 层dwd_access_log经过清洗、过滤、维度补全、会话标识生成的明细事实表粒度是一条访问记录。DWS 层dws_flow_stats按业务维度预先聚合比如按日期 省份 浏览器统计 PV/UV/会话数。ADS 层ads_flow_report面向业务方的最终结果表按天存储核心指标供报表查询。这样分层最大的好处是基础数据只清洗一次所有人都从 DWD 层读取指标口径收敛在 DWS 层每次加指标只影响新增字段不动老逻辑。我把“分层”理解为“把做饭和吃饭分开”ODS 是菜市场买回来的菜DWD 是洗好切好的菜DWS 是半成品ADS 是直接上桌的成品。3.2 核心表结构与分区策略ODS 表我建成了 Hive 外部表数据由 Flume 直接写到表目录下表本身没做复杂压缩转换CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log ( remote_addr STRING, time_local STRING, request_url STRING, status INT, body_bytes_sent BIGINT, http_referer STRING, user_agent STRING, request_time FLOAT ) PARTITIONED BY (dt STRING, hour STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY \t STORED AS TEXTFILE LOCATION /warehouse/ods/raw_log;注意外部表这个细节。如果 Hive 表是内部表一个误操作DROP TABLE会把 HDFS 上的原始日志全删掉找不回来外部表删表只删元数据数据文件还在 HDFS 上这是日志数据最基本的保命手段。DWD 层建议用列式存储加压缩CREATE TABLE IF NOT EXISTS dwd_access_log ( remote_addr STRING, province STRING, browser STRING, os STRING, request_path STRING, request_query STRING, status INT, body_bytes_sent BIGINT, access_time TIMESTAMP, session_id STRING ) PARTITIONED BY (dt STRING) STORED AS PARQUET TBLPROPERTIES (parquet.compressionSNAPPY);使用 Parquet 后的体积只有原始文本的 1/4 左右而且列式存储在按省份、浏览器做聚合时更快。为什么分区只保留dt而不是dt hour因为清洗后是按天分析小时分区只会导致大量小文件没必要保留小时粒度。3.3 PV/UV/跳出率等指标的计算口径与 SQL指标计算是整个系统的核心也是口径最容易打架的地方。我贴几个关键指标的 HQL 片段和边界条件说明。PV是最简单的SELECT dt, COUNT(*) AS pv FROM dwd_access_log WHERE dt ${dt} GROUP BY dt;UV如果直接用COUNT(DISTINCT remote_addr)在亿级数据量下很容易因为数据倾斜变慢。优化思路是先做一层子查询去重再 countSELECT dt, COUNT(*) AS uv FROM ( SELECT remote_addr FROM dwd_access_log WHERE dt ${dt} GROUP BY remote_addr ) t;这里要注意“UV 按 IP 口径”和“UV 按设备口径”是有本质差别的。如果业务方要求按 cookie 或用户 ID 去重DWD 层必须保留对应字段HQL 改 group by 字段即可。会话划分是跳出率和平均访问时长的基础。我们用的标准是同一个remote_addr或 cookie下相邻两条访问记录时间差超过 30 分钟算作一个新的会话。在 Hive 里可以用窗口函数LAG实现SELECT remote_addr, access_time, session_id FROM ( SELECT remote_addr, access_time, CASE WHEN LAG(access_time) OVER ( PARTITION BY remote_addr ORDER BY access_time ) IS NULL OR UNIX_TIMESTAMP(access_time) - UNIX_TIMESTAMP( LAG(access_time) OVER ( PARTITION BY remote_addr ORDER BY access_time ) ) 1800 THEN CONCAT(remote_addr, _, ROW_NUMBER() OVER ( PARTITION BY remote_addr ORDER BY access_time )) ELSE NULL END AS session_id FROM dwd_access_log WHERE dt ${dt} ) t;这段 SQL 生成的session_id会在下一次 CASE 判断时通过上一个值带下去所以必须再做一次LAST_VALUE或“向前填充”补齐 session_id。更标准的分会话方案是把数据导出到 Spark 或写 UDF 去做但在不引新组件的前提下Hive 窗口函数也能完成只是脚本会稍长。跳出率的定义是“跳出会话数 / 总会话数”。一个会话内如果只有一次页面浏览就算跳出SELECT dt, SUM(IF(session_cnt 1, 1, 0)) / COUNT(session_id) AS bounce_rate FROM ( SELECT session_id, COUNT(*) AS session_cnt FROM dwd_access_log WHERE dt ${dt} GROUP BY session_id ) t;平均访问时长同理先算出每个会话的MAX(access_time) - MIN(access_time)再求均值。需要特别说的是指标 SQL 写出来不是终点跑出来的数字要和业务方预期核对。我记得第一版跳出率脚本跑出来 68%业务方说“不可能这么高”后来一查是测试环境的一批爬虫只抓首页就离开把跳出率拉高了。过滤爬虫口 HQL 加了一个白名单判断再跑数值才回到 35% 左右。指标异常很多时候不是代码错而是口径和现实没对齐。4. 调度、导出与报表展示把离线链路串起来4.1 为什么不用 crontab 硬顶离线链路有三个任务阶段Flume 落 HDFS、Hive 清洗聚合、Sqoop 导出。如果只用 crontab每个任务之间的依赖只能靠“时间估算”去硬等例如设定 2 点跑清洗、6 点跑导出。集群一忙计算延迟 20 分钟清洗任务还没结束导出任务已经把空结果推到 MySQL 了。所以调度这么做用一个工作流引擎把三个环节编排成 DAG每个节点执行具体的脚本节点之间做依赖检查上游成功才触发下游失败自动重试最终任务全跑完可以发一条日报“数据已更新”。我们用过 Oozie后来切到了 Azkaban。简单对比一下能力OozieAzkaban配置方式XML复杂properties直观任务依赖支持 DAG支持 DAG失败重跑支持且灵活支持指定节点重跑使用体验学习成本高上手快如果你不是为了兼容 Hadoop 老版本生态我更推荐 Azkaban。它用纯 properties 就能定义任务还能在 Web UI 上直接点击失败节点重跑省去了频繁登录服务器敲命令的麻烦。调度脚本的核心是一个 Shell 包装传dt参数给 HQL并保证每个任务幂等。幂等是什么意思同一份数据重跑多次结果一致不会因为重试导致数据翻倍。最简单的方式是每个 HQL 开头都执行INSERT OVERWRITE指定dt分区这样重跑就是覆盖写不会留脏数据。4.2 Sqoop 导出细节指标算完后要推给业务看Hive 表直接查询不现实所以用 Sqoop 把 ADS 层表导出到 MySQLsqoop export \ --connect jdbc:mysql://mysql-host:3306/analytics \ --username reader --password xxx \ --table ads_flow_report \ --export-dir /warehouse/ads/ads_flow_report/dt${dt} \ --input-fields-terminated-by \001 \ --update-mode allowinsert \ --update-key dt \ --num-mappers 4这里有两个关键点并行度num-mappers控制并行导出线程数不要盲目调到 30。MySQL 写入能力有限并行太高会直接打爆数据库连接我们一般设 4 到 8 个。幂等update-mode allowinsert配合update-key dt让 Sqoop 对主键冲突的行执行 update不冲突则 insert这样重跑同样日期的导出不会导致报表出现两行相同日期数据。还有一个容易忽略的问题Hive 导出到 MySQL 时字段分隔符要统一。Hive 表如果设置了自定义分隔符比如\001Sqoop 必须要加上--input-fields-terminated-by \001否则 sqoop 默认按逗号切分所有字段都会错位报表里数字全变成 0。4.3 报表侧优化与数据校验指标表导出到 MySQL 后报表平台直接查询这张表。但 MySQL 单表数据量过大时查询会变慢所以要在报表表上建索引按dt province browser建联合索引这样业务方按天筛选时能快速出数。同时报表侧不是只看数字我们对每天的结果做数据质量校验当日 PV 和前一天对比波动超过 30% 自动告警。当日 UV 和前一天对比波动超过 30% 自动告警。检查 MySQL 报表表当天是否有数据、是否有重复行。这些校验用 Shell 脚本 定时任务就能完成不用引入复杂的监控系统。数据团队接到告警后第一时间检查 Nginx 日志是否有断流、Flume 是否有积压、Hive 任务是否有重试。5. 离线系统常见问题与排查实录5.1 小文件堆积离线任务的第一杀手我们系统刚上线时跑了一周之后发现 Hive 任务越跑越慢NameNode 的 RPC 请求经常超时。一查 HDFS 目录Flume 按小时分区每个分区下有几百个几 KB 到几十 MB 的小文件。因为配置里rollInterval300每 5 分钟就滚动一次一个小时内 12 个文件一天 288 个文件一周 2000 多个。小文件的危害不只是占 NameNode 内存Hive 读取时要同时打开几千个文件Map Task 数量暴涨集群调度和计算成本跟着翻倍。解决思路有两层第一层在源头控制把 Flume 的rollSize调大到 256MBrollInterval调大到 30 分钟让文件尽量大、数量尽量少。第二层在周期上兜底每天调度一个文件合并任务对当天的分区做INSERT OVERWRITE重新写成少量大文件INSERT OVERWRITE TABLE ods_raw_log PARTITION (dt ${dt}) SELECT * FROM ods_raw_log WHERE dt ${dt};注意这一步也会触发一次全分区重读所以只对时间超过一天的历史分区做定期合并比如一周合并一次。千万不要对当前正在写入的分区执行否则新采到的日志会被覆盖掉。5.2 数据倾斜按天聚合时最容易踩COUNT(DISTINCT remote_addr)最大的问题是会把相同 IP 的数据分到同一个 Reduce热门 IP 可能携带了上百万条记录单 Reduce 跑到内存溢出。我们专门做过一次数据倾斜排查发现某个爬虫 IP 在一天内产生了近 200 万条请求导致 UV 任务卡了 40 分钟没跑完。解决方式最常用的是“两阶段聚合”先打随机前缀分两次 group by。第一层把相同 key 打散到不同 Reduce第二层去掉前缀再聚合。在 Hive 里可以写成嵌套子查询SELECT dt, COUNT(*) AS uv FROM ( SELECT tmp.remote_addr FROM ( SELECT CONCAT(CAST(RAND() * 10 AS INT), _, remote_addr) AS diy_key, remote_addr FROM dwd_access_log WHERE dt ${dt} GROUP BY CONCAT(CAST(RAND() * 10 AS INT), _, remote_addr), remote_addr ) tmp GROUP BY tmp.remote_addr ) t;另外Hive 3.x 自带hive.groupby.skewindatatrue配置可以在不做手动加盐的前提下自动分桶聚合但手动加盐对倾斜场景更可控。加了这层优化之后UV 任务从卡死状态降到 15 分钟内跑完。5.3 重复数据与回溯重跑离线系统最怕的就是“跑错了重跑一遍数据不对”。我们遇到过一次调度平台因为网络闪断重跑了当天的清洗任务但 Sqoop 导出节点没有一起重跑导致 MySQL 里缺当天数据后来又手动补跑导出结果因为update-mode allowinsert配置没加插入了一行重复数据报表里出现两个相同 dt 的记录。这个问题的根治思路是让整条链路的每个节点都具备幂等性。Hive 脚本一律用INSERT OVERWRITESqoop 导出必须带--update-key。这样无论失败重跑多少遍结果都不会叠加。数据回溯时也要注意如果业务方要求重新计算过去 7 天数据调度脚本要把dt参数改成循环传入逐天执行不要一口气写一个跑 7 天数据的 SQL。逐天跑的好处是失败后只需要重跑失败那天的数据而不是整个区间推倒重来。5.4 任务超时与资源分配Hive 离线任务在集群资源紧张时很容易出现“任务排队半小时Map 阶段跑 40 分钟Reduce 阶段再跑 20 分钟”的情况。尤其是多张业务表同时跑日任务时资源队列抢占非常明显。我们当时的处理是两个方面给不同优先级任务分配独立 YARN 队列核心日报任务调度到高优先级队列临时分析任务到低优先级队列避免相互抢占。设置合理的 Hive 容器参数例如set hive.tez.container.size4096; set hive.tez.java.opts-Xmx3072m; set mapreduce.map.memory.mb4096; set mapreduce.reduce.memory.mb4096;注意容器内存不能超过 YARN 单节点可用内存不然任务会直接失败。先跑一次小分区数据用yarn logs查看实际使用的 Container 内存再微调参数比随便拍脑袋设一个大值靠谱得多。5.5 数据质量校验没有监控就不算完成系统上线前我花了不少时间写数据质量校验。最基础的三道检查数据完整性ODS 和 DWD 层当天分区是否存在记录数是否为 0。波动监控核心指标 PV/UV 的环比波动是否超过 30%。超过就触发告警人工排查是业务活动、流量异常还是采集故障。一致性校验DWS 和 ADS 的 PV 相加是否等于 ODS 明细总量避免聚合时漏数据。这套脚本放在每日任务最后如果校验失败直接给相关人员发报警并阻止报表导出。没有监控的离线系统就是走钢丝数据错了业务方已经看了一上午报表再回去解释就非常被动。最后再分享一个小技巧整个项目踩坑最多的地方其实集中在数据口径和幂等设计。很多同学搭离线系统时觉得“能跑就行”但真正上线跑一段时间就会明白离线链路的核心不是某个组件用得有多熟而是把每个环节都做成可重跑、可校验、不重复。哪怕你只是给一个小站点搭日志分析也建议从第一天就做好分层和幂等。我个人实际操作中的习惯是每个 Hive 脚本顶部都有一段统一的“参数校验”脚本判断传入的dt是否合法、要计算的分区是否存在出现问题直接失败退出避免在错误参数下跑出全表脏数据。这套系统后来还接入了更多业务表加了主题宽表也在尝试往实时方向扩展。但说起来离线 Hadoop 这套 T1 链路永远是数据分析的地基把它吃透后面学什么新框架都不慌。希望这篇复盘能帮你少踩几个坑。