Doris 介绍

Apache Doris 是一个基于 MPP（大规模并行处理） 架构的高性能、实时的分析型数据库，以极速易用的特点被人们所熟知，MPP是将任务并行的分散到多个服务器和节点上，在每个节点完成计算之后，将各自部分的结果汇总到一起得到最终得到结果，仅需亚秒级响应时间即可返回海量数据下的查询结果，有效的支持实时数据分析。
可以满足多种数据分析需求，例如：固定历史报表、实时数据分析，交互式数据分析和探索式数据分析

OLAP（OnlineAnalyticalProcessing联机分析处理），是大数据分析的应用技术，提供复杂的分析操作、侧重决策支持

发展历史

doris发展历史.jpg

架构简图

doris架构.png

核心特性

• 采用 MySQL 协议，高度兼容 MySQL 语法，支持标准 SQL
• 列式存储
• Doris 支持多种存储模型，Aggregate、Unique、Duplicate
• 查询引擎方面，基于MPP架构的分析型数据库;
• 向量化的查询引擎，所有的内存结构能够按照列式布局，能够达到大幅减少虚函数调用、提升 Cache 命中率，高效利用 SIMD 指令的效果
• 采用了 Adaptive Query Execution 技术， 可以根据 Runtime Statistics 来动态调整执行计划
• 优化器方面 Doris 使用 CBO(Cost-Based Optimizer)基于成本的优化器 和 RBO(Rule-Based Optimizer)基于规则的优化器 结合的优化策略，RBO 支持常量折叠、子查询改写、谓词下推等，CBO 支持 Join Reorder

行和列存储对比.png

OLAP引擎

• 上卷（Roll Up）/聚合：选定某些维度，根据这些维度来聚合事实，如果用SQL来表达就是select dim_a, aggs_func(fact_b) from fact_table group by dim_a.
• 下钻（Drill Down）：上卷和下钻是相反的操作。它是选定某些维度，将这些维度拆解出小的维度（如年拆解为月，省份拆解为城市），之后聚合事实。
• 切片（Slicing、Dicing）：选定某些维度，并根据特定值过滤这些维度的值，将原来的大Cube切成小cube。如dim_a in (‘CN’, ‘USA’)

• 旋转（Pivot/Rotate）：维度位置的互换。

  
  
  olap引擎常见操作.png

doris 的使用场景

数据源经过各种数据集成和加工处理后，通常会入库到实时数仓 Doris 和离线湖仓（Hive, Iceberg, Hudi 中），Apache Doris 被广泛应用在以下场景中

doris使用场景_0.png

doris使用场景.png

• 报表分析
  实时看板 （Dashboards）。
  面向企业内部分析师和管理者的报表。
  面向用户或者客户的高并发报表分析（Customer Facing Analytics）。比如面向网站主的站点分析、面向广告主的广告报表，并发通常要求成千上万的 QPS ，查询延时要求毫秒级响应。著名的电商公司京东在广告报表中使用 Apache Doris ，每天写入 100 亿行数据，查询并发 QPS 上万，99 分位的查询延时 150ms。

• 即席查询（Ad-hoc Query）
  面向分析师的自助分析，查询模式不固定，要求较高的吞吐。小米公司基于 Doris 构建了增长分析平台（Growing Analytics，GA），利用用户行为数据对业务进行增长分析，平均查询延时 10s，95 分位的查询延时 30s 以内，每天的 SQL 查询量为数万条。

• 统一数仓构建
  一个平台满足统一的数据仓库建设需求，简化繁琐的大数据软件栈。海底捞基于 Doris 构建的统一数仓，替换了原来由 Spark、Hive、Kudu、Hbase、Phoenix 组成的旧架构，架构大大简化。

• 数据湖联邦查询
  通过外表的方式联邦分析位于 Hive、Iceberg、Hudi 中的数据，在避免数据拷贝的前提下，查询性能大幅提升

环境初始化

• 下载doris安装包
• 安装部署，配置FE、BE的信息
• doris启动（doris用户）

sh start_fe.sh --daemon
sh start_be.sh --daemon

部署FE需要注意：

• FE 的磁盘空间主要用于存储元数据，包括日志和 image
• 多个FE所在服务器的时钟必须保持一致（允许最多5秒的时钟偏差）
• FE角色分为Leader、Follower和Observer，Leader为Follower组中选举出来的一种角色，后续统称为Follower
• FE节点数量至少为1个Follower，该Follower就是Leader。Follower 的数量必须为奇数（因为投票选主），Observer 数量随意
• 当FE中部署 1 个 Follower 和 1 个 Observer 时，可以实现读高可用。当部署 3 个 Follower 时，可以实现读写高可用（HA）
• 根据经验，当集群可用性要求很高时（比如提供在线业务），可以部署3个Follower和1-3个Observer。如果是离线业务，建议部署1个Follower和1-3个Observer

部署BE需要注意：

• BE 的磁盘空间主要用于存放用户数据，总磁盘空间按用户总数据量 * 3（3副本）计算，然后，再预留额外 40% 的空间用作后台 compaction 以及一些中间数据的存放
• 一台机器上可以部署多个 BE 实例，但是只能部署一个 FE。如果需要 3 副本数据，那么至少需要 3 台机器各部署一个 BE 实例（而不是1台机器部署3个BE实例），BE 实例数量直接决定了整体查询延迟

doris扩容

FE 扩容

1. 将node1节点上配置好的FE安装包发送到node2,node3节点上
2. 在node2、node3上修改fe.conf配置文件

• priority_networks:指定FE唯一的IP地址，必须配置（可以是ip段）
• meta_dir:元数据目录，可以不配置，默认是Doris FE安装目录下的doris-meta目录，如果指定其他目录需要提前创建好目录。在生成环境中建议目录放在单独的磁盘上
  如果缩容后再将该节点加入到集群中需要清空元数据目录：rm -rf /software/doris-1.2.1/apache-doris-fe/doris-meta/*

3. 启动FE ，必须借助已经成功部署的FE节点，MASTER_ADDRESS是FE的主节点（默认第一个Follower节点），端口（fe.conf配置文件的EditLogPort属性，默认90100）

sh start_fe.sh --helper ${MASTER_ADDRESS}:${HEADBEAT_PORT} --daemon

4. 添加FE Follower 到Doris集群

mysql> ALTER SYSTEM ADD FOLLOWER "node2:9010";

5.如果要缩容，对应

mysql> ALTER SYSTEM DROP FOLLOWER "node2:9010";

6.查看FE的状态

mysql> SHOW FRONTENDS\G

FE扩容需要注意点：

• Follower FE（包括 Leader）的数量必须为奇数，建议最多部署 3 个组成高可用（HA）模式即可。
• 当 FE 处于高可用部署时（1个 Leader，2个 Follower），我们建议通过增加 Observer FE 来扩展 FE 的读服务能力
• 添加FE 时需要将对应安装包中doris-meta目录清空
• helper 不能指向 FE 自身，必须指向一个或多个已存在并且正常运行中的 Master/Follower FE
• 删除 Follower FE 时，确保最终剩余的 Follower（包括 Leader）节点为奇数

BE 扩容

1. 将doris的be安装包通过scp指令传输到不同的机器
2. 修改be的配置文件
   priority_networks:指定BE唯一的IP地址（可以是ip段）
   storage_root_path:配置BE数据存储目录。默认目录在BE安装目录的storage目录下，如果指定其他目录需要提前创建好目录，可以用逗号分开指定多个路径，也可以在路径后加入.HDD/.SSD指定数据存储磁盘类型。
3. 启动BE

sh start_be.sh --daemon

4. 添加BE到doris集群

mysql> ALTER SYSTEM ADD BACKEND "node3:9050";

5.查看BE节点的命令

mysql> SHOW BACKENDS\G

BE扩容需要注意点：

• BE 扩容后，Doris 会自动根据负载情况，进行数据均衡，期间不影响使用
• 对BE进行缩容时建议使用DECOMMISSION 命令操作
• BE缩容不一定执行成功，例如：剩余 BE 存储空间不足以容纳下线 BE 上的数据，或者剩余机器数量不满足最小副本数时，该命令都无法完成
• 可以通过CANCEL DECOMMISSION BACKEND "be_host:be_heartbeat_service_port" 命令取消缩容

broker扩容

略

doris 基本使用

从官网了解doris基本功能，从创建用户、创建数据库、创建表、数据导入、查询数据、Rollup等
[使用指南]https://doris.apache.org/zh-CN/docs/dev/data-table/basic-usage

doris数据存储模型

官网介绍数据模型

• Aggregate Key [ 聚合模型,merge on read ]
  Aggregate 模型可以通过预聚合，极大地降低聚合查询时所需扫描的数据量和查询的计算量，非常适合有固定模式的报表类查询场景。但是该模型对 count(*) 查询很不友好。同时因为固定了 Value 列上的聚合方式，在进行其他类型的聚合查询时，需要考虑语意正确性

• Unique Key [ 唯一模型，跟聚合模型的区别，merge on write ]
  Unique 模型针对需要唯一主键约束的场景，可以保证主键唯一性约束。但是无法利用ROLLUP等预聚合带来的查询优势
  Unique 模型仅支持整行更新，如果用户既需要唯一主键约束，又需要更新部分列（例如将多张源表导入到一张 doris 表的情形），则可以考虑使用 Aggregate 模型，同时将非主键列的聚合类型设置为 REPLACE_IF_NOT_NULL

• Duplicate Key[明细模型，不会有任何聚合，DUPLICATE KEY(k1,k2)用于存储时排序]
  Duplicate 这种数据模型适用于既没有聚合需求，又没有主键唯一性约束的原始数据的存储，适合任意维度的 Ad-hoc 查询。虽然同样无法利用预聚合的特性，但是不受聚合模型的约束，可以发挥列存模型的优势（只读取相关列，而不需要读取所有 Key 列）

doris数据划分

在 Doris 中，数据都以表（Table）的形式进行逻辑上的描述

ROW & Column

• Row 代表用户的一行数据，
• Column用于描述一行数据中不同的字段
  Column 可以分为两大类：Key 和 Value。
  1. 从业务角度看，Key 和 Value 可以分别对应维度列和指标列。Doris的key列是建表语句中指定的列，建表语句中的关键字'unique key'或'aggregate key'或'duplicate key'后面的列就是key列，除了key列剩下的就是value列
  2. 从聚合模型的角度来说，Key 列相同的行，会聚合成一行

Tablet & Partition

在 Doris 的存储引擎中，用户Table数据被水平划分为若干个数据分片（Tablet，也称作数据分桶）
每个 Tablet 包含若干数据行
各个 Tablet 之间的数据没有交集，并且在物理上是独立存储的

多个 Tablet 在逻辑上归属于不同的分区（Partition）
一个 Tablet 只属于一个 Partition，一个 Partition 包含若干个 Tablet，一个 Table 包含若干个 Partition

因为 Tablet 在物理上是独立存储的，所以可以视为 Partition 在物理上也是独立

Tablet 是最小物理存储单元。数据移动、复制等操作
Partition 是逻辑上最小的管理单元。数据的导入与删除，仅能针对一个 Partition 进行

doris分区分桶

doris分区分桶

列定义建议

在定义Doris表中列类型时有如下建议：

• AGGREGATE KEY 数据模型Key 列必须在所有 Value 列之前
• 尽量选择整型类型。因为整型类型的计算和查找效率远高于字符串
• 对于不同长度的整型类型的选择原则，遵循够用即可
• 对于 VARCHAR 和 STRING 类型的长度，遵循够用即可
• 表中一行数据所有列总的字节数不能超过100KB。如果数据一行非常大，建议拆分数据进行多表存储

关于 Partition 和 Bucket 的数量和数据量的建议

• 一个表的 Tablet 总数量等于 (Partition num * Bucket num)
• 一个表的 Tablet 数量，在不考虑扩容的情况下，推荐略多于整个集群的磁盘数量
• 单个 Tablet 的数据量理论上没有上下界，但建议在 1G - 10G 的范围内。如果单个 Tablet 数据量过小，则数据的聚合效果不佳，且元数据管理压力大。如果数据量过大，则不利于副本的迁移、补齐，且会增加 Schema Change 或者 Rollup 操作失败重试的代价（这些操作失败重试的粒度是 Tablet）
• 当 Tablet 的数据量原则和数量原则冲突时，建议优先考虑数据量原则
• 在建表时，每个分区的 Bucket 数量统一指定。但是在动态增加分区时（ADD PARTITION），可以单独指定新分区的 Bucket 数量。可以利用这个功能方便的应对数据缩小或膨胀
• 一个 Partition 的 Bucket 数量一旦指定，不可更改。所以在确定 Bucket 数量时，需要预先考虑集群扩容的情况。比如当前只有 3 台 host，每台 host 有 1 块盘。如果 Bucket 的数量只设置为 3 或更小，那么后期即使再增加机器，也不能提高并发度

举一些例子：假设在有10台BE，每台BE一块磁盘的情况下：
如果一个表总大小为 500MB，则可以考虑4-8个分桶
如果一个表总大小为 5GB：8-16个分桶
如果一个表总大小为 50GB：32个分桶
如果一个表总大小为 500GB：建议分区，每个分区大小在 50GB 左右，每个分区16-32个分桶
如果一个表总大小为 5TB：建议分区，每个分区大小在 50GB 左右，每个分区16-32个分桶

doris索引

前缀索引

Doris中默认将一行数据的前36个字节作为这行数据的前缀索引，但是当遇到VARCHAR类型时，前缀索引会直接截断（varchar类型最多使用20个字符）

Bitmap索引

bitmap，位图，是一种数据结构，即bit的集合，每一个bit记录0或者1，代表状态。bitmap index是位图索引，可以针对Doris表中的某些列构建位图索引来加快数据查询速度
CREATE INDEX [IF NOT EXISTS] index_name ON table (某一个列) USING BITMAP COMMENT '注释';

bitmap 索引仅在单列上创建，不支持多列
bitmap 索引能够应用在 Duplicate、Uniqe数据模型的所有列和 Aggregate模型的key列上

Bloom Filter索引

本质上是一种位图结构，用于快速的判断一个给定的值是否在一个集合中，这种判断会产生小概率的误判，即如果返回false
在建表语句的PROPERTIES里加上"bloom_filter_columns"="k1,k2,k3"

• BloomFilter适用于非前缀过滤
• 查询会根据该列高频过滤，而且查询条件大多是 in 和 = 过滤，只对 in 和 = 过滤查询有加速效果
• 不同于Bitmap, BloomFilter适用于高基数列。比如UserID。因为如果创建在低基数的列上，比如 “性别” 列，则每个Block几乎都会包含所有取值，导致BloomFilter索引失去意义

Doris索引总结

• Doris对数据进行有序存储, 在数据有序的基础上为其建立稀疏索引,索引粒度为 block(1024行)
• 稀疏索引选取 schema 中固定长度的前缀作为索引内容, 目前 Doris 选取 36 个字节的前缀作为索引
• 建表时建议将查询中常见的过滤字段放在 Schema 的前面, 区分度越大，频次越高的查询字段越往前放
• 这其中有一个特殊的地方,就是 varchar 类型的字段。varchar 类型字段只能作为稀疏索引的最后一个字段。索引会在 varchar 处截断, 因此 varchar 如果出现在前面，可能索引的长度可能不足 36 个字节
• 除稀疏索引之外,Doris还提供bloomfilter索引, bloomfilter索引对区分度比较大的列过滤效果明显。如果考虑到varchar不能放在稀疏索引中, 可以建立bloomfilter索引

doris的 Rollup

Rollup 可以理解为 Base Table 的一个物化索引结构，“物化”是因为其数据在物理上独立存储，而“索引”的意思是建立 Rollup 时可只选取 Base Table 中的部分列作为 Schema,Schema 中的字段顺序也可与 Base Table 不同

ROLLUP使用说明

• ROLLUP 最根本的作用是提高某些查询的查询效率（无论是通过聚合来减少数据量，还是修改列顺序以匹配前缀索引）
• ROLLUP 是附属于 Base 表的，可以看做是 Base 表的一种辅助数据结构。用户可以在 Base 表的基础上，创建或删除 ROLLUP，但是不能在查询中显式的指定查询某 ROLLUP。是否命中 ROLLUP 完全由 Doris 系统自动决定
• ROLLUP 的数据是独立物理存储的。因此，创建的 ROLLUP 越多，占用的磁盘空间也就越大。同时对导入速度也会有影响（导入的ETL阶段会自动产生所有 ROLLUP 的数据），但是不会降低查询效率（只会更好）
• ROLLUP 的数据更新与 Base 表是完全同步的
• ROLLUP 中列的聚合方式，与 Base 表完全相同。在创建 ROLLUP 无需指定，也不能修改
• 查询能否命中 ROLLUP 的一个必要条件（非充分条件）是，查询所涉及的所有列（包括 select list 和 where 中的查询条件列等）都存在于该 ROLLUP 的列中。否则，查询只能命中 Base 表
• 某些类型的查询（如 count(*)）在任何条件下，都无法命中 ROLLUP

## 创建
ALTER TABLE table ADD ROLLUP example_rollup_index(k1, k3, v1, v2);

## 查看
SHOW ALTER TABLE ROLLUP;

## 取消当前正在执行的作业
CANCEL ALTER TABLE ROLLUP FROM table;

## 查看已经创建的 ROLLUP
SHOW ALTER TABLE ROLLUP FROM table;

## 查看执行计划是否命中 ROLLUP
explain select k1,k3 FROM table group by k1;

举例：
原表table的Schema如下

+----------+-------------+------+-------+---------+-------+
 |Field|Type|Null|Key|Default|Extra| 
+----------+-------------+------+-------+---------+-------+ 
|siteid|int(11)|No|true|10|| 
|citycode|smallint(6)|No|true|N/A|| 
|username|varchar(32)|No|true||| 
|pv|bigint(20)|No|false|0|SUM| 
|uv|bigint(20)|No|false|0|SUM| 
+----------+-------------+------+-------+---------+-------+

对于 table 明细数据是 siteid, citycode, username 三者构成一组 key，从而对 pv 字段进行聚合；如果业务方经常有看城市 pv 总量的需求，可以建立一个只有 citycode, pv 的rollup

doris 数据导入

Insert Into（同步）

适用场景

• 用户希望仅导入几条假数据，验证一下 Doris 系统的功能
• 用户希望将已经在 Doris 表中的数据进行 ETL 转换并导入到一个新的 Doris 表中
• 用户可以创建一种外部表，如 MySQL 外部表映射一张 MySQL 系统中的表。或者创建 Broker 外部表来映射 HDFS 上的数据文件

INSERT INTO tbl SELECT ...
INSERT INTO tbl (col1, col2, ...) VALUES (1, 2, ...), (1,3, ...);

示例：

INSERT INTO tbl2 WITH LABEL label1 SELECT * FROM tbl3;
INSERT INTO tbl1 VALUES ("qweasdzxcqweasdzxc"), ("a");

1. Session变量，数据导入异常，源数据列长度超过目的数据列长度、列类型不匹配、分区不匹配、列顺序不匹配
   如果 enable_insert_strict 设为 true（默认true,建议为true），则 Insert Into 可能会失败。
   如果 enable_insert_strict 设为 false，则可能出现仅导入了部分合格数据的情况，通过 show load where label="xxx"; 查看被过滤的行

2. 插入数据量大，可能会超时
   Insert Into 对数据量没有限制，大数据量导入也可以支持。用户预估的导入数据量过大，就需要修改系统的 Insert Into 导入超时时间
   2.1 预估导入时间
   导入预估时间 = 数据量大小 / doris集群平均导入速度，例如：10G / 5M/s = 2000s
   2.2 如果超时了，可以修改 FE 配置
   insert_load_default_timeout_second = 2000

Binlog Load（增量、异步）

用户在Mysql数据库的对数据更新操作的CDC(Change Data Capture)功能，
Binlog Load需要依赖canal作为中间媒介，让canal伪造成一个从节点去获取Mysql主节点上的Binlog并解析，再由Doris去获取Canal上解析好的数据;

1. 用户向FE提交一个数据同步作业，FE会为每个数据同步作业启动一个canal client，来向canal server端订阅并获取数据。
2. client中的receiver将负责通过Get命令接收数据，每获取到一个数据batch，都会由consumer根据对应表分发到不同的channel，每个channel都会为此数据batch产生一个发送数据的子任务Task
3. 在FE上，一个Task是channel向BE发送数据的子任务，里面包含分发到当前channel的同一个batch的数据
4. channel控制着单个表事务的开始、提交、终止。一个事务周期内，一般会从consumer获取到多个batch的数据，因此会产生多个向BE发送数据的子任务Task，在提交事务成功前，这些Task不会实际生效
5. 满足一定条件时（比如超过一定时间、达到提交最大数据大小），consumer将会阻塞并通知各个channel提交事务

• 当且仅当所有channel都提交成功，才会通过Ack命令通知canal并继续获取并消费数据
• 如果有任意channel提交失败，将会重新从上一次消费成功的位置获取数据并再次提交（已提交成功的channel不会再次提交以保证幂等性）

6. 整个数据同步作业中，FE通过以上流程不断的从canal获取数据并提交到BE，来完成数据同步

doris_binlog导入

Broker Load

适用场景

• 源数据在 Broker 可以访问的存储系统中，如 HDFS（BOS，AFS）
• 数据量在 几十到百GB 级别。

用户在提交导入任务后，FE 会生成对应的 Plan 并根据目前 BE 的个数和文件的大小，将 Plan 分给 多个 BE 执行，每个 BE 执行一部分导入数据
BE 在执行的过程中会从 Broker 拉取数据，在对数据 transform 之后将数据导入系统
所有 BE 均完成导入，由 FE 最终决定导入是否成功

                +
                 | 1. user create broker load
                 v
            +----+----+
            |         |
            |   FE    |
            |         |
            +----+----+
                 |
                 | 2. BE etl and load the data
    +--------------------------+
    |            |             |
+---v---+     +--v----+    +---v---+
|       |     |       |    |       |
|  BE   |     |  BE   |    |   BE  |
|       |     |       |    |       |
+---+-^-+     +---+-^-+    +--+-^--+
    | |           | |         | |
    | |           | |         | | 3. pull data from broker
+---v-+-+     +---v-+-+    +--v-+--+
|       |     |       |    |       |
|Broker |     |Broker |    |Broker |
|       |     |       |    |       |
+---+-^-+     +---+-^-+    +---+-^-+
    | |           | |          | |
+---v-+-----------v-+----------v-+-+
|       HDFS/BOS/AFS cluster       |
|                                  |
+----------------------------------+

数据量
这里仅讨论单个 BE 的情况，如果用户集群有多个 BE 则下面标题中的数据量应该乘以 BE 个数来计算。比如：如果用户有3个 BE，则 3G 以下（包含）则应该乘以 3，也就是 9G 以下（包含）。

3G 以下（包含）
用户可以直接提交 Broker load 创建导入请求
3G 以上
由于单个导入 BE 最大的处理量为 3G，超过 3G 的待导入文件就需要通过调整 Broker load 的导入参数来实现大文件的导入

作业调度

• 系统会限制一个集群内，正在运行的 Broker Load 作业数量，以防止同时运行过多的 Load 作业
• 首先， FE 的配置参数：desired_max_waiting_jobs 会限制一个集群内，未开始或正在运行（作业状态为 PENDING 或 LOADING）的 Broker Load 作业数量。默认为 100。如果超过这个阈值，新提交的作业将会被直接拒绝
• 一个 Broker Load 作业会被分为 pending task 和 loading task 阶段。其中 pending task 负责获取导入文件的信息，而 loading task 会发送给BE执行具体的导入任务
• FE 的配置参数 async_pending_load_task_pool_size 用于限制同时运行的 pending task 的任务数量。也相当于控制了实际正在运行的导入任务数量。该参数默认为 10。也就是说，假设用户提交了100个Load作业，同时只会有10个作业会进入 LOADING 状态开始执行，而其他作业处于 PENDING 等待状态
• FE 的配置参数 async_loading_load_task_pool_size 用于限制同时运行的 loading task 的任务数量。一个 Broker Load 作业会有 1 个 pending task 和多个 loading task （等于 LOAD 语句中 DATA INFILE 子句的个数）。所以 async_loading_load_task_pool_size 应该大于等于 async_pending_load_task_pool_size

HDFS Load、S3 Load

这两种导入方式和Broker Load基本相同，只是语法上有点不一样，WITH BROKER broker_name () 替换为对应的数据源

Spark Load

适用场景

• 源数据在 Spark 可以访问的存储系统中，如 HDFS。
• 数据量在 几十 GB 到 TB 级别。

因为 Doris 表里的数据是有序的，所以 Broker load 在导入数据的时是要利用doris 集群资源对数据进行排序，对 Doris 的集群资源占用要比较大。如果有 Spark 计算资源建议使用 Spark load

Spark load 任务的执行主要分为以下5个阶段

1. FE 调度提交 ETL 任务到 Spark 集群执行
2. Spark 集群执行 ETL 完成对导入数据的预处理。包括全局字典构建（ BITMAP 类型）、分区、排序、聚合等
3. ETL 任务完成后，FE 获取预处理过的每个分片的数据路径，并调度相关的 BE 执行 Push 任务
4. BE 通过 Broker 读取数据，转化为 Doris 底层存储格式
5. FE 调度生效版本，完成导入任务

                 +
                 | 0. User create spark load job
            +----v----+
            |   FE    |---------------------------------+
            +----+----+                                 |
                 | 3. FE send push tasks                |
                 | 5. FE publish version                |
    +------------+------------+                         |
    |            |            |                         |
+---v---+    +---v---+    +---v---+                     |
|  BE   |    |  BE   |    |  BE   |                     |1. FE submit Spark ETL job
+---^---+    +---^---+    +---^---+                     |
    |4. BE push with broker   |                         |
+---+---+    +---+---+    +---+---+                     |
|Broker |    |Broker |    |Broker |                     |
+---^---+    +---^---+    +---^---+                     |
    |            |            |                         |
+---+------------+------------+---+ 2.ETL +-------------v---------------+
|               HDFS              +------->       Spark cluster         |
|                                 <-------+                             |
+---------------------------------+       +-----------------------------+

Doris配置Spark
FE底层通过执行spark-submit的命令去提交 Spark 任务，因此需要为 FE 配置 Spark 客户端

配置 SPARK_HOME 环境变量

enable_spark_load = ture
spark_home_default_dir = /software/spark-2.3.1

配置 SPARK 依赖包
将spark客户端下的jars文件夹内所有jar包归档打包成一个zip文件，并在FE的配置文件配置 spark_resource_path 项指向此 zip 文件，若此配置项为空，则FE会尝试寻找FE根目录下的 lib/spark2x/jars/spark-2x.zip 文件
spark_resource_path = /software/spark-2.3.1/jars/spark-2x.zip

修改spark-dpp包名
当提交 spark load 任务时，除了以上spark-2x.zip依赖上传到指定的远端仓库，FE 还会上传 DPP 的依赖包至远端仓库，Spark进行数据预处理时需要依赖DPP此包,该包位于FE节点的/software/doris-1.2.1/apache-doris-fe/spark-dpp路径下，默认名称为spark-dpp-1.0-SNAPSHOT-jar-with-dependencies.jar，在提交Spark Load任务后，Doris默认在/software/doris-1.2.1/apache-doris-fe/spark-dpp路径下找名称为“spark-dpp-1.0.0-jar-with-dependencies.jar”的依赖包，所以这里需要在所有FE节点上进行改名

Routine Load（Kafka）

例行导入（Routine Load）功能，支持用户提交一个常驻的导入任务，通过不断的从指定的数据源读取数据，将数据导入到 Doris 中
目前Routine Load仅支持从Kafka中导入数据

1. FE 通过 JobScheduler 将一个导入作业拆分成若干个 Task。每个 Task 负责导入指定的一部分数据。Task 被 TaskScheduler 分配到指定的 BE 上执行
2. 在 BE 上，一个 Task 被视为一个普通的导入任务，通过 Stream Load 的导入机制进行导入。导入完成后，向 FE 汇报
3. FE 中的 JobScheduler 根据汇报结果，继续生成后续新的 Task，或者对失败的 Task 进行重试
4. 整个 Routine Load 作业通过不断的产生新的 Task，来完成数据不间断的导入

         +---------+
         |  Client |
         +----+----+
              |
+-----------------------------+
| FE          |               |
| +-----------v------------+  |
| |                        |  |
| |   Routine Load Job     |  |
| |                        |  |
| +---+--------+--------+--+  |
|     |        |        |     |
| +---v--+ +---v--+ +---v--+  |
| | task | | task | | task |  |
| +--+---+ +---+--+ +---+--+  |
|    |         |        |     |
+-----------------------------+
     |         |        |
     v         v        v
 +---+--+   +--+---+   ++-----+
 |  BE  |   |  BE  |   |  BE  |
 +------+   +------+   +------+

语法

CREATE ROUTINE LOAD example_db.test1 ON example_tbl
        COLUMNS TERMINATED BY ",",
        COLUMNS(k1, k2, k3, v1, v2, v3 = k1 * 100)
        PROPERTIES
        (
            "desired_concurrent_number"="3",
            "max_batch_interval" = "20",
            "max_batch_rows" = "300000",
            "max_batch_size" = "209715200",
            "strict_mode" = "false"
        )
        FROM KAFKA
        (
            "kafka_broker_list" = "broker1:9092,broker2:9092,broker3:9092",
            "kafka_topic" = "my_topic",
            "property.group.id" = "xxx",
            "property.client.id" = "xxx",
            "property.kafka_default_offsets" = "OFFSET_BEGINNING"
        );

Stream Load（同步）

Stream load 是一个同步的导入方式，用户通过发送 HTTP 协议发送请求将本地文件或数据流导入到 Doris 中。Stream load 同步执行导入并返回导入结果。用户可直接通过请求的返回体判断本次导入是否成功

应用场景

• 使用 Stream load 的最合适场景就是原始文件在内存中或者在磁盘中。
• 由于 Stream load 是一种同步的导入方式，所以用户如果希望用同步方式获取导入结果，也可以使用这种导入

1. Stream load 中，Doris 会选定一个节点作为 Coordinator 节点。该节点负责接数据并分发数据到其他数据节点
2. 用户通过 HTTP 协议提交导入命令。如果提交到 FE，则 FE 会通过 HTTP redirect 指令将请求转发给某一个 BE。用户也可以直接提交导入命令给某一指定 BE
3. 导入的最终结果由 Coordinator BE 返回给用户

                         ^      +
                         |      |
                         |      | 1A. User submit load to FE
                         |      |
                         |   +--v-----------+
                         |   | FE           |
4. Return result to user |   +--+-----------+
                         |      |
                         |      | 2. Redirect to BE
                         |      |
                         |   +--v-----------+
                         +---+Coordinator BE| 1B. User submit load to BE
                             +-+-----+----+-+
                               |     |    |
                         +-----+     |    +-----+
                         |           |          | 3. Distrbute data
                         |           |          |
                       +-v-+       +-v-+      +-v-+
                       |BE |       |BE |      |BE |
                       +---+       +---+      +---+

示例：

curl --location-trusted -u root -T date -H "label:123" http://abc.com:8030/api/test/date/_stream_load

数据量
由于 Stream load 的原理是由 BE 发起的导入并分发数据，建议的导入数据量在 1G 到 10G 之间。由于默认的最大 Stream load 导入数据量为 10G，所以如果要导入超过 10G 的文件需要修改 BE 的配置 streaming_load_max_mb

Insert into < Stream Load < Broker Load < Spark Load

外部表同步数据

Doris 可以创建外部表。创建完成后，可以通过 SELECT 语句直接查询外部表的数据，也可以通过 INSERT INTO SELECT 的方式导入外部表的数据。

Doris 外部表目前支持的数据源包括：

• MySQL
• Oracle
• PostgreSQL
• SQLServer
• Hive
• Iceberg
• ElasticSearch

1.创建 ODBC Resource

CREATE EXTERNAL RESOURCE `oracle_test_odbc`
PROPERTIES (
    "type" = "odbc_catalog",
    "host" = "192.168.0.10",
    "port" = "8086",
    "user" = "oracle",
    "password" = "oracle",
    "database" = "oracle",
    "odbc_type" = "oracle",
    "driver" = "Oracle"
);

2.创建外部表

CREATE EXTERNAL TABLE `ext_oracle_demo` (
  `k1` decimal(9, 3) NOT NULL COMMENT "",
  `k2` char(10) NOT NULL COMMENT "",
  `k3` datetime NOT NULL COMMENT "",
  `k5` varchar(20) NOT NULL COMMENT "",
  `k6` double NOT NULL COMMENT ""
) ENGINE=ODBC
COMMENT "ODBC"
PROPERTIES (
    "odbc_catalog_resource" = "oracle_test_odbc",
    "database" = "oracle",
    "table" = "baseall"
);

3.创建 Doris 表

CREATE TABLE `doris_oralce_tbl` (
  `k1` decimal(9, 3) NOT NULL COMMENT "",
  `k2` char(10) NOT NULL COMMENT "",
  `k3` datetime NOT NULL COMMENT "",
  `k5` varchar(20) NOT NULL COMMENT "",
  `k6` double NOT NULL COMMENT ""
)
COMMENT "Doris Table"
DISTRIBUTED BY HASH(k1) BUCKETS 2
PROPERTIES (
    "replication_num" = "1"
);

4.导入数据

INSERT INTO doris_oralce_tbl SELECT k1,k2,k3 FROM ext_oracle_demo limit 100;

注意事项
必须保证外部数据源与 Doris 集群是可以互通，包括BE节点和外部数据源的网络是互通的。
ODBC 外部表本质上是通过单一 ODBC 客户端访问数据源，因此并不合适一次性导入大量的数据，建议分批多次导入

doris的最佳实践

最佳实践 | Apache Doris在小米数据场景的应用实践与优化

最佳实践 | Apache Doris 向量化版本在知乎舰桥平台实践

叮咚买菜基于Doris引擎的应用实践

某些场景的使用：

Broadcast/Shuffle Join

系统默认实现 Join 的方式，是将小表进行条件过滤后，将其广播到大表所在的各个节点上，形成一个内存 Hash 表，然后流式读出大表的数据进行Hash Join。但是如果当小表过滤后的数据量无法放入内存的话，此时 Join 将无法完成，通常的报错应该是首先造成内存超限

如果遇到上述情况，建议显式指定 Shuffle Join，也被称作 Partitioned Join。即将小表和大表都按照 Join 的 key 进行 Hash，然后进行分布式的 Join。这个对内存的消耗就会分摊到集群的所有计算节点上

Doris会自动尝试进行 Broadcast Join，如果预估小表过大则会自动切换至 Shuffle Join。注意，如果此时显式指定了 Broadcast Join 也会自动切换至 Shuffle Join

Colocation Join

doris 除了支持Broadcast/Shuffle Join 之外，Colocation Join更是一大特色。Colocation Join 功能，是将一组拥有相同 CGS 的 Table 组成一个 CG。并保证这些 Table 对应的数据分片会落在同一个 BE 节点上。使得当 CG 内的表进行分桶列上的 Join 操作时，可以通过直接进行本地数据 Join，减少数据在节点间的传输耗时。

为了使得 Table 能够有相同的数据分布，同一 CG 内的 Table 必须保证以下属性相同

CREATE TABLE `tbl1`( 
  `k1` date NOT NULL COMMENT "", 
  `k2` int(11) NOT NULL COMMENT "", 
  `v1` int(11) SUM NOT NULL COMMENT "" 
)
ENGINE=OLAP 
AGGREGATE KEY(`k1`,`k2`) 
PARTITION BY RANGE(`k1`) ( 
  PARTITION p1 VALUES LESS THAN ('2019-05-31'), 
  PARTITION p2 VALUES LESS THAN ('2019-06-30') 
) 
DISTRIBUTED BY HASH(`k2`) BUCKETS 8 
PROPERTIES( 
  "colocate_with"="group1" 
); 

CREATE TABLE `tbl2`( 
  `k1` datetime NOT NULL COMMENT "", 
  `k2` int(11) NOT NULL COMMENT "", 
  `v1` double SUM NOT NULL COMMENT "" 
)
ENGINE=OLAP 
AGGREGATE KEY(`k1`,`k2`) 
DISTRIBUTED BY HASH(`k2`) BUCKETS 8 
PROPERTIES(
 "colocate_with"="group1" 
);

查看查询计划，如果 Colocation Join 生效，则 Hash Join 节点会显示 colocate: true。

DESC SELECT * FROM tbl1 INNER JOIN tbl2 ON ( tbl1.k2 = tbl2.k2);

动态分区

在某些使用场景下，用户会将表按照天进行分区划分，每天定时执行例行任务，这时需要使用方手动管理分区，否则，可能由于使用方没有创建分区导致数据导入失败，这给使用方带来了额外的维护成本。

在实现方式上, FE会启动一个后台线程，根据 fe.conf 中 dynamic_partition_enable 及 dynamic_partition_check_interval_seconds 参数决定该线程是否启动以及该线程的调度频率。每次调度时，会在注册表中读取动态分区表的属性，并根据动态分区属性动态添加及删除分区

CREATE TABLE example_db.dynamic_partition ( 
  k1 DATE, 
  k2 INT, 
  k3 SMALLINT, 
  v1 VARCHAR(2048), 
  v2 DATETIME DEFAULT "2023-02-0415:36:00" 
) ENGINE=olap 
DUPLICATE KEY(k1,k2,k3) 
PARTITION BY RANGE(k1) ( 
  PARTITION p20230821 VALUES LESS THAN ("2023-08-22"), 
  PARTITION p20230822 VALUES LESS THAN ("2023-08-23"),
  PARTITION p20230823 VALUES LESS THAN ("2023-08-24"), 
  PARTITION p20230824 VALUES LESS THAN ("2023-08-25") 
)
DISTRIBUTED BY HASH(k2) BUCKETS 32 
PROPERTIES( 
  "storage_medium"="SSD", 
  "dynamic_partition.enable"="true", // 是否开启动态分区特性，可指定为 TRUE 或 FALSE。如果不填写，默认为 TRUE。
  "dynamic_partition.time_unit"="DAY", //动态分区调度的单位，可指定为 DAY WEEK MONTH
  "dynamic_partition.start"="-3", //动态分区的开始时间, 以当天为基准，超过该时间范围的分区将会被删除
  "dynamic_partition.end"="3", //动态分区的结束时间, 以当天为基准，会提前创建N个单位的分区范围
  "dynamic_partition.prefix"="p", //动态创建的分区名前缀
  "dynamic_partition.buckets"="32" //动态创建的分区所对应的分桶数量
)

创建一张动态分区表，指定开启动态分区特性，以当天为2023-08-25为例，在每次调度时，会删除分区上界小于 2023-08-22 的分区，为了避免删除非动态创建的分区，动态删除分区只会删除分区名符合动态创建分区规则的分区，例如分区名为a1, 则即使分区范围在待删除的分区范围内，也不会被删除。同时在调度时会提前创建今天以及以后3天（总共4天）的分区(若分区已存在则会忽略)，分区名根据指定前缀分别为p20230825、p20230826、p20230827、p20230828,每个分区的分桶数量为32。同时会删除 p20230821 的分区

物化视图

官网物化视图demo
物化视图是将预先计算（根据定义好的 SELECT 语句）好的数据集，存储在 Doris 中的一个特殊的表。

物化视图的出现主要是为了满足用户，既能对原始明细数据的任意维度分析，也能快速的对固定维度进行分析查询。

在没有物化视图功能之前，用户一般都是使用 Rollup 功能通过预聚合方式提升查询效率的。但是 Rollup 具有一定的局限性，他不能基于明细模型做预聚合。

物化视图则在覆盖了 Rollup 的功能的同时，还能支持更丰富的聚合函数。所以物化视图其实是 Rollup 的一个超集

也就是说，之前 ALTER TABLE ADD ROLLUP 语法支持的功能现在均可以通过 CREATE MATERIALIZED VIEW 实现

CREATE MATERIALIZED VIEW < MV name > as < query >
[PROPERTIES ("key" = "value")]

为什么doris可以单机数万QPS？

单机数万QPS
对于高并发查询，其核心在于如何平衡有限的系统资源消耗与并发执行带来的高负载。换而言之，需要最大化降低单个 SQL 执行时的 CPU、内存和 IO 开销，其关键在于减少底层数据的 Scan 以及随后的数据计算，其主要优化方式有如下几种：

• 分区分桶裁剪
  Apache Doris 采用两级分区，第一级是 Partition，通常可以将时间作为分区键。第二级为 Bucket，通过 Hash 将数据打散至各个节点中，以此提升读取并行度并进一步提高读取吞吐。通过合理地划分区分桶，可以提高查询性能，以下列查询语句为例

select * from user_table where id=5122 and create_date='2022-01-01'

用户以 create_time 作为分区键、ID 作为分桶键，并设置了 16 个 Bucket， 经过分区分桶裁剪后可快速过滤非必要的分区数据，最终只需读取极少数据，比如 1 个分区的 1 个 Bucket 即可快速定位到查询结果，最大限度减少了数据的扫描量、降低了单个查询的延时

• 索引
  除了分区分桶裁剪， Doris 还提供了丰富的索引结构来加速数据的读取和过滤。
  索引的类型大体可以分为 智能索引 和 二级索引 两种，其中智能索引是在 Doris 数据写入时自动生成的，无需用户干预

智能索引

1. 前缀稀疏索引（Sorted Index）是建立在排序结构上的一种索引。Doris 存储在文件中的数据，是按照排序列有序存储的，Doris 会在排序数据上每 1024 行创建一个稀疏索引项。索引的 Key 即当前这 1024 行中第一行的前缀排序列的值，当用户的查询条件包含这些排序列时，可以通过前缀稀疏索引快速定位到起始行。
2. ZoneMap 索引是建立在 Segment 和 Page 级别的索引。对于 Page 中的每一列，都会记录在这个 Page 中的最大值和最小值，同样，在 Segment 级别也会对每一列的最大值和最小值进行记录。这样当进行等值或范围查询时，可以通过 MinMax 索引快速过滤掉不需要读取的行。

二级索引

需要用手动创建的索引，包括 Bloom Filter 索引、Bitmap 索引，以及 2.0 版本新增的 Inverted 倒排索引 和 NGram Bloom Filter 索引
doris官网索引介绍

以下列查询语句为例：

select * from user_table where id > 10 and id < 1024

假设按照 ID 作为建表时指定的 Key， 那么在 Memtable 以及磁盘上按照 ID 有序的方式进行组织，查询时如果过滤条件包含前缀字段时，则可以使用前缀索引快速过滤。Key 查询条件在存储层会被划分为多个 Range，按照前缀索引做二分查找获取到对应的行号范围，由于前缀索引是稀疏的，所以只能大致定位出行的范围。随后过一遍 ZoneMap、Bloom Filter、Bitmap 等索引，进一步缩小需要 Scan 的行数。通过索引，大大减少了需要扫描的行数，减少 CPU 和 IO 的压力，整体大幅提升了系统的并发能力

物化视图

物化视图是将预先计算（根据定义好的 SELECT 语句）好的数据集，存储在 Doris 中的一个特殊的表。

物化视图是一种典型的空间换时间的思路，其本质是根据预定义的 SQL 分析语句执⾏预计算，并将计算结果持久化到另一张对用户透明但有实际存储的表中。在需要同时查询聚合数据和明细数据以及匹配不同前缀索引的场景，命中物化视图时可以获得更快的查询相应，同时也避免了大量的现场计算，因此可以提高性能表现并降低资源消耗

doris物化视图创建

Runtime Filter

Doris 中还额外加入了动态过滤机制，即 Runtime Filter。

1. 在多表关联查询时，我们通常将右表称为 Build Table、左表称为 Probe Table，左表的数据量会大于右表的数据
2. 在实现上，会首先读取右表的数据，在内存中构建一个 HashTable（Build）。之后开始读取左表的每一行数据，并在 HashTable 中进行连接匹配，来返回符合连接条件的数据（Probe）
3. 而 Runtime Filter 是在右表构建 HashTable 的同时，为连接列生成一个过滤结构，可以是 Min/Max、IN 等过滤条件。之后把这个过滤列结构下推给左表。这样一来，左表就可以利用这个过滤结构，对数据进行过滤，从而减少 Probe 节点需要传输和比对的数据量
4. 在大多数 Join 场景中，Runtime Filter 可以实现节点的自动穿透，将 Filter 穿透下推到最底层的扫描节点或者分布式 Shuffle Join 中

整个过程如下：
当前存在T1表与T2表的Join查询，它的Join方式为HashJoin，T1是一张事实表，数据行数为100000，T2是一张维度表，数据行数为2000

|          >      HashJoinNode     <
|         |                         |
|         | 100000                  | 2000
|         |                         |
|   OlapScanNode              OlapScanNode
|         ^                         ^   
|         | 100000                  | 2000
|        T1                        T2

显而易见对T2扫描数据要远远快于T1，如果我们主动等待一段时间再扫描T1，等T2将扫描的数据记录交给HashJoinNode后，HashJoinNode根据T2的数据计算出一个过滤条件，比如T2数据的最大和最小值，或者构建一个Bloom Filter，接着将这个过滤条件发给等待扫描T1的ScanNode，后者应用这个过滤条件，将过滤后的数据交给HashJoinNode，从而减少probe hash table的次数和网络开销，这个过滤条件就是Runtime Filter，效果如下

|          >      HashJoinNode     <
|         |                         |
|         | 6000                    | 2000
|         |                         |
|   OlapScanNode              OlapScanNode
|         ^                         ^   
|         | 100000                  | 2000
|        T1                        T2

如果能将过滤条件（Runtime Filter）下推到存储引擎，则某些情况下可以利用索引来直接减少扫描的数据量，从而大大减少扫描耗时，效果如下

|          >      HashJoinNode     <
|         |                         |
|         | 6000                    | 2000
|         |                         |
|   OlapScanNode              OlapScanNode
|         ^                         ^   
|         | 6000                    | 2000
|        T1                        T2

doris官网 Runtime Filter介绍

TOPN 优化技术

在数据库中查询最大或最小几条数据的应用场景非常广泛，比如查询满足某种条件的时间最近 100 条数据、查询价格最高或者最低的几个商品等，此类查询的性能对于实时分析非常重要。在 Doris 中引入了 TOPN 优化来解决大数据场景下较高的 IO、CPU、内存资源消耗

Apache Doris 2.0 新特性

解决在超高并发要求（例如数万 QPS）的 Data Serving 场景中，doris存在的瓶颈：

• 列式存储引擎对于行级数据的读取不友好，宽表模型上列存格式将大大放大随机读取 IO
• OLAP 数据库的执行引擎和查询优化器对于某些简单的查询（如点查询）来说太重，需要在查询规划中规划短路径来处理此类查询
• SQL 请求的接入以及查询计划的解析与生成由 FE 模块负责，使用的是 Java 语言，在高并发场景下解析和生成大量的查询执行计划会导致高 CPU 开销

行式存储格式（Row Store Format）

与列式存储格式不同，行式存储格式在数据服务场景会更加友好，数据按行存储、应对单次检索整行数据时效率更高，可以极大减少磁盘访问次数

在 Apache Doris 2.0 版本中，引入了行式存储格式，将行存编码后存在单独的一列中，通过额外的空间来存储

行存模式默认是关闭的，如果您想开启则可以在建表语句的 property 中指定如下属性

"store_row_column" = "true"

doris官网行存demo

点查询短路径优化（Short-Circuit）

通常情况下，一条 SQL 语句的执行需要经过三个步骤：

1. 首先通过 SQL Parser 解析语句，生成抽象语法树(AST)
2. 随后通过 Query Optimizer 生成可执行计划（Plan）
3. 最终通过执行该计划得到计算结果

对于大数据量下的复杂查询，经由查询优化器生成的执行计划无疑具有更高效的执行效果，但对于低延时和高并发要求的点查询，则不适宜走整个查询优化器的优化流程，会带来不必要的额外开销

为了解决这个问题，doris 实现了点查询的短路径优化，绕过查询优化器以及 PlanFragment 来简化 SQL 执行流程，直接使用快速高效的读路径来检索所需的数据

点查询短路径优化.png

当查询被 FE 接收后，它将由规划器生成适当的 Short-Circuit Plan 作为点查询的物理计划。该 Plan 非常轻量级，不需要任何等效变换、逻辑优化或物理优化，仅对 AST 树进行一些基本分析、构建相应的固定计划并减少优化器的开销。

对于简单的主键点查询，如select * from tbl where pk1 = 123 and pk2 = 456，因为其只涉及单个 Tablet，因此可以使用轻量的 RPC 接口来直接与 StorageEngine 进行交互，以此避免生成复杂的Fragment Plan 并消除了在 MPP 查询框架下执行调度的性能开销

预处理语句优化（PreparedStatement）

高并发查询中的 CPU 开销，可以部分归因于 FE 层分析和解析 SQL 的 CPU 计算，为了解决这个问题，我们在 FE 端提供了与 MySQL 协议完全兼容的预处理语句（Prepared Statement）
当 CPU 成为主键点查的性能瓶颈时，Prepared Statement 可以有效发挥作用，实现 4 倍以上的性能提升

1. Prepared Statement 的工作原理是通过在 Session 内存 HashMap 中缓存预先计算好的 SQL 和表达式，在后续查询时直接复用缓存对象即可

2. Prepared Statement 使用 MySQL 二进制协议 作为传输协议。该协议在文件mysql_row_buffer.[h|cpp]中实现，符合标准 MySQL 二进制编码。
   通过该协议客户端例如 JDBC Client， 第一阶段发送PREPAREMySQL Command 将预编译语句发送给 FE 并由 FE 解析、Analyze 该语句并缓存到上图的 HashMap 中，接着客户端通过EXECUTEMySQL Command 将占位符替换并编码成二进制的格式发送给 FE， 此时 FE 按照 MySQL 协议反序列化后得到占位符中的值，生成对应的查询条件

3. 除了在 FE 缓存 Statement，我们还需要在 BE 中缓存被重复使用的结构，包括预先分配的计算 Block，查询描述符和输出表达式，由于这些结构在序列化和反序列化时会造成 CPU 热点， 所以需要将这些结构缓存下来。对于每个查询的 PreparedStatement，都会附带一个名为 CacheID 的 UUID。当 BE 执行点查询时，根据相关的 CacheID 找到对应的复用类， 并在 BE 中表达式计算、执行时重复使用上述结构

PrepareStatement.png

doris官网使用PreparedStatement的demo

Doris Manager 引入

Cluster Manager for Apache Doris（以下简称 Doris Manager）提供的主要功能如下：
部署集群 ：通过 Doris Manager 在物理机、虚拟机部署 Aache Doris 集群
接管集群 ：将现有的 Apache Doris 集群 接管到 Doris Manager 进行运维、监控
集群详情 ：查看集群的运行状态、详情、连接信息
集群扩缩 ：对 FE、BE 节点进行扩容、缩容
集群升级 ：便捷的集群版本升级，提供全量停服升级和在线滚动升级，可根据业务场景选择适合的升级方式
集群重启 ：对整个集群、FE、BE 以及节点进行重启操作
节点详情 ：查看节点的实时状态以及机器信息
监控告警 ：查看预制的监控指标，以及设置监控指标的告警，并通过邮件、聊天软件、Webhook 等方式进行告警
参数配置 ：关键参数的展示及配置，可单个或批量进行参数修改。
集群巡检 ：一键检查机器状况、集群运行状态，及时发现并定位性能瓶颈，并提供修复建议。
WebUI ：进入集群 WebUI 的入口

Doris Manager使用

为了解决手动部署doris的烦恼，以及拥有一个可视化的界面，我们尝试使用Doris Manager来接管现有集群

准备工作：

1. Doris Manager 安装包
2. Doris 安装包
3. 配置 ssh 登录，涉及服务间通讯

安装Doris Manager（以doris-manager-2.4.0版本为例）

第一步，将Manager的安装包上传到指定目录并解压

# 进入到manager的存放目录
$ tar -zxvf doris-manager-2.4.0-x64_avx2-bin.tar.gz

解压完的目录层级

doris-manager-2.4.0-x64_avx2-bin
   config-tool // 安装manager套件工具的二进制工具
   manager // doris 管控服务包
       sdbctrl // 管控命令行组件
           bin
           conf
       webserver // 管控服务器
           bin
           conf
           lib
           static
       deps // 管控依赖包
           alertmanager // 告警工具
           jdk // jdk依赖包
           prometheus // 指标存储
           grafana // 监控看板
              doris-overview_rev5.json // 默认仪表盘json文件，名称以实际为主
   studio // doris 查询工具包
       bin
       conf
       lib
       static

安装Doris Manager主要依赖 config-tool ，为了后续方便使用config-tool，可以选择将config-tool移动到linux的$PATH路径下

cp config-tool /usr/local/bin
cp config-tool /usr/bin

查看 config-tool 版本

$ config-tool -v
build_version: 23.7.1
build_time:  2023-08-26T23:07:43+0800

生成配置文件模板，doris manager 启动的配置，以及安装路径

# 直接查看原始模板
config-tool template

# 将模板生成到文件(名称任意)
config-tool template > init.conf

配置文件详情如下：

# template of conf for config-tool
base_deploy_dir: /opt/doris-manager           # doris-manager's deploy dir
webserver:
  version: 1.0.0                              # target installed doris-manager version, eg. 2.3.0
  host: 10.20.30.40                           # localhost or 127.0.0.1 is not suggested
  port: 8004                                  # default webserver port, can change it
sdbctrl:
  port: 8003                                  # default sdbctrl port, can change it
grafana:
  http_port: 3000                             # default grafana port, can change it
prometheus:
  http_port: 9090                             # default grafana port, can change it
alert_manager:
  web_port: 9093                              # default grafana port, can change it
# db has only two types: 'h2' 'mysql'
# Example 1: h2
db:
  type: h2
  data_path: /opt/manager-data                # data.mv.db file's location
# Example 2: mysql
# db:
#   type: mysql
#   host: 127.0.0.1                           # mysql host
#   port: 3306                                # mysql port
#   user: root                                # mysql user
#   password: doris                           # mysql password
#   name: manager                             # mysql database: manager (should be created first)
# mail config for alerting
mail_config:
  host: mailserver.com
  username: name@company.com
  password: Pass123456
  port: 25
  protocol: smtp
# doris and manger tar package path
path:
  doris: /opt/downloads/doris                 # eg. doris/selectdb tar package path
  webui: /opt/downloads/webui                 # eg. manager tar package path

核心配置

• base_deploy_dir字段，doris manager服务整体的目标部署路径，默认为/opt/doris-manager，用户可修改为实际部署路径，部署的过程会将doris-manager解压包中的文件拷贝到 base_deploy_dir目录
• webserver.version 字段，应该修改为需要部署的 doris-manager版本，比如23.7.1
• webserver.host 字段，应该修改为 本机的IP ，不推荐填写127.0.0.1或者localhost, 如果用户使用的是云上主机，最好使用公网IP（如果想在本地浏览器访问云主机部署的管控服务平台时，使用内网IP等其他非公网的方案可能会导致grafana监控数据获取失败）
• path.doris字段，应该修改为具体存放doris安装包所在的绝对路径，默认为/opt/downloads/doris
• path.webui字段，应该修改为doris-manager安装包所在的绝对路径，默认为/opt/downloads/webui (如果用户一开始下载的doris-manager安装包不在此目录需要拷贝到该目录下)

开始部署

config-tool deploy -f init.conf

成功的命令行

[root@p12180v bin]# config-tool deploy -f init.conf
INFO[0002] [webserver] host = 10.209.34.142             
INFO[0002] [webserver] deploy_dir = /opt/doris-manager/manager/webserver 
INFO[0002] [database] type = h2, data_path = /opt/manager-data 
INFO[0002] [sdbctrl] deploy_dir = /opt/doris-manager/manager/sdbctrl 
INFO[0002] [prometheus] deploy_dir = /opt/doris-manager/manager/deps/prometheus 
INFO[0002] [alertmanager] deploy_dir = /opt/doris-manager/manager/deps/alertmanager 
INFO[0002] [grafana] deploy_dir = /opt/doris-manager/manager/deps/grafana 
INFO[0002] [path] doris package dir = /data/doris-2.0.0 
INFO[0002] [path] webui package dir = /data/doris-manager-2.4.0 
INFO[0031] [grafana] ready to start ...                 
INFO[0077] [grafana] grafana is up and running!         
INFO[0077] [grafana] start successfully, on port [3000] 
INFO[0077] [grafana] ready to add datasource...         
INFO[0078] [grafana] add datasource successfully        
INFO[0078] [grafana] ready to import dashboard...       
INFO[0078] [grafana] import dashboard done              
INFO[0078] [alertmanager] ready to configure...         
INFO[0078] [alertmanager] configure successfully        
INFO[0078] [alertmanager] ready to start...             
INFO[0081] [alertmanager] alertmanager is up and healthy! 
INFO[0081] [alertmanager] start successfully, on port [9093] 
INFO[0081] [prometheus] ready to configure...           
INFO[0081] [prometheus] configure successfully          
INFO[0081] [prometheus] ready to starting...            
INFO[0084] [prometheus] prometheus is up and healthy!   
INFO[0084] [prometheus] start successfully, on port [9090] 
INFO[0084] [sdbctrl] ready to configure...              
INFO[0084] [sdbctrl] configure successfully.            
INFO[0084] [sdbctrl] ready to start...                  
INFO[0087] [sdbctrl] start successfully, on port [8003] 
INFO[0087] [webserver] ready to configure...            
INFO[0087] [webserver] configure successfully           
INFO[0087] [webserver] ready to start...                
INFO[0110] [webserver] start successfully, on port [8004] 
INFO[0110] deploy doris manager successfully !!!        
INFO[0110] notice here! manager meta data is saved in [~/.config-tool/meta.yaml] 
INFO[0110] now you can visit [http://10.209.34.142:8004] to use doris-manager platform 

也可以通过 cat ~/.config-tool/meta.yaml 查看当前manager运行的情况

[root@p12180v bin]# cat ~/.config-tool/meta.yaml
deploy_status: DONE
base_deploy_dir: /opt/doris-manager/manager
webserver:
  version: 23.7.1
  host: 10.209.34.142
  port: 8004
  deploy_dir: /opt/doris-manager/manager/webserver
  status: RUNNING
sdbctrl:
  host: 127.0.0.1
  port: 8003
  deploy_dir: /opt/doris-manager/manager/sdbctrl
  log: /opt/doris-manager/manager/sdbctrl/log
  status: RUNNING
prometheus:
  host: 127.0.0.1
  http_port: 9090
  deploy_dir: /opt/doris-manager/manager/deps/prometheus
  data_dir: /opt/doris-manager/manager/deps/prometheus/data
  alert_dir: /opt/doris-manager/manager/deps/prometheus/alert
  log_dir: /opt/doris-manager/manager/deps/prometheus/log
  status: RUNNING
grafana:
  host: 127.0.0.1
  http_port: 3000
  deploy_dir: /opt/doris-manager/manager/deps/grafana
  dashboard_uuid: 3fFiWJ4mz123
  datasource_name: DS_SELECTDB_ENTERPRISE
  dashboard_tpl_name: SelectDB-Enterprise-Overview
  log_dir: /opt/doris-manager/manager/deps/grafana/log
  status: RUNNING
alert_manager:
  host: 127.0.0.1
  web_port: 9093
  deploy_dir: /opt/doris-manager/manager/deps/alertmanager
  data_dir: /opt/doris-manager/manager/deps/alertmanager/data
  log_dir: /opt/doris-manager/manager/deps/alertmanager/log
  status: RUNNING
db:
  type: h2
  data_path: /opt/manager-data
mail_config:
  host: mailserver.com
  username: name@company.com
  password: Pass123456
  port: 25
  protocol: smtp
path:
  doris: /data/doris-2.0.0
  webui: /data/doris-manager-2.4.0

Manager的扩容、缩容注意点

1. Manager的集群管理，不会配置自动重启功能
2. 针对FE扩容时，首次启动需要依赖Master节点，并且，需要手工维护自动重启脚本
3. Manager管理集群依赖 doris 的安装包，版本一定要一致，否则，会出现兼容性问题

doris_manager工作台.jpg

部署过程中异常处理

1. ERROR com.selectdb.enterprise.manager.common.dorisctrl.client.APIClient - server execute failed: status = 500, response = {
    "code": 1,
    "message": "failed to import cluster, broker_ipc_port is 0"
}

安装了 BE 的服务，对应broker 也得安装

接管集群的时候，使用 ssh 免密登录的方式，否则，会有账号权限问题

failed to import cluster, get no PE pid using command [netstat -nltp 2>/dev/null | grep '/java' | grep ':8030'],check if user [xxx] has permission

启动过程中某些组件失败，主要检查端口占用情况，这部分问题通过重启、重装解决问题

INFO[0026] [grafana] ready to start ...                 
ERRO[0034] deploy failed: failed to connect to Grafana: Get "http://127.0.0.1:3100/api/health": context deadline exceeded (Client.Timeout exceeded while awaiting headers)