★ FEATURED ARTICLE “基于 MaxCompute 的实时数据处理实践”——看到这个主题可能不少人第一反应是MaxCompute 不是跑离线批处理的吗怎么和“实时”搭上边我第一次接到这个需求时也是同样的疑问。直到后来真正把链路搭起来才意识到所谓的“实时数据处理”在绝大多数业务场景里并不是让 MaxCompute 去抢流计算引擎的活而是让整个链路从“采集 → 计算 → 存储 → 服务”形成一个端到端的准实时闭环MaxCompute 在这个闭环里承担的是大规模存储、批量合并和历史数据服务这一层关键底座。这篇文章会把我在实际项目中搭建这套链路时的整体设计、核心配置、参数调优和真实踩坑记录完整梳理出来。如果你是正在做实时数仓、或者想把实时计算结果和离线大数据存储打通的数据工程师这篇文章应该能帮你少走不少弯路。我不写教科书式的理论只讲能直接落地的方案和为什么这么做。1. 先理清思路MaxCompute 在实时链路里到底该站在哪1.1 为什么“MaxCompute 只适合离线”这个印象该更新了MaxCompute 本质上是一个大规模并行计算的数据仓库服务最擅长的是跑批任务、处理海量历史数据。但这并不代表它只能在 T1 的场景里出现。我见过很多团队上了 Flink 之后实时结果就算出来了结果落库的时候犯了难业务方要的是能按任意维度、任意时间范围去查的明细数据而不是简单的 Redis 里存几个聚合值。这时候你总不能拿 Redis 存上亿条明细吧也不能把 HBase 里的数据再同步一份到关系型数据库里做分析。MaxCompute 真正擅长的恰好就是这个存得下海量明细、能跑复杂的分析 SQL、支持分区裁剪和列存优化而且对于准实时场景它完全可以通过增量分区、周期合并的方式来承接。也就是说实时链路里的“实时”这个词不应该被狭隘地理解成“从事件发生到可见只有几秒钟”在大多数业务场景里分钟级或者十分钟级的可见延迟配合海量离线历史数据的统一分析能力才是更实际的需求。把 MaxCompute 放回这条链路里它的定位就不是“实时计算引擎”而是“实时与离线统一的数据底座”。前端 Flink 做流式清洗和指标计算结果落到 MaxCompute 的增量分区里再由 MaxCompute 的定时任务把增量合并进主表最终用统一的 SQL 接口对外提供查询。这套思路既能拿到准实时的数据新鲜度又不用把体量庞大的数仓整体迁移到流式框架上成本和收益之间会有个非常好的平衡。1.2 实时数据落地的四种路线怎么选我先画一下目前主流的实时数据落地路线。这么说吧如果你现在打开任何一个数据团队的架构图大概率跑不掉下面这几种组合。第一种是“纯流式存储”。把数据全部放在消息队列或者内存数据库里比如用 Kafka 做短期窗口、用 Redis 做热点查询。这种方式延迟最低但能承载的数据量和服务能力有限只适合做缓存层或实时看板没法支撑复杂的关联分析而且数据保留周期短一旦超过保留时间就查不到了。第二种是“实时数仓专用引擎”。用专门为实时设计的数据仓库比如某些能支持高并发点查的分析型数据库接收 Flink 或 spark streaming 的输出结果。好处是查询延迟低、体验好缺点是存储成本通常比 MaxCompute 这类批量数仓要高不少而且如果要存几十 TB 甚至上百 TB 的明细费用会很敏感。第三种是“流批混合”。也就是 Flink 做实时统计写进一套存储离线任务每天再跑一套全量结果对账用。这种方式最稳但也是最费人力的因为同一套指标要维护实时和离线两套代码数据对不上永远是说不完的痛。第四种就是我这次要讲的“MaxCompute 为中心实时写入加周期合并”。Flink 负责流式计算把结果按分区粒度写进 MaxCompute 的增量分区MaxCompute 上的周期任务再把增量数据合并进主表。查询端看到的是一张统一的、完整的数据表底层真实数据从增量到合并的过程对业务方完全透明。四种方式没有绝对的好坏我整理一个表格直接对比方案可见延迟存储成本支持数据量运维复杂度适合场景纯流式存储秒级较高中低实时大屏、热点数据实时数仓专用引擎秒级到分钟级高中高中高并发交互式分析流批混合分钟级到 T1中高很高对账严谨、指标众多MaxCompute 实时写入加合并分钟级到小时级低极高中海量明细统一分析1.3 最终采用的方案实时写入加周期合并我参与的这个项目典型场景是业务方每天要看到当天到当前时刻为止的订单、转化、库存流水等明细数据而且要和上个月、去年同比对比意味着必须有完整的历史数据。一开始的方案是走流批混合但很快发现一个问题很多指标实时算一次、离线再算一次两边的口径越来越难以保持一致。而且业务方不关心底层是实时还是离线他们只想要一套数。所以最终敲定的方案就是围绕 MaxCompute 这条线搭一套“准实时链路”。具体工作机制是实时计算引擎负责把原始事件流清洗、去重、关联维表然后以分钟级频率把结果写到 MaxCompute 的增量分区表里。MaxCompute 内部通过周期合并任务把增量分区的数据归并到主分区或者主表对外查询时走的是合并后的统一视图。这套方案的核心优势在于数据只有一套口径只有一套实时和离线之间的边界在做完合并之后就消失了。2. 整体链路怎么搭从数据源到 MaxCompute 的关键设计2.1 一条完整链路的六个环节每个环节别偷懒整体数据链路我拆成了六个环节数据采集、消息缓冲、流式计算、增量落地、周期合并、统一服务。每个环节对应一个清晰职责这样出了问题容易排查也容易做降级。数据采集业务系统的日志、订单变更、埋点数据通过采集组件或者直接 SDK 上报统一进入消息队列。这个环节的关键是消息丢失和重复的问题采集端要能确认消息是否成功到达最好有本地的缓冲机制。消息缓冲这里承载的是削峰填谷的作用。实时数据最怕的就是流量洪峰如果没有缓冲层直接压到下游计算任务Flink 的背压机制会把压力传导到整条链路最终就是处理延迟从毫秒级变成秒级甚至分钟级。我们当时用消息队列做了至少三天的数据保留这样即使下游任务挂了恢复后也能继续消费而不丢数据。流式计算Flink 作业负责解析、清洗、去重、关联维表、计算指标。这一层我建议别贪多把能交给 MaxCompute 的批量计算全交出去流上只做必须实时做、不做就看不到结果的事。增量落地Flink 输出到 MaxCompute 时最常见的模式是写进一个以日期和小时为粒度的增量分区表。这块后面我会专门讲因为这里藏着一个小文件的问题和并发度的问题。周期合并MaxCompute 上起一个定时调度任务每隔一段时间把增量分区数据合并进主表合并的窗口依据业务容忍的延迟来定。比如业务能接受 10 分钟延迟就每 10 分钟跑一次合并。统一服务合并完成后上层应用、报表、数据服务 API 都通过 MaxCompute SQL 或者查询引擎读取最终结果。这套东西对业务方是透明的他们感知不到增量分区的存在。2.2 分层建模和分区设计先想清楚再建表实时链路最怕的是“上来就写表”结果过两周发现表结构撑不住需求又得重跑历史数据。在 MaxCompute 里建表之前我强烈建议你先做分层规划。ODS 层直接对接实时计算产出的清洗结果字段最原始按业务时间分区。DWD 层做明细整合比如把订单明细、支付明细、退款明细合并成一张事实宽表。DWS 层做汇总比如按店铺、商品、小时粒度的累计指标。ADS 层服务应用。分区设计上一个容易踩的坑是“分区字段粒度太小”。比如为了追求实时性有人会把分区粒度设到分钟级结果 MaxCompute 里产生海量小分区。分区文件数超过一定量之后查询扫描的元数据开销就非常可观甚至比数据本身的读取还慢。我们实验下来小时级分区是比较平衡的选择对分钟级延迟的业务场景通过合并任务把延迟控制在 10 到 15 分钟内完全够用。另一个容易忽略的点是时区。数据源如果是带时区的时间戳落地到 MaxCompute 分区时一定要统一转成东八区否则跨天统计会出现数据落到前一天分区的情况。这个坑我们踩过一次凌晨 0 点到 8 点的数据全部跑到前一天分区里去了排查了半天才意识到是时区没有统一。2.3 写入 MaxCompute 的三种通道什么时候用哪种要把实时数据写进 MaxCompute实际可选的方式大致有三种。第一种是用 Tunnel SDK 直接写。Tunnel 是 MaxCompute 的底层数据上传通道通过 PartitionUploadSession 创建上传会话然后把记录批量写入。这个方式最灵活适合在 Flink 的 Sink 里自己控制批量大小和提交时机。缺点是代码开发量会大一些需要对 SDK 的机制比较熟。第二种是用流计算连接器也就是 Flink 官方或者云厂商提供的 MaxCompute Connector。这种方式把 Tunnel 封装好了只要配置表名、分区字段、批量大小等参数就能直接用。好处是省事坏处是连接器版本各异一些隐藏的坑不好排查。第三种是离线同步工具比如定时把实时结果从某个存储同步到 MaxCompute。这种方式适合数据先落到其他系统再定期批量拉取的场景但严格来说绕了一圈延迟会变大不太适合“准实时”的目标。从我的实践来看正式的实时链路里推荐用第二种因为可维护性最好。只有在需要做很复杂的写入控制、比如自定义去重逻辑、动态分区路由时才会考虑用第一种 SDK 方式。第三条路留给非实时场景的补充通道我后面会再提到。3. 核心实现动态分区写入与增量合并3.1 让 Flink 把结果直接写进 MaxCompute 动态分区实时计算结果写 MaxCompute最关键的是“攒批提交”和“动态分区路由”的配合。逐条写肯定不行MaxCompute 毕竟是批量接口单条写入的会话开销非常大。我的经验是控制在每个分区至少攒够一定数据量再提交同时让 Flink Sink 里根据事件时间自动路由到对应的分区字段。用连接器的模式一个典型的配置思路是指定一个分区列比如ds和hh然后 Flink 在处理时把每条记录按时间戳计算出来属于哪一天哪个小时。连接器内部会维护每个分区对应的上传会话一批记录攒够之后统一写入提交。这里有个很重要的点分区上传会话的提交和 Flink 检查点机制要配合好避免任务从 checkpoint 恢复时出现数据重复或丢失。下面是一个基于 DataStream API 的大致代码骨架实际生产环境里还需要封装成可配置的连接器但核心逻辑长这样// Flink Sink 中按分区攒批写入 MaxCompute 的示意逻辑 // 真实环境请结合你选用的 Connector 与 SDK 版本调整 public class McPartitionSink extends RichSinkFunctionRowData { private TableTunnel tunnel; private MapString, UploadSession sessionMap; // 每个分区对应一个上传会话 private int flushIntervalMs; private long lastFlushTime; Override public void invoke(RowData value, Context context) throws Exception { String partition generatePartition(value); // 事件时间 - yyyyMMdd/HH UploadSession session sessionMap.computeIfAbsent(partition, k - tunnel.createUploadSession(project, table, partition)); // 先写入本地缓冲累积到阈值再统一 openRecordWriter 写入 buffer(partition, value); if (shouldFlush(partition)) { flushPartition(partition); } // 周期性兜底防止长时间小流量导致数据迟滞可见 if (System.currentTimeMillis() - lastFlushTime flushIntervalMs) { flushAll(); } } private void flushPartition(String partition) throws Exception { UploadSession session sessionMap.get(partition); try (RecordWriter writer session.openRecordWriter(0)) { for (Record rec : bufferedRecords.get(partition)) { writer.write(rec); } } session.commit(new int[] {0}); // 提交上传会话 sessionMap.remove(partition); bufferedRecords.remove(partition); } }注意上面的代码是表达“按分区维护会话、攒批提交”这个思想。实际开发中如果是用现成的连接器你只需要正确配置动态分区参数以及设置批量写入的条数和时间窗口比如每 10000 条或每 60 秒提交一次。3.2 状态、去重和顺序流上最容易翻车的三个点流计算之所以难不是因为“实时”两个字而是因为数据可能是乱序的、重复的、晚到的。这三个问题如果不在流上处理好写进 MaxCompute 之后后面合并任务看到的数据就是错的而且很难再回头修正。去重是第一关。消息队列是 at least once 语义也就是说一条消息可能会被消费多次。如果业务上对主键有唯一性要求Flink 里必须用状态做去重。我们常用的做法是用KeyedProcessFunction以主键分组后记录每条数据的更新时间只处理比当前状态更新的记录。这里要特别关注状态的过期时间状态设置得太长会撑爆内存设置得太短超过窗口的数据又可能被当成新数据再次写入产生重复。实践中可以根据业务数据的延迟分布设置比如 24 小时到 48 小时的 TTL并且配合 RocksDB 状态后端来控制堆内存的占用。乱序问题是第二关。判断一条数据是不是“晚到了”靠的是事件时间和水位线。Flink 里通过WatermarkStrategy来声明允许乱序的时间范围。比如我们允许 5 分钟内的乱序水位线就设置为当前最大事件时间减 5 分钟。这里需要想清楚一个问题晚到的数据到底是丢弃、重新计算、还是先写入一个补数据通道如果业务能接受少量延迟数据可以在窗口计算时把它丢弃如果不能接受就得让 Flink 在事件时间超过水位线之后仍然输出到旁路流。顺序问题是第三关。对同一主键的记录如果后到的先被处理前面的记录反而更新就会出现数据“倒退”。一个稳妥的方案是在流上对同一主键的更新做排序或者在下游合并 SQL 里通过ROW_NUMBER()按更新时间取最新值后者实现起来更简单而且天然具备幂等性。这也是后面合并脚本的基本逻辑之一。3.3 周期合并不是简单的 overwrite要看数据怎么“变”增量分区落地之后合并任务把增量合入主表。这里最忌讳的思路是直接把增量分区INSERT OVERWRITE到主表因为INSERT OVERWRITE会把目标分区的旧数据整个覆盖掉如果增量数据里只包含发生变化的部分而不是全量数据直接覆盖会导致旧数据丢失。我们用的合并策略是把主表的历史数据加增量分区数据放到一起按业务主键和更新时间去重保留最新的一条然后整体覆盖到目标分区。听起来很绕但执行起来就是一个标准的 SQL。我以订单维度为例写个示例-- 将增量分区数据合并进主表 -- 思路历史全量 当日增量 - 按主键去重取最新 - 覆盖写入主分区 INSERT OVERWRITE TABLE dwd_order_info PARTITION (ds${bizdate}) SELECT order_id, user_id, product_id, amount, status, update_time FROM ( SELECT order_id, user_id, product_id, amount, status, update_time, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY update_time DESC) AS rn FROM ( SELECT order_id, user_id, product_id, amount, status, update_time FROM dwd_order_info WHERE ds ${prev_date} UNION ALL SELECT order_id, user_id, product_id, amount, status, update_time FROM dwd_order_delta WHERE ds ${bizdate} ) t ) tmp WHERE rn 1;这段 SQL 的逻辑核心是ROW_NUMBER()加PARTITION BY去重保证同一订单在历史数据和增量数据里都出现时取更新时间更大的那一条。用这种方式即使实时增量重复推送了几次合并结果也是幂等的——这个特性非常重要因为流任务重启、重跑是常态幂等能让我们在出问题时大胆重算不需要担心产生脏数据。合并任务虽然简单但执行频率会直接影响成本和延迟。执行太频繁一个小时跑几十次全量合并MaxCompute 的计算资源消耗会很大执行太稀疏数据的可见延迟就高。比较好的做法是小时级增量分区每 10 到 15 分钟合并最近一个分区这样既保证了可见性又把计算量控制在可以接受的范围。对数据量特别大的核心表我建议拆两个阶段先做一个轻量的增量汇总把分区的数据量降下来再做全量合并避免每次都扫描几亿条历史数据。4. 参数调优与成本控制跑得快又要花得少4.1 小文件是隐形杀手控制文件数的几个参数实时写入 MaxCompute几乎必然会遇到小文件问题。所谓小文件就是每个分区下面的数据文件特别多、每个文件又特别小。小文件多会带来两个隐患一是查询时扫描文件的元数据开销大明明只有几百万条数据查询却慢得像在扫全表二是合并任务处理这些碎片时会消耗大量计算资源。小文件是怎么产生的很简单如果 Flink 连接器里设置的批量提交阈值太小比如几十条就提交一次或者并行度设得很大而在同一分区里同时写入的并发很高就会形成大量小块文件。我碰过最夸张的一次某个小时的增量分区下挂了 3000 多个小文件跑一次合并任务要花二十多分钟完全不可接受。解决的思路有三个方向。第一适当调大批量提交的阈值。比如单批至少 1 万条如果流量低再配合一个时间阈值比如 60 秒强制提交。第二控制写入并行度。同一个分区的写入并发不要太高宁可让部分并行度空置也不要在写同一个小时分区时起二十个并发。第三通过 MaxCompute 自带的文件合并能力定期做小文件合并。对于已经产生的小文件写一个周期任务把分区数据重写一遍即可但要注意频率别把这个任务又变成另一个成本黑洞。这里我建议一个关键参数范围实际需要根据数据量做微调批量大小设置在 5000 到 20000 条之间强制提交时间设置在 60 秒左右。这样既能保证分钟级延迟又不会疯狂产生小文件。如果你选用的连接器有类似write.bucket.size的配置可以根据目标分区大小反向推算单桶条数原则就是“让每个文件尽量接近 128MB 甚至更高”。4.2 数据倾斜的治理加盐、分桶、剪枝写入阶段的数据倾斜很多团队不太关注直到被坑过才回头看。倾斜的典型表现是Flink 作业整体并行度是 16结果某个 subtask 处理的数据量是其他 subtask 的十倍CPU 跑满了其他 subtask 在空转。整体吞吐上不去背压往上传播最后延迟飙升。这种情况通常是因为分区的路由键选择不当。比如按订单 id 分区而某些大客户的订单量特别大那么对应分桶的数据就多。处理手段也很经典加盐。给分桶键加一个随机后缀把大 key 拆散到多个桶里写入之后再在合并阶段按原 key 聚合。加盐的方式要分层设计不能全局加盐否则查询时就找不回原始 key 的连续性了。另一个容易被忽略的是“时间戳分区倾斜”。流量天然有高峰和低谷凌晨写入量小晚高峰写入量大。如果只按小时分区晚高峰那个小时分区的文件数和数据量会明显偏大。这个问题很难彻底消除但可以做的处理是对高峰时段单独调高并行度或者把这个小时分区的存储格式优化成列式压缩比更高的模式减少后续查询的压力。对 MaxCompute 侧的查询分析也要善用分区裁剪和列裁剪。写 SQL 时永远把分区过滤条件放在最外层比如查某一天的订单明细就只扫当天的分区不要让查询去扫描全表分区。实时链路落地的表如果常用过滤字段没有放在分区里可以考虑建二级索引或者物化视图来优化。关于物化视图有些团队忽略了 MaxCompute 的物化视图能力对于经常要跑相同聚合查询的业务建一个定时刷新的物化视图能显著降低重复计算成本。4.3 账单视角的成本控制存储与计算要怎么取舍实时链路的成本说起来就两块存储费用和计算费用。存储上增量分区会保存一份数据主表又会保存一份全量数据相当于短期内数据有冗余。这不可避免但可以控制生命周期。增量分区在合并完成后它的使命就已经结束了你可以给增量分区设置一个较短的 Lifecycle比如 7 天过期自动清理避免无限累积占用存储费用。计算费用上最有效的控制方式是“避免重复计算”。同一个指标不要在 Flink 算一遍、合并任务里又算一遍、物化视图里再算一遍。我的原则是能在实时计算阶段算好的指标就在实时计算阶段算好合并阶段只做数据归并不再做复杂聚合。如果业务方临时要加一个新指标先评估能不能从现有明细里用 SQL 查出来能查出来就不要额外加一个实时计算任务。还有一点关于生命周期管理MaxCompute 的分区表支持配置生命周期对于 ODS 层的原始日志类数据保留 30 天就够了DWD 层和 DWS 层按业务审计需要保留半年或者一年DWS 以上的汇总数据尽量用压缩率更高的模式存储。否则随着实时链路长期运行存储成本会像滚雪球一样涨上去月底账单出来再优化就晚了。5. 真实环境常见问题与排查实录5.1 现象一任务运行正常但 MaxCompute 分区迟迟不更新这个现象我遇到不止一次。Flink 作业运行状态一切正常Checkpoint 也在正常做但查询 MaxCompute 的分区数据就是看不到新数据。排查下来发现要么是连接器里没有设置定时提交攒批的阈值又设得特别大低流量时段要等很久才凑够一批要么是分区字段配置错误数据写到别的时间分区去了。排查方式很简单去 MaxCompute 表的分区列表里看一眼最新的分区确认有没有新分区生成。如果新分区有生成但查询不到数据再看一下上传会话有没有成功 commit。Tunnel 上传有个特点打开 session 后如果只是写入记录但不提交数据是查询不到的。连接器如果没处理这个细节就会表现成“什么都正常但啥都查不到”的诡异状态。建议的兜底方案在连接器或者自定义 Sink 里强制增加一个时间触发提交比如每 60 秒不管有没有攒够都提交一次。毕竟准实时链路的核心目标是低延迟为了攒批优化而牺牲可见性是本末倒置。5.2 现象二写入越来越慢最终内存溢出Write 慢往往不是 MaxCompute 的问题而是 Flink 侧攒批逻辑出了问题。常见的原因缓冲列表没有上限数据涌入速度大于写出速度导致堆内存持续增长或者 RecordWriter 没有正确关闭连接数被不断打开最终把资源耗光。这里我吃过一次亏。当时的 Sink 实现里每个分区都维护一个上传会话并且把待写入的数据放在了一个没有容量限制的 List 里。高峰流量一上来List 无限膨胀OOM 就来了。后续改成有界缓冲满了就触发 flush同时对每个分区的会话数量做了上限控制问题才解决。另一个容易被忽略的因素是分区数量过多导致 session 数量膨胀。比如动态分区按分钟粒度一个任务同时维护几十个分区上传会话每个会话都有独立的内存缓冲区累加起来非常可观。所以无论从减少小文件的角度还是从降低内存压力的角度控制动态分区的数量都极其重要。5.3 现象三实时报表和离线任务跑出来的结果对不上实时和离线对不上是最让人头大的问题。因为两边跑数的逻辑本来就有差异再叠加乱序数据和重复数据差异会在关键指标上被放大。常见的原因有这么几个。其一实时计算使用处理时间而不是事件时间晚上补录的数据被算进了补录时刻的时段而离线任务按业务发生时间归档自然不一致。其二去重逻辑不同实时链路按主键去重但离线 SQL 用了distinct或者group by结果也会不同。其三合并任务执行过程中正在合并的分区可能同时被实时写入导致查询到半合并状态的数据。针对这些问题我的建议是实时计算统一使用事件时间加水位线这个没得商量对账的基准以离线为准因为离线对于历史数据是重算全量口径更稳。出现不一致时先查增量分区里有没有覆盖那条记录的更新再看合并任务执行的窗口有没有和写入窗口重叠。如果重叠就把合并窗口和写入窗口错开 5 分钟再拉取数据可以规避大部分半合并状态导致的对不上。5.4 常见问题速查表问题可能原因快速排查方向分区不更新攒批阈值过大、未提交 session、分区字段配置错误查最新分区列表确认 commit 日志写入越来越慢Sink 缓冲无界、session 过多、连接未释放检查堆内存和 GC 日志统计 session 数量实时与离线对不上时间口径不一致、去重逻辑不一致对比增量分区和离线结果检查水位线配置小文件过多flush 太频繁、并发过高查看分区文件数统计调大批量阈值合并后数据丢失overwrite 目标分区时未做历史数据合并检查 SQL 是否 UNION ALL 历史分区内存溢出分区上传会话过多、缓冲无上限限制分区数量增加 flush 时间触发数据重复at-least-once 语义、未做去重状态管理检查主键去重逻辑确认状态 TTL 设置我最后再分享两个个人经验。第一个是永远保留一个离线补数通道。实时链路再稳定也会遇到源头数据修正、上游逻辑变更这类事情这时候你不能只依赖实时任务去补历史数据一定要能通过离线任务重刷一张表并切到线上。第二个是链路里每一层都要有“幂等”意识。写入可以重复、合并可以重跑、查询结果不能变这个原则能让你在处理故障时有底气不用怕把数据弄坏。这套基于 MaxCompute 的实时数据处理链路在我实际项目中跑了很长时间稳定性和成本都达到了预期的效果。如果你也在搭类似链路希望这篇实践能给你提供一些参考。
阅读完成 · 觉得有帮助？
