大数据离线处理流程详解

在电商公司做数据运营的小李，每天早上第一件事就是查看昨天的销售报表。这些报表不是实时生成的，而是凌晨从海量用户行为日志中统计得出。这类任务背后依赖的就是大数据离线处理流程。

什么是离线处理

离线处理指的是对已经存储的历史数据进行批量分析的过程。它不追求即时响应，更关注吞吐量和处理复杂任务的能力。比如月底财务结算、用户画像更新、月度活跃趋势分析，都是典型的离线场景。

典型流程拆解

一个完整的离线处理链路通常包含五个环节：数据采集、数据存储、资源调度、计算执行和结果输出。

1. 数据采集

系统日志、数据库变更、用户点击流等原始数据需要先汇聚到中心化存储。常用工具包括 Flume 收集日志，Kafka 作为缓冲队列。例如：

agent.sources = r1
agent.sinks = k1
agent.channels = c1

agent.sources.r1.type = exec
agent.sources.r1.command = tail -F /var/log/app.log
agent.sinks.k1.type = org.apache.flume.sink.kafka.KafkaSink
agent.sinks.k1.topic = log_topic

2. 数据存储

采集来的数据会写入分布式文件系统，最常见的是 HDFS。结构化数据可能导入 Hive 数仓，方便后续 SQL 查询。建表语句就像这样：

CREATE TABLE user_behavior (
  user_id STRING,
  action_type STRING,
  page_url STRING,
  ts BIGINT
) PARTITIONED BY (dt STRING)
STORED AS PARQUET
LOCATION '/data/user_behavior';

3. 资源调度

每天凌晨两点要跑一批任务，谁先谁后？YARN 或 Kubernetes 负责分配 CPU 和内存资源。Airflow 这类工具则用来编排任务依赖关系。比如必须等日志入库完成后才开始统计 UV。

4. 计算引擎

真正干活的是计算框架。MapReduce 适合简单批处理，Spark 因为内存计算更快，成为主流选择。一段统计每日订单总额的 Spark 代码可能是：

val orders = spark.read.parquet("/data/orders/dt=20240405")
val total = orders.filter(_.amount > 0).agg(sum("amount"))
total.write.mode("overwrite").csv("/report/daily_revenue/20240405")

5. 结果落地

计算完的结果通常写回 Hive 表，或者导出到 MySQL 供 BI 系统展示。有些还会推送到邮件列表，像小李收到的那份销售日报，就是这么来的。

为什么不用实时处理

实时系统虽然快，但成本高、逻辑复杂。一次全量用户标签更新可能涉及上亿条记录关联，用离线方式跑一整夜也没问题，反而更稳定可靠。就像做饭，爆炒讲究速度，炖汤则需要时间沉淀。

常见挑战与应对

任务失败重试机制很重要。如果某天数据质量差导致计算中断，要有自动告警通知负责人。分区缺失、字段空值这些问题，得在 ETL 阶段就做好清洗规则。

随着数据量增长，原来两小时跑完的任务现在要六小时，就得考虑优化 SQL 写法、调整并行度，甚至升级集群配置。这不是一劳永逸的事，而是持续调优的过程。

大数据离线处理流程详解 使用技巧与常见问题解析