Olist 项目不是简单把几个 CSV 读进 Spark 再画图，而是一条湖仓链路的工程练习：业务机采集、HDFS 原始层、ODS/DWD/DWS/ADS 分层、Hive Metastore 注册、数据质量门禁、Kafka 重放设计、Spark Structured Streaming、图分析与 Doris 服务层设计。

我把它当成一项需要留下工程证据的毕业设计来做：每一层都要有输入输出、验证口径和失败复盘，而不是只展示最终大屏。下面会区分已经跑通的离线验证与还依赖 Kafka broker 的端到端实验，避免把设计稿写成生产结论。

01集群与主链路

项目跑在一组 Hadoop 集群上，业务机和大数据节点分开：

节点角色
main业务数据落地、Flume source、MySQL、Redis、Docker
hadoop01NameNode、Hive Metastore、Spark driver、Doris FE/BE
hadoop02ResourceManager
hadoop03SecondaryNameNode

主链路是：

text
main:/bigdata/soft/olist/raw
  -> Flume
  -> hdfs://hadoop01:8020/data/olist/raw
  -> ODS Parquet
  -> DWD fact/dim
  -> DWS monthly aggregates
  -> quality / graph / streaming outputs
  -> Hive Metastore
  -> Doris serving / ADS views

Spark Scala 项目通过 assembly JAR 交付，核心 JAR 为 batch-etl-assembly-0.1.0.jar。本地验证可以用 local[2]，最终保留 spark-submit 入口，方便提交到集群。

02ODS：CSV 不是天然干净的

Olist 原始数据里最典型的问题来自 olist_order_reviews_dataset.csv：评论文本可能包含带引号的换行。最开始 Flume 使用 line-based deserialization，结果把一条评论拆坏，下游出现了荒诞的 review score。

修复方式：

  1. Flume 离线采集改用 BlobDeserializer。
  2. Spark CSV 读取开启 multiLine=true、quote="\""、escape="\""。
  3. 重新采集 review 原始数据，并重跑 ODS、DWD、DWS、Quality。

修复后 review score 分布恢复正常，非法评分被清空。这件事提醒我：数据仓库的第一关不是建模，而是把文件格式读对。

03DWD：修正事实表粒度

DWD 层最重要的设计是把不同事实粒度拆开：

表粒度
fact_order_item一行一个订单明细 item
fact_order_payment一行一个订单聚合后的支付信息
fact_order_review一行一个订单聚合后的评论信息

早期直接把 order、item、payment、review 做大宽表 join，多对多关系会把 item 行数撑起来——同一笔订单的多次支付和一条评论会被交叉复制，join 出来的行数明显高于真实的 item 数。

最终修正为：先在 payment/review 侧按订单聚合，再与 item fact 做下游关联。这样既保住 item 明细粒度，也避免 GMV、订单数和质量规则被 join 膨胀污染。

04DWS 与 GMV 对账

DWS 输出三类月度汇总：

表粒度
dws_user_month用户 × 月
dws_category_month品类 × 月
dws_region_month区域 × 月

我用 GMV 做核心对账指标：fact 表算一次，三张 DWS 汇总表各算一次，四条口径的结果完全相等。三个 DWS 主题的 GMV 都能回到事实表，说明分层没有在汇总阶段丢数据或重复计算。这个项目里，GMV 对账比多做几个图表更重要。

05质量门禁：从文档规则变成可执行规则

质量模块用 Deequ 与 DataFrame 规则组合，单表规则和跨表一致性规则分开写，规则本身和它们的定义一起存表，而不是散在代码里。

规则覆盖：

  1. 完整性：主键、金额、关键维度不可为空。
  2. 唯一性：订单 item、payment、review 的业务键唯一。
  3. 合法性：金额非负、订单状态枚举、评分在合法区间。
  4. 一致性：fact GMV 与 user/category/region DWS GMV 对齐。
  5. 膨胀检测：修正后的事实粒度不允许明显偏离。
  6. 漂移检测：有 baseline 时检查 GMV 和订单量波动。

Baseline 批次的结果是全部规则通过、没有 BLOCK。

随后我做了异常注入实验：空主键、重复 item、负金额、非法评分、GMV 漂移。注入进去的错误都被对应规则拦下，触发 BLOCK 的那几条规则命中预期，Precision、Recall、F1 没有漏项。这个实验比“我写了质量规则”更有说服力，因为它验证了规则真的能拦住预期错误。

06Kafka 重放与批流对账设计

实时部分不是为了包装成生产级实时湖仓，而是设计一条可解释的批流对账链路：

text
Olist 历史 CSV
  -> Kafka JSON events
  -> Spark Structured Streaming
  -> per-minute GMV aggregates
  -> replay_manifest
  -> batch-stream consistency check

代码和设计文档已经明确了两个容易忽略的约束：重放要按 event_time + event_id 排序，末尾需要用 watermark event 推进窗口关闭；批侧只比较 replay_manifest 中真正进入重放范围的 (order_id, order_item_id)，不能拿全量 batch 与部分 streaming 直接对比。

目前本地已经完成了离线链路、质量规则和相关测试的验证。Kafka broker 依赖的 producer -> broker -> Structured Streaming -> 对账这条端到端运行，在当前环境没有实际拉起 broker，因此我不把它写成已经跑出的吞吐或一致性结论。replay_manifest、事件时间排序和对账脚本是下一次接通运行环境后要验证的基线。

07图分析与 Doris 服务层

图分析把 customer、product、seller、category 建成交易图，一次运行会输出节点规模、边规模、最大连通分量占比和卖家特征数这几项指标，结果写回 graph_metrics 与 seller_graph_features。

PageRank 和连通分量更像实验层，用来解释卖家、商品和品类之间的网络结构，而不是直接替代经营指标。

Doris 服务层的表设计只选择适合查询和展示的结果，例如 dws_user_month、dws_category_month、quality_batch_result、graph_metrics、seller_graph_features、batch_stream_consistency。ADS view 再面向前端组织 KPI、月度 GMV、品类 Top、区域 Top、质量状态、图指标和批流对账口径。当前优先保证离线结果和表设计可追溯；Doris smoke check 与查询性能实验仍在后续清单里。

08项目复盘

这个项目让我最有收获的不是某一个技术点，而是“每层都要能解释为什么存在”：

  1. ODS 要解决原始文件读取和类型问题。
  2. DWD 要守住事实粒度。
  3. DWS 要能回到事实表对账。
  4. Quality 要能通过异常注入验证。
  5. Streaming 要有 manifest 才能在接通 broker 后公平比较。
  6. Doris 服务层要服务查询，不是盲目同步所有表。

后续如果继续做，我会优先补齐 Kafka broker 端到端重放、Doris smoke check、查询性能实验和前端 API 的真实结果绑定，而不是再堆更多中间层。