StarRocks实战——携程酒店实时数仓

一、实时数仓

二、实时数仓架构介绍

2.1 Lambda架构

2.2 Kappa架构

三、携程酒店实时数仓架构

3.1 架构选型

3.2 实时计算引擎选型

3.3 OLAP选型

四、携程酒店实时订单

4.1 数据源

4.2 ETL数据处理

4.3 应用效果

4.4 总结

原文大佬的这篇实时数仓建设案例有借鉴意义，属于数据治理范畴，这里直接摘抄下来用作学习和知识沉淀。

一、实时数仓

当前，企业对于数据实时性的需求越来越迫切，因此需要实时数仓来满足这些需求。传统的离线数仓的数据时效性为T+1,并且调度频率以天为单位，无法支持实时场景的数据需求，即使将调度频率设置为小时，也仅能解决部分时效性要求低的场景，对于时效性要求较高的场景仍然无法优雅地支撑。因此，实时数据使用的问题必须得到有效解决。实时数仓主要用于解决传统数仓数据时效性较低的问题，通常会用实时的OLAP分析，实时数据看板、业务指标实时监控等场景。

二、实时数仓架构介绍

2.1 Lambda架构

Lambda架构将数据分为实时数据和离线数据，并分别使用流式计算引擎（例如Flink 或者 SparkStreaming）和批量计算引擎（例如 Hive、Spark）对数据进行计算，然后，将计算结果存储在不同的存储引擎上，并对外提供数据服务。

2.2 Kappa架构

Kappa架构将所有数据源的数据转换成流式数据，并将计算统一到流式计算引擎上，相比Lambda架构， Kappa 架构省去了离线数据流程，使得流程变得更加简单。Kappa 架构之所以流行，主要是因为kafka不仅可以作为消息队列使用，还可以保存更长时间的历史数据，以替代Lambda架构中的批处理层数据仓库。流处理引擎以更早的时间作为起点开始消费，起到了批处理的作用。

三、携程酒店实时数仓架构

3.1 架构选型

采用的是Lambda+OLAP 变体架构。Lambda架构具有灵活性高、容错性高、成熟度高和迁移成本低的优点，但是实时数据和离线数据需要分别使用两套代码。

OLAP变体架构：将实时计算中的聚合计算由OLAP引擎承担，从而减轻实时计算部分的聚合处理压力。这样做的优点是既可以满足数据分析师的实时自助分析需求，并且可以减轻计算引擎的处理压力，同时也减少了相应的开发和维护成本。缺点是对OLAP 引擎的数据写入性能和计算性能有更高的要求。

3.2 实时计算引擎选型

Flink具备Exactly-once的语义，轻量级checkpoint容错机制、低延迟、高吞吐和易用性高的特点。SparkStreaming 更适合微批处理。我们选择了使用 Flink。

3.3 OLAP选型

我们选择 StarRocks 作为 OLAP 计算引擎。主要原因有3个：

StarRocks 是一种使用MPP分布式执行框架的数据库，集群查询性能强大；
StarRocks在高并发查询和多表关联等复杂多维分析场景中表现出色，并发能力强于clickhouse，而携程酒店的业务场景需要OLAP数据库支持每小时几万次的查询量；
StarRocks 提供了4种数据模型，可以更好的应对携程酒店的各种业务场景

四、携程酒店实时订单

4.1 数据源

Mysql Binlog，通过携程自研平台 Muise接入生成 Kafka。

4.2 ETL数据处理

问题一：如何保证消息处理的有序性？

Muisev平台保证了Binlog消息的有序性，这里需要讨论的是ETL过程中如何保证消息的有序性。例如：一个酒店订单先在同一张表触发了两次更新操作，共计有了两条 Binlog 消息，消息1和消息2会先后进入流处理系统，如果这两个消息是在不同的Flink Task上进行处理，那么就有可能由于两个并发处理的速度不一致，先发生的消息后处理，导致最终输出的结果不对（出现乱序）

上图是一个简化的过程，业务库流入到Kafka，Binlog 日志是顺序写入的，根据主键进行Hash分区，保证同一个主键的数据写入到kafka同一个分区。当Flink消费kafka时，需要设置合理的并发，保证同一个分区的数据由一个Task负责，另外尽量采取逻辑主键作为 Shuffle Key，从而保证了Flink内部的有序性。最后在写入StarRocks时，按照主键进行更新或删除操作，这样才能保证端到端的一致性。

问题二：如何生产实时订单宽表？

为了方便分析师和数据应用使用，我们需要生成明细订单宽表并存储在 StarRocks 上。酒店订单涉及的业务过程相对复杂，数据源来自多个数据流中，且由于酒店订单变化生命周期较长，客人可能会提前几个月甚至更久预订下单。这些都给生产实时订单宽表带来一定的困难。

上图中生成订单宽表的sql逻辑在离线批处理场景下没有问题，但是实时场景下，这个sql会按照双流join的方式依次处理，每次只能处理一个join，所以上面代码有9个 Join 节点，Join节点会将左流的数据和右流的数据全部保存下来，最终会导致join过程中state状态存储膨胀了9倍。

因此，我们采用了union all + group by的方式替代join；先用union all把数据错位拼接到一起，然后再最外层进行group by。这种方式相当于将 Join 关联转换成group by，不会放大 Flink的状态存储。

还有一个问题，上面说过酒店订单的生命周期很长，用 union all 的方式，状态周期只保存了30分钟，一些订单的状态可能已经过期，当出现订单状态时，我们需要获取订单的历史状态，这样就需要一个中间层保存历史状态数据来做补充。历史数据我们选择存放在 Redis 中，第一次选择从离线数据导入，实时更新数据的同时，还更新 Redis和StarRocks。

问题三：如何做数据校验？

实时数据存在数据丢失或逻辑变更不及时的风险，为了保证数据的准确性，每日凌晨将实时数据和离线T-1数据做比对，如果数据校验不一致，会用离线数据更新StarRocks中对应的数据，并排查原因。

整体流程见下图: