Logo

菜单

神经蛙
神经蛙
发布于 2026-10-30 / 29 阅读
0
0

Apache Doris 实时数仓核心实战：MPP 架构、数据模型、导入与高并发查询

#Apache Doris#大数据#实时数仓#OLAP#MPP

前面写了 ClickHouse 实时 OLAP、Kafka 消息、Spark/Flink 计算、Hive 离线数仓、HBase 存储、ZooKeeper 协调——这套"采-存-算-协"链路里，还差一块：既能扛高并发点查、又能做实时多维分析的"服务化数仓"。ClickHouse 擅长大宽表聚合，但高并发点查和频繁更新是短板；Apache Doris 正好补上这个位置。本文把 Doris 的核心架构、数据模型、导入与查询优化一次讲透。

一、Doris 解决什么问题

传统数仓有两个老大难：

场景 离线数仓（Hive） 纯 OLAP 列存（ClickHouse）
即席多维分析 慢（分钟级） 快，但高并发点查易抖
数据更新 / 覆盖 不便（重写分区） 弱（Mutation 异步）
高并发（几百 QPS 点查） 扛不住 非设计目标
上手成本 需 Hive/Spark 栈 需独立运维

Doris 的定位：一个兼容 MySQL 协议、MPP 架构、既适合实时多维分析又适合高并发点查 / 实时更新的实时数仓。它把"数仓"做成了一个可以直接被业务系统（报表、API、BI）查询的服务，而不是一套批处理任务。

一句话区分：ClickHouse 是"分析型列存引擎"，Doris 是"服务化的实时数仓"。前者极致单表聚合，后者兼顾分析、更新、点查与生态（MySQL 协议、物化视图自动路由）。两者常组合使用，而非互相替代。

二、整体架构：FE 与 BE

Doris 采用存算一体的 MPP 架构，只有两类进程：

         MySQL Client / BI / JDBC  (port 9030)
                    │
            ┌───────▼───────┐
            │   FE (Frontend)│  ← 元数据 + 查询规划 + 接入层
            │ Leader/Follower│
            │    /Observer   │
            └───────┬───────┘
        ┌───────────┼───────────┐
        ▼           ▼           ▼
   ┌─────────┐ ┌─────────┐ ┌─────────┐
   │  BE 0   │ │  BE 1   │ │  BE 2   │  ← 存储 + 执行
   └─────────┘ └─────────┘ └─────────┘
        │           │           │
      Tablet      Tablet      Tablet   (数据分片，三副本)
  • FE（Frontend）：负责元数据管理、查询解析与规划、客户端接入。通过 Raft（BDBJE） 保证元数据高可用，角色分 Leader / Follower（参与选主、奇数部署）/Observer（只同步、扩展查询能力）。
  • BE（Backend）：负责数据存储与查询执行，每个 Tablet 默认三副本，副本间通过分桶键哈希分布。
  • 接入零改造：客户端用标准 MySQL 协议（默认 9030 端口）连 FE，直接用 mysql 命令行或任意 JDBC/ORM 就能查，这是它"服务化"的关键。

Doris 早期依赖 ZooKeeper/MySQL 存元数据，现代版本（1.x 之后）已内置 BDBJE 做元数据一致性，外部组件极少，部署比 Hive+Spark 轻得多。

三、核心概念：Tablet、分区、分桶

概念 含义 作用
Partition（分区） 按时间/枚举范围切分大表 分区裁剪，删旧数据只 drop 分区
Bucket（分桶） 分区内按哈希分桶键再切 数据均匀分布，Join/聚合本地化
Tablet 分桶后的最小数据分片 调度、均衡、备份的基本单位
Rowset / Version 每次导入生成一个 Rowset 多版本，Compaction 合并
Compaction 小 Rowset 合并 控制读放大与空间放大

建表示例（明细模型 + 天级分区 + 城市分桶）：

CREATE TABLE dwd_order_detail (
    dt          DATE        NOT NULL,
    order_id    BIGINT      NOT NULL,
    city        VARCHAR(32) NOT NULL,
    user_id     BIGINT      NOT NULL,
    amount      DECIMAL(10,2),
    status      TINYINT
)
ENGINE=OLAP
DUPLICATE KEY(dt, order_id)        -- 排序键（前缀索引来源）
PARTITION BY RANGE(dt) (
    PARTITION p20261001 VALUES LESS THAN ('2026-10-02'),
    PARTITION p20261002 VALUES LESS THAN ('2026-10-03')
)
DISTRIBUTED BY HASH(city) BUCKETS 16  -- 分桶键选高基数列
PROPERTIES (
    "replication_num" = "3",
    "dynamic_partition.enable" = "true",      -- 自动建/删分区
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.end" = "3"
);

四、三种数据模型

Doris 的"数据模型"决定了一行数据写入后如何被处理，是选型的命门：

模型 写入语义 适用场景 典型用法
Duplicate Key 保留所有明细 原始日志、订单流水 最常用，明细层 DWD
Aggregate Key 相同 Key 列预聚合 指标累加（SUM/MAX） 轻度汇总 DWS
Unique Key 相同 Key 覆盖更新 维表、状态、CDC 结果 用户画像、订单状态 
-- Aggregate Key：相同 (dt,sku) 自动把 amount 累加、pv 取 max
CREATE TABLE agg_sku (
    dt   DATE    NOT NULL,
    sku  VARCHAR(32) NOT NULL,
    amount DECIMAL(10,2) SUM,
    pv     BIGINT MAX
) ENGINE=OLAP
AGGREGATE KEY(dt, sku)
DISTRIBUTED BY HASH(sku) BUCKETS 16;

-- Unique Key：相同 user_id 后写入覆盖前写入（实现 UPSERT）
CREATE TABLE dim_user (
    user_id BIGINT NOT NULL,
    name    VARCHAR(32) REPLACE,
    level   TINYINT REPLACE
) ENGINE=OLAP
UNIQUE KEY(user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 16;

Unique Key 在 1.2 之前是 Merge-on-Read（读时合并，点查慢），2.0 起默认 Merge-on-Write（写时合并，点查接近行存性能）。做高并发点查务必确认版本与 。

五、分区 vs 分桶（与 Hive 对照）

维度 Doris Hive
分区 RANGE 为主，支持动态分区 静态/动态分区，文件目录
分桶 HASH 分桶，Tablet 均匀分布 同 HASH/分桶，但偏文件级
裁剪 分区+分桶双重裁剪 分区裁剪为主
更新 Unique Key 原生 UPSERT 需重写分区/ACID 表
小文件 Compaction 自动合并 易产生大量小文件

经验：分区按查询最常用的时间列，分桶键选高基数且常做 Join/Group 的列（如 user_id、city），桶数按数据量评估（每桶 1~10GB 较健康）。

六、物化视图与 Rollup

Doris 支持异步物化视图，对上层的复杂聚合做预计算，并能在查询时自动路由到命中的物化视图：

CREATE MATERIALIZED VIEW mv_sku_daily
AS
SELECT dt, sku, SUM(amount) sum_amt, COUNT(*) cnt
FROM agg_sku
GROUP BY dt, sku;
-- 之后查 SELECT dt,sku,SUM(amount) ... 会被自动改写命中 mv_sku_daily

物化视图的"透明加速"是 Doris 区别于手算汇总表的最大便利：业务 SQL 不用改，优化器自动选最优物化视图，避免 Hive 里维护一堆派生表的痛苦。

七、导入方式全景

导入方式 触发 数据来源 场景
Stream Load HTTP 推 本地 CSV/JSON 文件 小批量、手动/程序推送
Routine Load 常驻 Kafka 主题 实时增量消费
Insert Into SQL 内部表/外部表 ETL 内部流转
Broker Load 异步 HDFS / S3 / OSS 大数据量离线导入
Spark Load 异步 Hive 外部表 超大规模、借助 Spark 预处理

Stream Load 示例（从文件推）：

curl -u root: -H "label:order_20261030_01" \
     -H "column_separator:," \
     -T order.csv \
     http://fe-host:8030/api/db1/dwd_order_detail/_stream_load

八、实时入仓链路（Kafka / Flink）

结合本博客已有的 Kafka、Flink 文章，Doris 是它们的实时落地点：

Kafka Topic  ──Routine Load──►  Doris 明细表  ──► 物化视图  ──►  BI/API
     │
     └──Flink Job──► Doris（Flink Connector，两阶段提交 Exactly-Once）

Routine Load（纯 SQL 消费 Kafka）：

CREATE ROUTINE LOAD db1.rl_order ON dwd_order_detail
COLUMNS(dt, order_id, city, user_id, amount, status)
PROPERTIES (
    "desired_concurrent_number" = "3",
    "max_batch_interval" = "10"
)
FROM KAFKA (
    "kafka_broker_list" = "k1:9092,k2:9092",
    "kafka_topic" = "order_event",
    "property.group.id" = "doris_consumer"
);

Flink 写入（两段提交保证 Exactly-Once，与 Flink 篇的 Checkpoint 对齐）：

-- Flink SQL Sink
CREATE TABLE doris_sink (
    dt DATE, order_id BIGINT, amount DECIMAL(10,2)
) WITH (
    'connector' = 'doris',
    'fenodes' = 'fe-host:8030',
    'table.identifier' = 'db1.dwd_order_detail',
    'username' = 'root',
    'sink.enable-2pc' = 'true',         -- 两阶段提交
    'sink.label-prefix' = 'flink_order'
);

九、查询优化要点

优化手段 原理 适用
前缀索引 对排序键前 36 字节建稀疏索引 按排序键过滤的点查/范围
倒排索引 2.0+ 对任意列建索引 等值/范围/全文检索
高并发点查 Unique Key + 行存 + 主键索引 几百~几千 QPS 的单行查
Colocate Join 同分桶键表同分布，免 Shuffle 大表 Join 大表
Broadcast Join 小表广播 小维表 Join 事实表 
-- 开启某列倒排索引，加速 status 过滤
ALTER TABLE dwd_order_detail
ADD INDEX idx_status (status) USING INVERTED;

-- Colocate Join：两表同 distribution 即可本地 Join
CREATE TABLE fact_a (...) DISTRIBUTED BY HASH(user_id) BUCKETS 16
PROPERTIES("colocate_with" = "group_user");

十、Doris vs ClickHouse（定位对照）

维度 Apache Doris ClickHouse
接入协议 MySQL 协议 自有 HTTP/TCP
高并发点查 强（行存+主键） 一般
数据更新 Unique Key UPSERT Mutation 异步
大宽表聚合 强 极强（向量化）
物化视图 自动路由 手动/物化较弱
运维复杂度 低（FE+BE） 中（需 ZooKeeper 等）

结论：分析型重聚合用 ClickHouse，服务化实时数仓 / 高并发点查 / 频繁更新用 Doris，二者可在同一架构并存。

十一、生产避坑清单

  1. 分桶键别选低基数列（如性别），否则数据严重倾斜、BE 热点。
  2. Unique Key 点查先确认 Merge-on-Write（2.0+），否则读放大严重。
  3. 动态分区要显式开启，否则历史分区堆积、磁盘爆满。
  4. Routine Load 注意 label 唯一与并发，消费堆积时调 desired_concurrent_number。
  5. 一次导入的 Tablet 版本别太多，否则 Compaction 跟不上、查询变慢。
  6. 高并发点查走 PreparedStatement + 连接池，别每次建连。
  7. Colocate Join 需同桶数同键，否则退化成 Shuffle Join 反而更慢。
  8. FE Follower 必须奇数（3/5），Observer 可任意扩展查询。
  9. 大批量导入优先 Broker/Spark Load，别用 Stream Load 推超大文件。
  10. 监控 Compaction Score 与 BE 磁盘水位，是稳定性的两大先行指标。

本文补上了本博客大数据"采-存-算-协"链路里服务化实时数仓这块拼图：Kafka 负责采集、Spark/Flink 负责计算、Hive 负责离线、HBase/ClickHouse 负责存储与分析，Doris 则把分析结果直接变成可被业务高并发查询的服务。下一篇可延伸到 Doris 与 Hive/Iceberg 湖仓一体 或 Flink CDC 实时同步入 Doris，敬请期待。

微前端架构实战：Module Fe...
Apache Iceberg 数据...

评论