有部分网站打不开,龙湖镇华南城网站建设,网站有备案 为企业,网站建设类文章要发多少片1、Doris简介
1.1、doris概述
Apache Doris 由百度大数据部研发#xff08;之前叫百度 Palo#xff0c;2018 年贡献到 Apache 社区后#xff0c;
更名为 Doris #xff09;#xff0c;在百度内部#xff0c;有超过 200 个产品线在使用#xff0c;部署机器超过 1000 台…1、Doris简介
1.1、doris概述
Apache Doris 由百度大数据部研发之前叫百度 Palo2018 年贡献到 Apache 社区后
更名为 Doris 在百度内部有超过 200 个产品线在使用部署机器超过 1000 台单一
业务最大可达到上百 TB。
Apache Doris 是一个现代化的 MPPMassively Parallel Processing即大规模并行处理
分析型OLAP数据库产品。仅需亚秒级响应时间即可获得查询结果有效地支持实时数据分析。
Apache Doris 的分布式架构非常简洁易于运维并且可以支持 10PB 以上的超大数据集。
Apache Doris 可以满足多种数据分析需求例如固定历史报表实时数据分析交互式数据分析和探索式数据分析等。
1.2、OLAP和OLTP的区别
联机事务处理OLTP(On-Line Transaction Processing)
公司业务系统使用数据库的场景针对业务系统数据库有大量随机的增删改查高并发速度要快支持事务 联机分析处理OLAP(On-Line Analytical Processing) 公司的数据分析使用数据库的场景对已经生成好的数据进行统计分析
一次操作都是针对的整个数据集只有查这个动作不会去增删改查询的响应速度相对慢点也能接受并发量要求不是太高 常见的OLAP引擎 1.3、使用场景
1、报表分析
实时看板 Dashboards 》rolap中我给过他一个sql亚秒级的响应我想要的结果面向企业内部分析师和管理者的报表面向用户或者客户的高并发报表分析Customer Facing Analytics。比如面向网站主的站点分析、面向广告主的广告报表并发通常要求成千上万的 QPS 查询延时要求毫秒级响应。著名的电商公司京东在广告报表中使用 Apache Doris 每天写入 100 亿行数据查询并发 QPS 上万99 分位的查询延时 150ms。
2、即时查询Ad-hoc Query 面向分析师的自助分析查询模式不固定要求较高的吞吐。小米公司基于 Doris 构建了增长分析平台Growing AnalyticsGA利用用户行为数据对业务进行增长分析平均查询延时 10s95 分位的查询延时 30s 以内每天的 SQL 查询量为数万条。
3、统一数仓构建 一个平台满足统一的数据仓库建设需求简化繁琐的大数据软件栈。海底捞基于 Doris 构建的统一数仓替换了原来由 Spark、Hive、Hbase、Phoenix 组成的旧架构架构大大简化。
4、数据湖联邦查询 通过外表的方式联邦分析位于 Hive、Hudi 中的数据在避免数据拷贝的前提下查询性能大幅提升
1.4、doris优势 1.5、架构
Doris 的架构很简洁只设 FE(Frontend)前端进程、BE(Backend)后端进程两种角色、两个后台的服务进程不依赖于外部组件方便部署和运维FE、BE 都可在线性扩展。
1、FEFrontend存储、维护集群元数据负责接收、解析查询请求规划查询计划,调度查询执行返回查询结果。
主要有三个角色
Leader 和 Follower主要是用来达到元数据的高可用保证单节点宕机的情况下,元数据能够实时地在线恢复而不影响整个服务。Observer用来扩展查询节点同时起到元数据备份的作用。如果在发现集群压力非常大的情况下需要去扩展整个查询的能力那么可以加 observer 的节点。observer 不参与任何的写入只参与读取。
2、BEBackend负责物理数据的存储和计算依据 FE 生成的物理计划分布式地执行查询。数据的可靠性由 BE 保证BE 会对整个数据存储多副本或者是三副本。副本数可根据需求动态调整。
3、MySQL Client:Doris 借助 MySQL 协议用户使用任意 MySQL 的 ODBC/JDBC 以及 MySQL 的客户端都可以直接访问 Doris。
4、Broker:一个独立的无状态进程。封装了文件系统接口提供 Doris 读取远端存储系统中文件的能力包括 HDFSS3BOS 等。 1.6、默认端口 2、Doris安装
2.1、准备 操作系统环境要求 设置系统最大文件打开句柄数 启动一个程序的时候打开文件的数量就是句柄数
1.打开文件
security /sɪˈkjʊərəti/
vi /etc/security/limits.conf
2.在文件最后添加下面几行信息(注意* 也要赋值进去)* soft nofile 65535
* hard nofile 65535
* soft nproc 65535
* hard nproc 65535ulimit -n 65535 临时生效修改完文件后需要重新启动虚拟机
重启永久生效也可以用 。如果不修改这个句柄数大于等于60000回头启动doris的be节点的时候就会报如下的错
File descriptor number is less than 60000. Please use (ulimit -n) to set a value equal or greater than 60000
W1120 18:14:20.934705 3437 storage_engine.cpp:188] check fd number failed, error: Internal error: file descriptors limit is too small
W1120 18:14:20.934713 3437 storage_engine.cpp:102] open engine failed, error: Internal error: file descriptors limit is too small
F1120 18:14:20.935087 3437 doris_main.cpp:404] fail to open StorageEngine, resfile descriptors limit is too small时钟同步 Doris 的元数据要求时间精度要小于5000ms所以所有集群所有机器要进行时钟同步避免因为时钟问题引发的元数据不一致导致服务出现异常。
如何时间同步
首先安装 ntpdate
# ntpdate是一个向互联网上的时间服务器进行时间同步的软件
[rootzuomm01 doris]# yum install ntpdate -y然后开始三台机器自己同步时间[rootnode01 ~]# ntpdate ntp.sjtu.edu.cn美国标准技术院时间服务器time.nist.gov192.43.244.18
上海交通大学网络中心NTP服务器地址ntp.sjtu.edu.cn202.120.2.101
中国国家授时中心服务器地址cn.pool.ntp.org210.72.145.44# 将当前时间写入bios这样才能永久生效不变不然reboot后还会恢复到原来的时间
clock -w 关闭交换分区swap
交换分区是linux用来当做虚拟内存用的磁盘分区
linux可以把一块磁盘分区当做内存来使用虚拟内存、交换分区
Linux使用交换分区会给Doris带来很严重的性能问题建议在安装之前禁用交换分区
1、查看 Linux 当前 Swap 分区
free -m
2、关闭 Swap 分区
swapoff -a[rootzuomm01 app]# free -mtotal used free shared buff/cache available
Mem: 5840 997 4176 9 666 4604
Swap: 6015 0 6015
[rootzuomm01 app]# swapoff -a3.验证是否关闭成功
[rootzuomm01 app]# free -m total used free shared buff/cache available
Mem: 5840 933 4235 9 671 4667
Swap: 0 0 0注意事项
1、FE 的磁盘空间主要用于存储元数据包括日志和 image。通常从几百 MB 到几个GB 不等。
2、BE 的磁盘空间主要用于存放用户数据总磁盘空间按用户总数据量* 33 副本计算然后再预留额外 40%的空间用作后台 compaction 以及一些中间数据的存放。
3、一台机器上可以部署多个 BE 实例但是只能部署一个 FE。如果需要 3 副本数 据那么至少需要 3 台机器各部署一个 BE 实例而不是 1 台机器部署 3 个 BE 实例。多 个 FE 所在服务器的时钟必须保持一致允许最多 5 秒的时钟偏差
4、测试环境也可以仅适用一个 BE 进行测试。实际生产环境BE 实例数量直接决定了整体查询延迟。
5、所有部署节点关闭 Swap。
6、FE 节点数据至少为 11 个 Follower。当部署 1 个 Follower 和 1 个 Observer 时,可以实现读高可用。当部署 3 个 Follower 时可以实现读写高可用HA。
7、Follower 的数量必须为奇数Observer 数量随意。
8、根据以往经验当集群可用性要求很高时比如提供在线业务可以部署 3 个Follower 和 1-3 个 Observer。如果是离线业务建议部署 1 个 Follower 和 1-3 个 Observer。
9、Broker 是用于访问外部数据源如 HDFS的进程。通常在每台机器上部署一个 broker 实例即可。
2.2、安装FE
1、去官网下载源码包官网地址https://doris.apache.org 2、根据自己的配置选择性点击下载 3、上传到linux解压修改配置文件
-- 去自己的路劲中找到fe.conf文件
vi /opt/app/doris/fe/conf/fe.conf
#配置文件中指定元数据路径 注意这个文件夹要自己创建
meta_dir /opt/data/dorisdata/doris-meta
#修改绑定 ip每台机器修改成自己的 ip
priority_networks 192.168.17.0/24 4、分发集群
[rootzuomm01 app]# for i in 2 3 doscp /et/profile linux0$i:/etc/profilescp -r /opt/app/doris/ linux0$i:/opt/app/done5、启动
进入到fe的bin目录下执行
[rootzuomm01 bin]# ./start_fe.sh --daemon生产环境强烈建议单独指定目录不要放在 Doris 安装目录下最好是单独的磁盘如果有 SSD 最好。如果机器有多个 ip, 比如内网外网, 虚拟机 docker 等, 需要进行 ip 绑定才能正确识别。JAVA_OPTS 默认 java 最大堆内存为 4GB建议生产环境调整至 8G 以上。
2.3、安装BE
1、进入到be的conf目录下修改配置文件
vi be.conf #配置文件中指定数据存放路径
storage_root_path /opt/data/dorisdata/bedata#修改绑定 ip每台机器修改成自己的 ip
priority_networks 192.168.17.0/24 注意事项
storage_root_path 默认在 be/storage 下需要手动创建该目录。多个路径之间使用英文状态的分号;分隔最后一个目录后不要加。
可以通过路径区别存储目录的介质HDD 或 SSD。可以添加容量限制在每个路径的末尾通过英文状态逗号隔开如
storage_root_path/home/disk1/doris.HDD,50;/home/disk2/doris.SSD,10;/home/disk2/doris
说明
/home/disk1/doris.HDD,50表示存储限制为 50GBHDD;
/home/disk2/doris.SSD,10存储限制为 10GBSSD
/home/disk2/doris存储限制为磁盘最大容量默认为 HDD
这样就好了嘛不是哦
因为FE和BE两个都是单独的个体所以他俩相互间还不认识就需要我们通过mysql的客户端将他们建立起联系
2、装mysql
-- 安装yum
0 yum list
-- 安装wget
1 yum -y install wget
-- 让wget直接去网页上安装mysql的安装包可以解决一些依赖问题
2 wget -i -c http://dev.mysql.com/get/mysql57-community-release-el7-10.noarch.rpm
-- 安装刚才下载的mysql安装包
3 yum -y install mysql57-community-release-el7-10.noarch.rpm
4-- 安装mysql的服务
yum -y install mysql-community-server
这边有时候会报错
Failing package is: mysql-community-server-5.7.37-1.el7.x86_64
GPG Keys are configured as: file:///etc/pki/rpm-gpg/RPM-GPG-KEY-mysql
解决办法 在yum install 版本后面加上 --nogpgcheck即可绕过GPG验证成功安装。
比如yum install mysql-community-server --nogpgcheck
亲测没啥问题安装成功
-- 启动mysql
5 service mysqld start
-- 查看mysql的3306端口确认下mysql的服务是否打开
6 netstat -nltp | grep 3306
-- 去mysql的log日志中查找他的初始密码
7 grep password /var/log/mysqld.log 查看原始密码
grep password /var/log/mysqld.log
2020-06-24T07:21:25.731630Z 1 [Note] A temporary password is generated for rootlocalhost: Apd;WYEc2Ir
-- 注释:在密码不是以空格开头的
-- 如果登录不进去的情况下,可以尝试 mysql -u root -p 然后回车在将密码复制进去
2020-06-24T07:21:48.097350Z 2 [Note] Access denied for user rootlocalhost (using password: NO)
-- 登录mysql
8 登录 mysql -uroot -pWYEc2Ir
-- 修改mysql的密码首先设置两个参数这样密码就可以设置成123456这种简单的密码
9 修改密码mysql set global validate_password_policy0;mysql set global validate_password_length1; 这个两个设置以后 密码很简单不会报错
-- 修改mysql的密码
alter user user() identified by XXXXXX; xxxx 就是你的新密码3、使用 MySQL Client 连接 FE
mysql -h zuomm01 -P 9030 -uroot这个只是用了mysql的客户端去连接doris的fe不是启动的mysql哦并且第一次进去的话是不需要密码的
解释
-h 连接地址
-P 端口号
-u 账号
-p 密码--这个可以设置可以不设置啦正常生产过程中都会设置一个相对比较复杂的密码学习的时候就无所谓了
--如果想设置下面的命令就可以
SET PASSWORD FOR root PASSWORD(123456);4、fe启动完成后可以查看fe的运行状态
SHOW PROC /frontends\G;5、添加 BE 节点
ALTER SYSTEM ADD BACKEND zuomm01:9050;
ALTER SYSTEM ADD BACKEND zuomm02:9050;
ALTER SYSTEM ADD BACKEND zuomm03:9050;6、查看BE状态
SHOW PROC /backends;Alive 为 false 表示该 BE 节点还是死的7、添加环境变量
#doris_fe
export DORIS_FE_HOME/opt/app/doris1.1.4/fe
export PATH$PATH:$DORIS_FE_HOME/bin#doris_be
export DORIS_BE_HOME/opt/app/doris1.1.4/be
export PATH$PATH:$DORIS_BE_HOME/bin8、启动BE
启动 BE每个节点
/opt/app/doris/be/bin/start_be.sh --daemon 启动后再次查看BE的节点
mysql -h zuomm01 -P 9030 -uroot -p 123456
SHOW PROC /backends;
Alive 为 true 表示该 BE 节点存活。2.4、部署 FS_Broker可选
Broker 以插件的形式独立于 Doris 部署。如果需要从第三方存储系统导入数据需要部署相应的 Broker默认提供了读取 HDFS、百度云 BOS 及 Amazon S3 的 fs_broker。fs_broker 是无状态的建议每一个 FE 和 BE 节点都部署一个 Broker。
直接启动就可以啦
1、启动 Broker
/opt/app/doris/fe/apache_hdfs_broker/bin/start_broker.sh --daemon 2、使用 mysql-client 连接启动的 FE执行以下命令
mysql -h zuomm01 -P 9030 -uroot -p 123456
ALTER SYSTEM ADD BROKER broker_name zuomm01:8000,zuomm02:8000,zuomm03:8000;当然你也可以一个个的加并且 broker_name 这只是一个名字可以自己取3、查看 Broker 状态
使用 mysql-client 连接任一已启动的 FE执行以下命令查看 Broker 状态
SHOW PROC /brokers; 2.5、扩容和缩容
1、FE 扩容和缩容
可以通过将 FE 扩容至 3 个以上节点来实现 FE 的高可用。
使用 MySQL 登录客户端后可以使用 sql 命令查看 FE 状态目前就一台 FE
mysql -h zuomm01 -P 9030 -uroot -p
mysql SHOW PROC /frontends\G;*************************** 1. row ***************************Name: 192.168.17.3_9010_1661510658077IP: 192.168.17.3HostName: zuomm01EditLogPort: 9010HttpPort: 8030QueryPort: 9030RpcPort: 9020Role: FOLLOWERIsMaster: trueClusterId: 1133836578Join: trueAlive: true
ReplayedJournalId: 2472LastHeartbeat: 2022-08-26 13:07:47IsHelper: trueErrMsg: Version: 1.1.1-rc03-2dbd70bf9CurrentConnected: Yes
1 row in set (0.03 sec)添加FE的新节点
FE 分为 LeaderFollower 和 Observer 三种角色。默认一个集群只能有一个 Leader,可以有多个 Follower 和 Observer。其中 Leader 和 Follower 组成一个 Paxos 选择组如果Leader 宕机则剩下的 Follower 会自动选出新的 Leader保证写入高可用。Observer 同步 Leader 的数据但是不参加选举。
1、如果只部署一个 FE则 FE 默认就是 Leader。在此基础上可以添加若干 Follower 和 Observer。
ALTER SYSTEM ADD FOLLOWER zuomm02:9010;
ALTER SYSTEM ADD OBSERVER zuomm03:9010;2、在zuomm02和zuomm03上分别启动FE节点
/opt/app/doris/fe/bin/start_fe.sh --helper zuomm01:9010 --daemon记住哦如果是第一次添加的话一定要加这两个参数 --helper zuomm01:9010 mysql SHOW PROC /frontends\G;
*************************** 1. row ***************************Name: 192.168.17.4_9010_1661490723344IP: 192.168.17.4HostName: zuomm02EditLogPort: 9010HttpPort: 8030QueryPort: 0RpcPort: 0Role: FOLLOWERIsMaster: falseClusterId: 1133836578Join: falseAlive: false
ReplayedJournalId: 0LastHeartbeat: NULLIsHelper: trueErrMsg: java.net.ConnectException: Connection refused (Connection refused)Version: NULLCurrentConnected: No
*************************** 2. row ***************************Name: 192.168.17.5_9010_1661490727316IP: 192.168.17.5HostName: zuomm03EditLogPort: 9010HttpPort: 8030QueryPort: 0RpcPort: 0Role: OBSERVERIsMaster: falseClusterId: 1133836578Join: falseAlive: false
ReplayedJournalId: 0LastHeartbeat: NULLIsHelper: falseErrMsg: java.net.ConnectException: Connection refused (Connection refused)Version: NULLCurrentConnected: No
*************************** 3. row ***************************Name: 192.168.17.3_9010_1661510658077IP: 192.168.17.3HostName: zuomm01EditLogPort: 9010HttpPort: 8030QueryPort: 9030RpcPort: 9020Role: FOLLOWERIsMaster: trueClusterId: 1133836578Join: trueAlive: true
ReplayedJournalId: 2577LastHeartbeat: 2022-08-26 13:13:33IsHelper: trueErrMsg: Version: 1.1.1-rc03-2dbd70bf9CurrentConnected: Yes
3 rows in set (0.04 sec)3、删除FE节点
ALTER SYSTEM DROP FOLLOWER[OBSERVER] fe_host:edit_log_port; ALTER SYSTEM DROP FOLLOWER zuomm01:9010; 2、BE扩容和缩容
1、增加 BE 节点
在 MySQL 客户端通过
ALTER SYSTEM ADD BACKEND 命令增加 BE 节点。
ALTER SYSTEM ADD BACKEND zuomm01:9050; 2、DROP 方式删除 BE 节点不推荐
ALTER SYSTEM DROP BACKEND be_host:be_heartbeat_service_port;
ALTER SYSTEM DROP BACKEND zuomm01:9050; 注意DROP BACKEND 会直接删除该 BE并且其上的数据将不能再恢复所以我们强烈不推荐使用 DROP BACKEND 这种方式删除 BE 节点。当你使用这个语句时,会有对应的防误操作提示。3、DECOMMISSION 方式删除 BE 节点推荐
ALTER SYSTEM DECOMMISSION BACKEND be_host:be_heartbeat_service_port;
ALTER SYSTEM DECOMMISSION BACKEND zuomm01:9050; 1、该命令用于安全删除 BE 节点。命令下发后Doris 会尝试将该 BE 上的数据向其 他 BE 节点迁移当所有数据都迁移完成后Doris 会自动删除该节点。
2、该命令是一个异步操作。执行后可以通过 SHOW PROC ‘/backends’; 看到该 BE节点的 isDecommission 状态为 true。表示该节点正在进行下线。
3、该命令不一定执行成功。比如剩余 BE 存储空间不足以容纳下线 BE 上的数据或者剩余机器数量不满足最小副本数时该命令都无法完成并且 BE 会一直处于isDecommission 为 true 的状态。
4、DECOMMISSION 的进度可以通过 SHOW PROC ‘/backends’; 中的 TabletNum 查看如果正在进行TabletNum 将不断减少。
5、该操作可以通过如下命令取消CANCEL DECOMMISSION BACKEND “be_host:be_heartbeat_service_port”; 取消0后该 BE 上的数据将维持当前剩余的数据量。后续 Doris 重新进行负载均衡。
3、Broker 扩容和缩容
Broker 实例的数量没有硬性要求。通常每台物理机部署一个即可。Broker 的添加和删除可以通过以下命令完成
ALTER SYSTEM ADD BROKER broker_name broker_host:broker_ipc_port;
ALTER SYSTEM DROP BROKER broker_name broker_host:broker_ipc_port;
ALTER SYSTEM DROP ALL BROKER broker_name; Broker 是无状态的进程可以随意启停。当然停止后正在其上运行的作业会失败重试即可。
3、数据表设计
3.1、字段类型 3.2、表的基本概念
1、Row Column
一张表包括行Row和列Column
Row 即用户的一行数据。Column 用于描述一行数据中不同的字段。
doris中的列分为两类key列和value列key列在doris中有两种作用聚合表模型中key是聚合和排序的依据其他表模型中key是排序依据2、分区与分桶
partition分区是在逻辑上将一张表按行(横向)划分
tablet又叫bucket分桶在物理上对一个分区再按行(横向)划分 3、Partition
1、Partition 列可以指定一列或多列在聚合模型中分区列必须为 KEY 列。
2、不论分区列是什么类型在写分区值时都需要加双引号。
3、分区数量理论上没有上限。
4、当不使用 Partition 建表时系统会自动生成一个和表名同名的全值范围的 Partition。该 Partition 对用户不可见并且不可删改。
5、创建分区时不可添加范围重叠的分区。
range分区
range分区创建语法
-- Range Partition
drop table if exists test.expamle_range_tbl;
CREATE TABLE IF NOT EXISTS test.expamle_range_tbl
(user_id LARGEINT NOT NULL COMMENT 用户id,date DATE NOT NULL COMMENT 数据灌入日期时间,timestamp DATETIME NOT NULL COMMENT 数据灌入的时间戳,city VARCHAR(20) COMMENT 用户所在城市,age SMALLINT COMMENT 用户年龄,sex TINYINT COMMENT 用户性别
)
ENGINEOLAP
DUPLICATE KEY(user_id, date) -- 表模型
-- 分区的语法
PARTITION BY RANGE(date) -- 指定分区类型和分区列
(-- 指定分区名称分区的上界 前闭后开PARTITION p201701 VALUES LESS THAN (2017-02-01), PARTITION p201702 VALUES LESS THAN (2017-03-01),PARTITION p201703 VALUES LESS THAN (2017-04-01)
)
DISTRIBUTED BY HASH(user_id) BUCKETS 1;1、分区列通常为时间列以方便的管理新旧数据。
2、Partition 支持通过 VALUES LESS THAN (…) 仅指定上界系统会将前一个分区的上界作为该分区的下界生成一个左闭右开的区间。同时也支持通过 VALUES […) 指定上下界生成一个左闭右开的区间。
3、通过 VALUES […) 同时指定上下界比较容易理解。这里举例说明当使用 VALUES LESS THAN (…) 语句进行分区的增删操作时分区范围的变化情况
如上 expamle_range_tbl 得建表语句中可以看到当建表完成后会自动生成如下3个分区
-- 查看表中分区得情况
SHOW PARTITIONS FROM test.expamle_range_tbl \G;mysql SHOW PARTITIONS FROM test.expamle_range_tbl \G;
*************************** 1. row ***************************PartitionId: 12020PartitionName: p201701VisibleVersion: 1VisibleVersionTime: 2022-08-30 21:57:36State: NORMALPartitionKey: dateRange: [types: [DATE]; keys: [0000-01-01]; ..types: [DATE]; keys: [2017-02-01]; )DistributionKey: user_idBuckets: 1ReplicationNum: 3StorageMedium: HDDCooldownTime: 9999-12-31 23:59:59
LastConsistencyCheckTime: NULLDataSize: 0.000 IsInMemory: falseReplicaAllocation: tag.location.default: 3
*************************** 2. row ***************************PartitionId: 12021PartitionName: p201702VisibleVersion: 1VisibleVersionTime: 2022-08-30 21:57:36State: NORMALPartitionKey: dateRange: [types: [DATE]; keys: [2017-02-01]; ..types: [DATE]; keys: [2017-03-01]; )DistributionKey: user_idBuckets: 1ReplicationNum: 3StorageMedium: HDDCooldownTime: 9999-12-31 23:59:59
LastConsistencyCheckTime: NULLDataSize: 0.000 IsInMemory: falseReplicaAllocation: tag.location.default: 3
*************************** 3. row ***************************PartitionId: 12022PartitionName: p201703VisibleVersion: 1VisibleVersionTime: 2022-08-30 21:57:35State: NORMALPartitionKey: dateRange: [types: [DATE]; keys: [2017-03-01]; ..types: [DATE]; keys: [2017-04-01]; )DistributionKey: user_idBuckets: 1ReplicationNum: 3StorageMedium: HDDCooldownTime: 9999-12-31 23:59:59
LastConsistencyCheckTime: NULLDataSize: 0.000 IsInMemory: falseReplicaAllocation: tag.location.default: 3
3 rows in set (0.00 sec)生成三个分区
p201701: [MIN_VALUE, 2017-02-01)
p201702: [2017-02-01, 2017-03-01)
p201703: [2017-03-01, 2017-04-01)当我们增加一个分区 p201705 VALUES LESS THAN (“2017-06-01”)分区结果如下
ALTER TABLE test.expamle_range_tbl ADD PARTITION p201705 VALUES LESS THAN (2017-06-01);
p201701: [MIN_VALUE, 2017-02-01)
p201702: [2017-02-01, 2017-03-01)
p201703: [2017-03-01, 2017-04-01)
p201705: [2017-04-01, 2017-06-01)此时我们删除分区 p201703则分区结果如下
ALTER TABLE test.expamle_range_tbl DROP PARTITION p201703;
p201701: [MIN_VALUE, 2017-02-01)
p201702: [2017-02-01, 2017-03-01)
p201705: [2017-04-01, 2017-06-01)注意到 p201702 和 p201705 的分区范围并没有发生变化而这两个分区之间出现了一个空洞[2017-03-01, 2017-04-01)。即如果导入的数据范围在这个空洞范围内是无法导入的。
分区的删除不会改变已存在分区的范围。删除分区可能出现空洞。通过 VALUES LESS THAN 语句增加分区时分区的下界紧接上一个分区的上界。
Range分区除了上述我们看到的单列分区也支持多列分区示例如下
PARTITION BY RANGE(date, id) 前闭后开
(PARTITION p201701_1000 VALUES LESS THAN (2017-02-01, 1000),PARTITION p201702_2000 VALUES LESS THAN (2017-03-01, 2000),PARTITION p201703_all VALUES LESS THAN (2017-04-01)-- 默认采用id类型的最小值
)在以上示例中我们指定 date(DATE 类型) 和 id(INT 类型) 作为分区列。以上示例最终得到的分区如下
* p201701_1000: [(MIN_VALUE, MIN_VALUE), (2017-02-01, 1000) )
* p201702_2000: [(2017-02-01, 1000), (2017-03-01, 2000) )
* p201703_all: [(2017-03-01, 2000), (2017-04-01, MIN_VALUE)) 注意最后一个分区用户缺失只指定了 date 列的分区值所以 id 列的分区值会默认填充 MIN_VALUE。当用户插入数据时分区列值会按照顺序依次比较最终得到对应的分区。举例如下 List分区
1、分区列支持 BOOLEAN, TINYINT, SMALLINT, INT, BIGINT, LARGEINT, DATE, DATETIME, CHAR, VARCHAR 数据类型分区值为枚举值。只有当数据为目标分区枚举值其中之一时才可以命中分区。
2、Partition 支持通过 VALUES IN (…) 来指定每个分区包含的枚举值。
下面通过示例说明进行分区的增删操作时分区的变化。
List分区创建语法
-- List PartitionCREATE TABLE IF NOT EXISTS test.expamle_list_tbl
(user_id LARGEINT NOT NULL COMMENT 用户id,date DATE NOT NULL COMMENT 数据灌入日期时间,timestamp DATETIME NOT NULL COMMENT 数据灌入的时间戳,city VARCHAR(20) NOT NULL COMMENT 用户所在城市,age SMALLINT NOT NULL COMMENT 用户年龄,sex TINYINT NOT NULL COMMENT 用户性别,last_visit_date DATETIME REPLACE DEFAULT 1970-01-01 00:00:00 COMMENT 用户最后一次访问时间,cost BIGINT SUM DEFAULT 0 COMMENT 用户总消费,max_dwell_time INT MAX DEFAULT 0 COMMENT 用户最大停留时间,min_dwell_time INT MIN DEFAULT 99999 COMMENT 用户最小停留时间
)
ENGINEolap
AGGREGATE KEY(user_id, date, timestamp, city, age, sex)
PARTITION BY LIST(city)
(PARTITION p_cn VALUES IN (Beijing, Shanghai, Hong Kong),PARTITION p_usa VALUES IN (New York, San Francisco),PARTITION p_jp VALUES IN (Tokyo)
)
-- 指定分桶的语法
DISTRIBUTED BY HASH(user_id) BUCKETS 1
PROPERTIES
(replication_num 3
);如上 example_list_tbl 示例当建表完成后会自动生成如下3个分区
p_cn: (Beijing, Shanghai, Hong Kong)
p_usa: (New York, San Francisco)
p_jp: (Tokyo)List分区也支持多列分区示例如下
PARTITION BY LIST(id, city)
(PARTITION p1_city VALUES IN ((1, Beijing), (1, Shanghai)),PARTITION p2_city VALUES IN ((2, Beijing), (2, Shanghai)),PARTITION p3_city VALUES IN ((3, Beijing), (3, Shanghai))
)在以上示例中我们指定 id(INT 类型) 和 city(VARCHAR 类型) 作为分区列。以上示例最终得到的分区如下 * p1_city: [(1, Beijing), (1, Shanghai)]
* p2_city: [(2, Beijing), (2, Shanghai)]
* p3_city: [(3, Beijing), (3, Shanghai)]4、Bucket
1、如果使用了 Partition则 DISTRIBUTED … 语句描述的是数据在各个分区内的划分规则。如果不使用 Partition则描述的是对整个表的数据的划分规则。
2、分桶列可以是多列但必须为 Key 列。分桶列可以和 Partition 列相同或不同。
3、分桶列的选择是在 查询吞吐 和 查询并发 之间的一种权衡
如果选择多个分桶列则数据分布更均匀。如果一个查询条件不包含所有分桶列的等值条件那么该查询会触发所有分桶同时扫描这样查询的吞吐会增加单个查询的延迟随之降低。这个方式适合大吞吐低并发的查询场景。如果仅选择一个或少数分桶列则对应的点查询可以仅触发一个分桶扫描。此时当多个点查询并发时这些查询有较大的概率分别触发不同的分桶扫描各个查询之间的IO影响较小尤其当不同桶分布在不同磁盘上时所以这种方式适合高并发的点查询场景。
4、分桶的数量理论上没有上限
关于 Partition 和 Bucket的数量和数据量的建议。
1、一个表的 Tablet 总数量等于 (Partition num * Bucket num)。
2、一个表的 Tablet 数量在不考虑扩容的情况下推荐略多于整个集群的磁盘数量。
3、单个 Tablet 的数据量理论上没有上下界但建议在 1G - 10G 的范围内。如果单个 Tablet 数据量过小则数据的聚合效果不佳且元数据管理压力大。如果数据量过大则不利于副本的迁移、补齐且会增加 Schema Change 或者 Rollup 操作失败重试的代价这些操作失败重试的粒度是 Tablet。分桶应该控制桶内数据量 不易过大或者过小
4、当 Tablet 的数据量原则和数量原则冲突时建议优先考虑数据量原则。
5、在建表时每个分区的 Bucket 数量统一指定。但是在动态增加分区时ADD PARTITION可以单独指定新分区的 Bucket 数量。可以利用这个功能方便的应对数据缩小或膨胀。
6、一个 Partition 的 Bucket 数量一旦指定不可更改。所以在确定 Bucket 数量时需要预先考虑集群扩容的情况。比如当前只有 3 台 host每台 host 有 1 块盘。如果 Bucket 的数量只设置为 3 或更小那么后期即使再增加机器也不能提高并发度。
举例
假设在有10台BE每台BE一块磁盘的情况下。如果一个表总大小为 500MB则可以考虑4-8个分片。5GB8-16个分片。50GB32个分片。500GB建议分区每个分区大小在 50GB 左右每个分区16-32个分片。5TB建议分区每个分区大小在 500GB 左右每个分区16-32个分片。注表的数据量可以通过 SHOW DATA命令查看结果除以副本数即表的数据量。5、复合分区与单分区的选择
复合分区 第一级称为 Partition即分区。用户可以指定某一维度列作为分区列当前只支持整型和时间类型的列并指定每个分区的取值范围。 第二级称为 Distribution即分桶。用户可以指定一个或多个维度列以及桶数对数据进行 HASH 分布。
以下场景推荐使用复合分区 有时间维度或类似带有有序值的维度可以以这类维度列作为分区列。分区粒度可以根据导入频次、分区数据量等进行评估。地域、时间 历史数据删除需求如有删除历史数据的需求比如仅保留最近N 天的数据。使用复合分区可以通过删除历史分区来达到目的。也可以通过在指定分区内发送 DELETE 语句进行数据删除。 解决数据倾斜问题每个分区可以单独指定分桶数量。如按天分区当每天的数据量差异很大时可以通过指定分区的分桶数合理划分不同分区的数据,分桶列建议选择区分度大的列。
用户也可以不使用复合分区即使用单分区。则数据只做 HASH 分布。
练习
6、建表指定分区和分桶数
需求现在有如下数据需要插入到表中请创建一个表要求按照月份分区2个分桶 》
用哪一个几个列作为分桶字段
-- 数据
uid name age gender province birthday1 zss 18 male jiangsu 2022-11-012 lss 18 male zhejiang 2022-11-103 ww 18 male jiangsu 2022-12-014 zll 18 female zhejiang 2022-09-115 tqq 18 female jiangsu 2022-09-026 aa 18 female jiangsu 2022-10-117 bb 18 male zhejiang 2022-11-08CREATE TABLE IF NOT EXISTS test.user_info
(
uid int,
name varchar(50),
age int,
gender varchar(20),
province varchar(100),
birthday date
)
ENGINEolap
duplicate KEY(uid,name)
PARTITION BY range(birthday)
(
partition p202209 values less than (2022-10-01),
partition p202210 values less than (2022-11-01),
partition p202211 values less than (2022-12-01),
partition p202212 values less than (2023-01-01)
)
-- 指定分桶的语法
DISTRIBUTED BY HASH(uid) BUCKETS 1
PROPERTIES(replication_num 2,
);7、PROPERTIES
在建表语句的最后可以用 PROPERTIES 关键字来设置一些表的属性参数参数有很多
分片副本数
replication_num
每个 Tablet 的副本数量。默认为 3建议保持默认即可。在建表语句中所有 Partition中的 Tablet 副本数量统一指定。而在增加新分区时可以单独指定新分区中 Tablet 的副本数量。
副本数量可以在运行时修改。强烈建议保持奇数。
最大副本数量取决于集群中独立 IP 的数量注意不是 BE 数量。Doris 中副本分布的原则是不允许同一个 Tablet 的副本分布在同一台物理机上而识别物理机即通过 IP。所以即使在同一台物理机上部署了 3 个或更多 BE 实例如果这些 BE 的 IP 相同则依然只能设置副本数为 1。对于一些小并且更新不频繁的维度表可以考虑设置更多的副本数。这样在 Join 查询时可以有更大的概率进行本地数据 Join。
存储介质 和 热数据冷却时间
storage_medium
storage_cooldown_time datetime
建表时可以统一指定所有 Partition 初始存储的介质及热数据的冷却时间如
storage_medium SSD
storage_cooldown_time 2022-11-30 00:00:00默认初始存储介质可通过 fe 的配置文件 fe.conf 中指定 default_storage_mediumxxx,如果没有指定则默认为 HDD。如果指定为 SSD则数据初始存放在 SSD 上。没设storage_cooldown_time则默认 30 天后数据会从 SSD 自动迁移到 HDD上。如果指定了 storage_cooldown_time则在到达 storage_cooldown_time 时间后数据才会迁移。
注意当指定 storage_medium 时如果 FE 参数 enable_strict_storage_medium_check 为False 该参数只是一个“尽力而为”的设置。即使集群内没有设置 SSD 存储介质也不会报错而是自动存储在可用的数据目录中。同样如果 SSD 介质不可访问、空间不足都可能导致数据初始直接存储在其他可用介质上。而数据到期迁移到 HDD 时如果 HDD 介质不 可 访 问 、 空 间 不 足 也 可 能 迁 移 失 败 但 是 会 不 断 尝 试 。如 果 FE 参 数enable_strict_storage_medium_check 为 True 则当集群内没有设置 SSD 存储介质时会报错Failed to find enough host in all backends with storage medium is SSD。
3.3、数据表模型
Doris 的数据模型主要分为3类:
Aggregate 聚合模型Unique 唯一模型Duplicate 明细模型
1、Aggregate 模型
是相同key的数据进行自动聚合的表模型。表中的列按照是否设置了 AggregationType分为 Key维度列和 Value指标列没有设置 AggregationType 的称为 Key设置了 AggregationType 的称为 Value。当我们导入数据时对于 Key 列相同的行会聚合成一行而 Value 列会按照设置的AggregationType 进行聚合。AggregationType 目前有以下四种聚合方式
SUM求和多行的 Value 进行累加。REPLACE替代下一批数据中的 Value 会替换之前导入过的行中的 Value。REPLACE_IF_NOT_NULL 当遇到 null 值则不更新。MAX保留最大值。MIN保留最小值。
有如下场景需要创建一个表来记录公司每个用户的每一次消费行为信息有如下字段 而且公司对这份数据特别关心一个报表
每次要看这个报表都需要在“明细表”上运行一个统计sql
Selectuser_id,data,city,age,gender,max(visit_data) as last_visit_data,sum(cost) as cost,max(dwell_time) as max_dwell_time,min(dwell_time) as min_dwell_time
From t
Group by user_id,data,city,age,gender -- 对应的是聚合模型型key聚合模型 sql示例
-- 这是一个用户消费和行为记录的数据表
CREATE TABLE IF NOT EXISTS test.ex_user
(user_id LARGEINT NOT NULL COMMENT 用户 id,date DATE NOT NULL COMMENT 数据灌入日期时间,city VARCHAR(20) COMMENT 用户所在城市,age SMALLINT COMMENT 用户年龄,sex TINYINT COMMENT 用户性别,last_visit_date DATETIME REPLACE DEFAULT 1970-01-01 00:00:00 COMMENT 用户最后一次访问时间,cost BIGINT SUM DEFAULT 0 COMMENT 用户总消费,max_dwell_time INT MAX DEFAULT 0 COMMENT 用户最大停留时间,min_dwell_time INT MIN DEFAULT 99999 COMMENT 用户最小停留时间 )
ENGINEolap
AGGREGATE KEY(user_id, date, city, age, sex)
-- 分区
-- 分桶
DISTRIBUTED BY HASH(user_id) BUCKETS 1;插入数据
insert into test.ex_user values\
(10000,2017-10-01,北京,20,0,2017-10-01 06:00:00,20,10,10),\
(10000,2017-10-01,北京,20,0,2017-10-01 07:00:00,15,2,2),\
(10001,2017-10-01,北京,30,1,2017-10-01 17:05:45,2,22,22),\
(10002,2017-10-02,上海,20,1,2017-10-02 12:59:12,200,5,5),\
(10003,2017-10-02,广州,32,0,2017-10-02 11:20:00,30,11,11),\
(10004,2017-10-01,深圳,35,0,2017-10-01 10:00:15,100,3,3),\
(10004,2017-10-03,深圳,35,0,2017-10-03 10:20:22,11,6,6);查看数据的时候发现数据只剩下6条了就是因为再key相同的时候将后面的结果聚合了 数据聚合在 Doris 中有如下三个阶段发生
1、每一批次数据导入的 ETL 阶段。该阶段会在每一批次导入的数据内部进行聚合。2、底层 BE 进行数据 Compaction 的阶段。BE 会对已导入的不同批次的数据进行进一步的聚合。3、数据查询阶段。在数据查询时对于查询涉及到的数据会进行对应的聚合。 2、UNIQUE 模型
是相同key的数据进行自动去重的表模型。在某些多维分析场景下用户更关注的是如何保证 Key 的唯一性即如何获得 Primary Key 唯一性约束。因此引入了 Uniq 的数据模型。该模型本质上是聚合模型的一个特例也是一种简化的表结构表示方式。
drop table if exists test.user;
CREATE TABLE IF NOT EXISTS test.user
(user_id LARGEINT NOT NULL COMMENT 用户 id,username VARCHAR(50) NOT NULL COMMENT 用户昵称,city VARCHAR(20) COMMENT 用户所在城市,age SMALLINT COMMENT 用户年龄,sex TINYINT COMMENT 用户性别,phone LARGEINT COMMENT 用户电话,address VARCHAR(500) COMMENT 用户地址,register_time DATETIME COMMENT 用户注册时间 )
UNIQUE KEY(user_id, username)
DISTRIBUTED BY HASH(user_id) BUCKETS 1;插入数据
insert into test.user values\
(10000,zss,北京,18,0,12345678910,北京朝阳区 ,2017-10-01 07:00:00),\
(10000,zss,北京,19,0,12345678910,北京顺义区 ,2018-10-01 07:00:00),\
(10000,lss,北京,20,0,12345678910,北京海淀区,2017-11-15 06:10:20);insert into test.user1 values\
(10000,zss,北京,18,0,12345678910,北京朝阳区 ,2017-10-01 07:00:00),\
(10000,zss,北京,19,0,12345678910,北京顺义区 ,2018-10-01 07:00:00),\
(10000,lss,北京,20,0,12345678910,北京海淀区,2017-11-15 06:10:20);查询结果后发现相同的数据就会被替换掉 Uniq 模型完全可以用聚合模型中的 REPLACE 方式替代。其内部的实现方式和数据存储方式也完全一样。
3、Duplicate 模型
就是存明细数据的表模型既不做聚合也不做去重。在某些多维分析场景下数据既没有主键也没有聚合需求。Duplicate 数据模型可以满足这类需求。数据完全按照导入文件中的数据进行存储不会有任何聚合。即使两行数据完全相同也都会保留。而在建表语句中指定的 DUPLICATE KEY只是用来指明底层数据按照那些列进行排序。
建表语句
CREATE TABLE IF NOT EXISTS test.log_detail
(timestamp DATETIME NOT NULL COMMENT 日志时间,type INT NOT NULL COMMENT 日志类型,error_code INT COMMENT 错误码,error_msg VARCHAR(1024) COMMENT 错误详细信息,op_id BIGINT COMMENT 负责人 id,op_time DATETIME COMMENT 处理时间 )
DUPLICATE KEY(timestamp, type)
DISTRIBUTED BY HASH(timestamp) BUCKETS 1;插入部分数据
insert into test.log_detail values\
(2017-10-01 08:00:05,1,404,not found page, 101, 201e7-10-01 08:00:05),\
(2017-10-01 08:00:05,1,404,not found page, 101, 2017-10-01 08:00:05),\
(2017-10-01 08:00:05,2,404,not found page, 101, 2017-10-01 08:00:06),\
(2017-10-01 08:00:06,2,404,not found page, 101, 2017-10-01 08:00:07);查询结果后发现插入的数据全部会被保留即使两条数据一模一样也会保留正常可以操作用户行为日志数据这种 4、数据模型的选择
数据模型在建表时就已经确定且无法修改所以选择一个合适的数据模型非常重要。 Aggregate 模型可以通过预聚合极大地降低聚合查询时所需扫描的数据量和查询的计算量非常适合有固定模式的报表类查询场景。 Uniq 模型针对需要唯一主键约束的场景可以保证主键唯一性约束。但是无法利用 ROLLUP 等预聚合带来的查询优势因为本质是 REPLACE没有 SUM 这种聚合方式。 Duplicate 适合任意维度的查询。虽然同样无法利用预聚合的特性但是不受聚合模型的约束可以发挥列存模型的优势只读取相关列而不需要读取所有 Key 列
3.4、索引
索引用于帮助快速过滤或查找数据。
目前 Doris 主要支持两类索引 内建的智能索引包括前缀索引和 ZoneMap 索引。 用户创建的二级索引包括 Bloom Filter 索引 和 Bitmap倒排索引。
其中 ZoneMap 索引是在列存格式上对每一列自动维护的索引信息包括 Min/MaxNull 值个数等等。这种索引对用户透明。
1、前缀索引
doris中对于前缀索引有如下约束
1、他的索引键最大长度是36个字节
2、当他遇到了varchar数据类型的时候即使没有超过36个字节也会自动截断
示例1:以下表中我们定义了: user_id,age,message作为表的key
那么doris为这个表创建前缀索引时它生成的索引键如下
user_id(8 Bytes) age(4 Bytes) message(prefix 24 Bytes示例2:以下表中我们定义了ageuser_namemessage作为表的key
age(4 Bytes) user_name(20 Bytes) 指定key的时候为什么是这个结果呢
虽然还没有超过36个字节但是已经遇到了一个varchar字段它自动截断不会再往后面取了
当我们的查询条件是前缀索引的前缀时可以极大的加快查询速度。比如在第一个例子中我们执行如下查询
SELECT * FROM table WHERE user_id1829239 and age20该查询的效率会远高于以下查询
SELECT * FROM table WHERE age20所以在建表时正确的选择列顺序能够极大地提高查询效率。
2、Bloom Filter 索引
总结
1、Bloom Filter 本质上是一种位图结构用于判断一个值是否存在
2、会产生小概率的误判因为hash算法天生的碰撞
3、在doris中是以tablet为粒度创建的给每一个tablet创建一个布隆过滤器索引
3、如何创建BloomFilter索引?
1、建表的时候指定
PROPERTIES (
bloom_filter_columnsname,age
)CREATE TABLE IF NOT EXISTS sale_detail_bloom (sale_date date NOT NULL COMMENT 销售时间,customer_id int NOT NULL COMMENT 客户编号,saler_id int NOT NULL COMMENT 销售员,sku_id int NOT NULL COMMENT 商品编号,category_id int NOT NULL COMMENT 商品分类,sale_count int NOT NULL COMMENT 销售数量,sale_price DECIMAL(12,2) NOT NULL COMMENT 单价,sale_amt DECIMAL(20,2) COMMENT 销售总金额
)
Duplicate KEY(sale_date, customer_id,saler_id,sku_id,category_id)
PARTITION BY RANGE(sale_date)
(
PARTITION P_202111 VALUES [(2021-11-01), (2021-12-01))
)
DISTRIBUTED BY HASH(saler_id) BUCKETS 1
PROPERTIES (
replication_num 1,
bloom_filter_columnssaler_id,category_id,
storage_medium SSD
);2、alter修改表的时候指定
ALTER TABLE sale_detail_bloom SET (bloom_filter_columns k1,k3);ALTER TABLE sale_detail_bloom SET (bloom_filter_columns k1,k4);3、查看BloomFilter索引
mysql SHOW CREATE TABLE sale_detail_bloom \G;
*************************** 1. row ***************************Table: sale_detail_bloom
Create Table: CREATE TABLE sale_detail_bloom (sale_date date NOT NULL COMMENT 销售时间,customer_id int(11) NOT NULL COMMENT 客户编号,saler_id int(11) NOT NULL COMMENT 销售员,sku_id int(11) NOT NULL COMMENT 商品编号,category_id int(11) NOT NULL COMMENT 商品分类,sale_count int(11) NOT NULL COMMENT 销售数量,sale_price decimal(12, 2) NOT NULL COMMENT 单价,sale_amt decimal(20, 2) NULL COMMENT 销售总金额
) ENGINEOLAP
DUPLICATE KEY(sale_date, customer_id, saler_id, sku_id, category_id)
COMMENT OLAP
PARTITION BY RANGE(sale_date)
(PARTITION P_202111 VALUES [(2021-11-01), (2021-12-01)),
PARTITION P_202208 VALUES [(2022-08-01), (2022-09-01)),
PARTITION P_202209 VALUES [(2022-09-01), (2022-10-01)),
PARTITION P_202210 VALUES [(2022-10-01), (2022-11-01)))
DISTRIBUTED BY HASH(saler_id) BUCKETS 1
PROPERTIES (
replication_allocation tag.location.default: 3,
bloom_filter_columns category_id, saler_id
)
1 row in set (0.00 sec)4、修改/删除BloomFilter索引
ALTER TABLE sale_detail_bloom SET (bloom_filter_columns );5、Doris BloomFilter适用场景
满足以下几个条件时可以考虑对某列建立Bloom Filter 索引
1、BloomFilter是在无法利用前缀索引的查询场景中来加快查询速度的。
2、查询会根据该列高频过滤而且查询条件大多是 in 和 过滤。 不同于Bitmap, BloomFilter适用于高基数列。比如UserID。因为如果创建
3、在低基数的列上比如 “性别” 列则每个Block几乎都会包含所有取值导致BloomFilter索引失去意义。字段随机
Doris BloomFilter使用注意事项
1、不支持对Tinyint、Float、Double 类型的列建Bloom Filter索引。
2、Bloom Filter索引只对 in 和 过滤查询有加速效果。
3、可以通过explain来查看命中了哪种索引 --没办法查看
6、Bitmap 索引
1、创建索引
CREATE INDEX [IF NOT EXISTS] index_name ON table1 (siteid) USING BITMAP COMMENT balabala;
create index user_id_bitmap on sale_detail_bloom(sku_id) USING BITMAP COMMENT 使用user_id创建的bitmap索引;2、查看索引
SHOW INDEX FROM example_db.table_name;3、删除索引
DROP INDEX [IF EXISTS] index_name ON [db_name.]table_name;注意事项
1、bitmap 索引仅在单列上创建。
2、bitmap 索引能够应用在 Duplicate、Uniq 数据模型的所有列和 Aggregate模型的key列上。
3、bitmap 索引支持的数据类型如下:(老版本只支持bitmap类型) TINYINTSMALLINTINTBIGINTCHARVARCHARDATEDATETIMELARGEINTDECIMALBOOL
4、bitmap索引仅在 Segment V2 下生效(Segment V2是升级版本的文件格式)。当创建 index 时表的存储格式将默认转换为 V2 格式
练习
-- 数据
uid name age gender province term1 zss 18 male jiangsu 12 lss 16 male zhejiang 23 ww 19 male jiangsu 14 zll 18 female zhejiang 35 tqq 17 female jiangsu 26 aa 18 female jiangsu 27 bb 17 male zhejiang 3提要求
这张表以后需要经常按照如下条件查询
-- 前缀索引 key 》 term province
where term ??
where term ?? and province ??-- 布隆过滤器索引
where name??
-- bitmap索引
where uid?? SET GLOBAL enable_profiletrue;--主要是索引怎么建
create table stu(
term int,
province varchar(100),
uid int,
name varchar(30),
age int,
gender varchar(30)
)
engine olap
duplicate key(term ,province)
distributed by hash(uid) buckets 2
properties(
bloom_filter_columnsname
);3.5、Rollup
ROLLUP 在多维分析中是“上卷”的意思即将数据按某种指定的粒度进行进一步聚合。
之前的聚合模型
1、求每个城市的每个用户的每天的总销售额
select
user_id,city,date
sum(sum_cost) as sum_cost
from t
group by user_id,city,date
-- user_id date city sum_cost10000 2017/10/2 北京 19510000 2017/10/1 上海 10010000 2017/10/2 上海 30 10000 2017/10/3 上海 55 10000 2017/10/4 上海 65 10001 2017/10/1 上海 3010001 2017/10/2 上海 10 10001 2017/10/2 天津 18 10001 2017/10/1 天津 4610002 2017/10/1 天津 5510002 2017/10/3 北京 55 10002 2017/10/2 天津 20 10002 2017/10/2 北京 35 2、求每个用户、每个城市的总消费额
select
user_id,city
sum(sum_cost) as sum_cost
from t
group by user_id,city
user_id city sum_cost
10000 北京 195
10000 上海 100
10001 上海 40
10001 天津 64
10002 天津 75
10002 北京 90
3.求每个用户的总消费额
3、求每个用户的总消费额
select
user_id
sum(sum_cost) as sum_cost
from t
group by user_id
user_id sum_cost
10000 295
10001 104
10002 1654、基本概念
通过建表语句创建出来的表称为 Base 表Base Table,基表
在 Base 表之上我们可以创建任意多个 ROLLUP 表。这些 ROLLUP 的数据是基于 Base 表产生的并且在物理上是独立存储的。
Rollup表的好处
1、和基表共用一个表名doris会根据具体的查询逻辑选择合适的数据源(合适的表)来计算结果
2、对于基表中数据的增删改rollup表会自动更新同步
3.6、物化视图
就是查询结果预先存储起来的特殊的表。物化视图的出现主要是为了满足用户既能对原始明细数据的任意维度分析也能快速的对固定维度进行分析查询
1、物化视图好处
1、可以复用预计算的结果来提高查询效率
2、自动实时的维护物化视图表中的结果数据无需额外人工成本(自动维护会有计算资源的开销)
3、查询时会自动选择最优物化视图
2、物化视图 VS Rollup
1、明细模型表下rollup和物化视图的差别
物化视图都可以实现预聚合新增一套前缀索引rollup对于明细模型新增一套前缀索引
2、聚合模型下功能一致
3、创建物化视图
CREATE MATERIALIZED VIEW [MV name] as
[query] -- sql逻辑--[MV name]雾化视图的名称
--[query]查询条件基于base表创建雾化视图的逻辑物化视图创建成功后用户的查询不需要发生任何改变也就是还是查询的 base 表。Doris 会根据当前查询的语句去自动选择一个最优的物化视图从物化视图中读取数据并计算。
用户可以通过 EXPLAIN 命令来检查当前查询是否使用了物化视图。
4、示例
创建一个 Base 表
用户有一张销售记录明细表存储了每个交易的交易id销售员售卖门店销售时间以及金额
create table sales_records(
record_id int,
seller_id int,
store_id int,
sale_date date,
sale_amt bigint)
duplicate key (record_id,seller_id,store_id,sale_date)
distributed by hash(record_id) buckets 2
properties(replication_num 1);-- 插入数据
insert into sales_records values \
(1,1,1,2022-02-02,100),\
(2,2,1,2022-02-02,200),\
(3,3,2,2022-02-02,300),\
(4,3,2,2022-02-02,200),\
(5,2,1,2022-02-02,100),\
(6,4,2,2022-02-02,200),\
(7,7,3,2022-02-02,300),\
(8,2,1,2022-02-02,400),\
(9,9,4,2022-02-02,100);如果用户需要经常对不同门店的销售量进行统计
1、创建一个物化视图
-- 不同门店看总销售额的一个场景
select store_id, sum(sale_amt)
from sales_records
group by store_id; CREATE MATERIALIZED VIEW store_id_sale_amonut as
select store_id, sum(sale_amt)
from sales_records
group by store_id;CREATE MATERIALIZED VIEW store_amt as
select store_id, sum(sale_amt) as sum_amount
from sales_records
group by store_id; --针对上述场景做一个物化视图
create materialized view store_amt as
select store_id, sum(sale_amt) as sum_amount
from sales_records
group by store_id; 2、检查物化视图是否构建完成(物化视图的创建是个异步的过程)
show alter table materialized view from 库名 order by CreateTime desc limit 1;show alter table materialized view from test order by CreateTime desc limit 1;------------------------------------------------------------------------------------------------------------------------------------------------------------
| JobId | TableName | CreateTime | FinishTime | BaseIndexName | RollupIndexName | RollupId | TransactionId | State | Msg | Progress | Timeout |
------------------------------------------------------------------------------------------------------------------------------------------------------------
| 15093 | sales_records | 2022-11-25 10:32:33 | 2022-11-25 10:32:59 | sales_records | store_amt | 15094 | 3008 | FINISHED | | NULL | 86400 |
------------------------------------------------------------------------------------------------------------------------------------------------------------查看 Base 表的所有物化视图
desc sales_records all;
---------------------------------------------------------------------------------------
| IndexName | IndexKeysType | Field | Type | Null | Key | Default | Extra | Visible |
---------------------------------------------------------------------------------------
| sales_records | DUP_KEYS | record_id | INT | Yes | true | NULL | | true |
| | | seller_id | INT | Yes | true | NULL | | true |
| | | store_id | INT | Yes | true | NULL | | true |
| | | sale_date | DATE | Yes | true | NULL | | true |
| | | sale_amt | BIGINT | Yes | false | NULL | NONE | true |
| | | | | | | | | |
| store_amt | AGG_KEYS | store_id | INT | Yes | true | NULL | | true |
| | | sale_amt | BIGINT | Yes | false | NULL | SUM | true |
---------------------------------------------------------------------------------------3、查询
看是否命中刚才我们建的物化视图
EXPLAIN SELECT store_id, sum(sale_amt) FROM sales_records GROUP BY store_id;------------------------------------------------------------------------------------
| Explain String |
------------------------------------------------------------------------------------
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:slot 2 store_id | slot 3 sum(sale_amt) |
| PARTITION: UNPARTITIONED |
| |
| VRESULT SINK |
| |
| 4:VEXCHANGE |
| |
| PLAN FRAGMENT 1 |
| |
| PARTITION: HASH_PARTITIONED: slot 2 store_id |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 04 |
| UNPARTITIONED |
| |
| 3:VAGGREGATE (merge finalize) |
| | output: sum(slot 3 sum(sale_amt)) |
| | group by: slot 2 store_id |
| | cardinality-1 |
| | |
| 2:VEXCHANGE |
| |
| PLAN FRAGMENT 2 |
| |
| PARTITION: HASH_PARTITIONED: default_cluster:study.sales_records.record_id |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 02 |
| HASH_PARTITIONED: slot 2 store_id |
| |
| 1:VAGGREGATE (update serialize) |
| | STREAMING |
| | output: sum(sale_amt) |
| | group by: store_id |
| | cardinality-1 |
| | |
| 0:VOlapScanNode |
| TABLE: sales_records(store_amt), PREAGGREGATION: ON |
| partitions1/1, tablets10/10, tabletList15095,15097,15099 ... |
| cardinality7, avgRowSize1560.0, numNodes3 |
------------------------------------------------------------------------------------5、删除物化视图语法
-- 语法
DROP MATERIALIZED VIEW 物化视图名 on base_table_name; --示例
drop materialized view store_amt on sales_records;6、计算广告的 pv、uv
用户有一张点击广告的明细数据表
需求针对用户点击计广告明细数据的表算每天每个页面每个渠道的 pv,uv
pv:page view页面浏览量或点击量
uv:unique view通过互联网访问、浏览这个网页的自然人
drop table if exists ad_view_record;
create table ad_view_record(
dt date,
ad_page varchar(10),
channel varchar(10),
refer_page varchar(10),
user_id int
)
distributed by hash(dt)
properties(replication_num 1);select
dt,ad_page,channel,
count(refer_page) as pv,
count(distinct user_id ) as uv
from ad_view_record
group by dt,ad_page,channel插入数据
insert into ad_view_record values \
(2020-02-02,a,app,/home,1),\
(2020-02-02,a,web,/home,1),\
(2020-02-02,a,app,/addbag,2),\
(2020-02-02,b,app,/home,1),\
(2020-02-02,b,web,/home,1),\
(2020-02-02,b,app,/addbag,2),\
(2020-02-02,b,app,/home,3),\
(2020-02-02,b,web,/home,3),\
(2020-02-02,c,app,/order,1),\
(2020-02-02,c,app,/home,1),\
(2020-02-03,c,web,/home,1),\
(2020-02-03,c,app,/order,4),\
(2020-02-03,c,app,/home,5),\
(2020-02-03,c,web,/home,6),\
(2020-02-03,d,app,/addbag,2),\
(2020-02-03,d,app,/home,2),\
(2020-02-03,d,web,/home,3),\
(2020-02-03,d,app,/addbag,4),\
(2020-02-03,d,app,/home,5),\
(2020-02-03,d,web,/addbag,6),\
(2020-02-03,d,app,/home,5),\
(2020-02-03,d,web,/home,4);创建物化视图
-- 怎么去计算pvuv
select
dt,ad_page,channel,
count(ad_page) as pv,
count(distinct user_id) as uv
from ad_view_record
group by dt,ad_page,channel;-- 1.物化视图中不能够使用两个相同的字段
-- 2.在增量聚合里面不能够使用count(distinct) bitmap_union
-- 3.count(字段)create materialized view dpc_pv_uv as
select
dt,ad_page,channel,
-- refer_page 没有null的情况
count(refer_page) as pv,
-- doris的物化视图中不支持count(distint) bitmap_union
-- count(distinct user_id) as uv
bitmap_union(to_bitmap(user_id)) uv_bitmap
from ad_view_record
group by dt,ad_page,channel;create materialized view tpc_pv_uv as
select
dt,ad_page,channel,
count(refer_page) as pv,
-- refer_page 不能为null
-- count(user_id) as pv
-- count(1) as pv,
bitmap_union(to_bitmap(user_id)) as uv_bitmap
--count(distinct user_id) as uv
from ad_view_record
group by dt,ad_page,channel;
--结论在doris的物化视图中一个字段不能用两次,并且聚合函数后面必须跟字段名称在 Doris 中count(distinct) 聚合的结果和 bitmap_union_count 聚合的结果是完全一致的。而 bitmap_union_count 等于 bitmap_union 的结果求 count所以如果查询中涉及到count(distinct) 则通过创建带 bitmap_union 聚合的物化视图方可加快查询。因为本身 user_id 是一个 INT 类型所以在 Doris 中需要先将字段通过函数 to_bitmap 转换为 bitmap 类型然后才可以进行 bitmap_union 聚合。
查询自动匹配
explain
select
dt,ad_page,channel,
count(refer_page) as pv,
count(distinct user_id) as uv
from ad_view_record
group by dt,ad_page,channel;会自动转换成。
explain
select
dt,ad_page,channel,
count(1) as pv,
bitmap_union_count(to_bitmap(user_id)) as uv
from ad_view_record
group by dt,ad_page,channel;总结
1、在创建doris的物化视图中同一个字段不能被使用两次,并且聚合函数后面必须跟字段名称(不能使用count(1)这样的聚合逻辑)
2、doris在选择使用哪一个物化视图表的时候按照维度上卷的原则选距离查询维度最接近并且指标可以复用的物化视图
3、一张基表可以创建多个物化视图(计算资源占用比较多)
7、调整前缀索引
场景:用户的原始表有k1, k2, k3三列。其中 k1, k2 为前缀索引列。这时候如果用户查询条件中包含 where k11 and k22 就能通过索引加速查询。
但是有些情况下用户的过滤条件无法匹配到前缀索引比如 where k33。则无法通过索引提升查询速度。
解决方法
创建以 k3 作为第一列的物化视图就可以解决这个问题。
查询
desc sales_records all;
---------------------------------------------------------------------------------------
| IndexName | IndexKeysType | Field | Type | Null | Key | Default | Extra | Visible |
---------------------------------------------------------------------------------------
| sales_records | DUP_KEYS | record_id | INT | Yes | true | NULL | | true |
| | | seller_id | INT | Yes | true | NULL | | true |
| | | store_id | INT | Yes | true | NULL | | true |
| | | sale_date | DATE | Yes | true | NULL | | true |
| | | sale_amt | BIGINT | Yes | false | NULL | NONE | true |
---------------------------------------------------------------------------------------
5 rows in set (0.00 sec)--针对上面的前缀索引情况执行下面的sql是无法利用前缀索引的
explain
select record_id,seller_id,store_id from sales_records
where store_id3;创建物化视图
create materialized view sto_rec_sell as
select store_id, record_id, seller_id, sale_date, sale_amt
from sales_records;通过上面语法创建完成后物化视图中既保留了完整的明细数据且物化视图的前缀索
引为 store_id 列。
查看表结构
desc sales_records all;
---------------------------------------------------------------------------------------
| IndexName | IndexKeysType | Field | Type | Null | Key | Default | Extra | Visible |
---------------------------------------------------------------------------------------
| sales_records | DUP_KEYS | record_id | INT | Yes | true | NULL | | true |
| | | seller_id | INT | Yes | true | NULL | | true |
| | | store_id | INT | Yes | true | NULL | | true |
| | | sale_date | DATE | Yes | true | NULL | | true |
| | | sale_amt | BIGINT | Yes | false | NULL | NONE | true |
| | | | | | | | | |
| sto_rec_sell | DUP_KEYS | store_id | INT | Yes | true | NULL | | true |
| | | record_id | INT | Yes | true | NULL | | true |
| | | seller_id | INT | Yes | true | NULL | | true |
| | | sale_date | DATE | Yes | false | NULL | NONE | true |
| | | sale_amt | BIGINT | Yes | false | NULL | NONE | true |
---------------------------------------------------------------------------------------查询匹配
explain select record_id,seller_id,store_id from sales_records where store_id3;
------------------------------------------------------------------------------------
| Explain String |
------------------------------------------------------------------------------------
| PLAN FRAGMENT 0 |
| OUTPUT EXPRS:record_id | seller_id | store_id |
| PARTITION: UNPARTITIONED |
| |
| VRESULT SINK |
| |
| 1:VEXCHANGE |
| |
| PLAN FRAGMENT 1 |
| |
| PARTITION: HASH_PARTITIONED: default_cluster:study.sales_records.record_id |
| |
| STREAM DATA SINK |
| EXCHANGE ID: 01 |
| UNPARTITIONED |
| |
| 0:VOlapScanNode |
| TABLE: sales_records(sto_rec_sell), PREAGGREGATION: ON |
| PREDICATES: store_id 3 |
| partitions1/1, tablets10/10, tabletList15300,15302,15304 ... |
| cardinality0, avgRowSize12.0, numNodes1 |
------------------------------------------------------------------------------------这时候查询就会直接从刚才创建的sto_rec_sell物化视图中读取数据。物化视图对 store_id是存在前缀索引的查询效率也会提升。
4、数据的导入和导出
按照使用场景划分 4.1、使用 Insert 方式同步数据
用户可以通过 MySQL 协议使用 INSERT 语句进行数据导入。
INSERT 语句的使用方式和 MySQL 等数据库中 INSERT 语句的使用方式类似。INSERT 语句支持以下两种语法
* INSERT INTO table SELECT ...
* INSERT INTO table VALUES(...)对于 Doris 来说一个 INSERT 命令就是一个完整的导入事务。
因此不论是导入一条数据还是多条数据我们都不建议在生产环境使用这种方式进行数据导入。高频次的 INSERT 操作会导致在存储层产生大量的小文件会严重影响系统性能。
该方式仅用于线下简单测试或低频少量的操作。
或者可以使用以下方式进行批量的插入操作
INSERT INTO example_tbl VALUES
(1000, baidu1, 3.25)
(2000, baidu2, 4.25)
(3000, baidu3, 5.25);4.2、导入本地数据
Stream Load 用于将本地文件导入到doris中。Stream Load 是通过 HTTP 协议与 Doris 进行连接交互的。
该方式中涉及 HOST:PORT 都是对应的HTTP 协议端口。
BE 的 HTTP 协议端口默认为 8040。FE 的 HTTP 协议端口默认为 8030。
但须保证客户端所在机器网络能够联通FE, BE 所在机器。
基本原理 | || | 1. User submit load to FE 1.提交导入请求| || --v-----------| | FE | 生成导入计划
4. Return result to user | -------------| || | 2. Redirect to BE 下发给每一个BE节点| || --v--------------Coordinator BE| 1B. User submit load to BE-----------| | |----- | -----| | | 3. Distrbute data 分发数据并导入| | |-v- -v- -v-|BE | |BE | |BE |--- --- ---1、创建一张表
drop table if exists load_local_file_test;
CREATE TABLE IF NOT EXISTS load_local_file_test
(id INT,name VARCHAR(50),age TINYINT
)
unique key(id)
DISTRIBUTED BY HASH(id) BUCKETS 3;1,zss,28
2,lss,28
3,ww,882、导入数据 执行 curl 命令导入本地文件这个命令不是在mysql端执行的哦
# 语法示例curl \-u user:passwd \ # 账号密码-H label:load_local_file_test \ # 本次任务的唯一标识-T 文件地址 \http://主机名:端口号/api/库名/表名/_stream_loadcurl \-u root:123456 \-H label:load_local_file \-H column_separator:, \-T /root/data/loadfile.txt \http://zuomm01:8040/api/test/load_local_file_test/_stream_loaduser:passwd 为在 Doris 中创建的用户。初始用户为 admin / root密码初始状态下为空。 host:port 为 BE 的 HTTP 协议端口默认是 8040可以在 Doris 集群 WEB UI页面查看。 label: 可以在 Header 中指定 Label 唯一标识这个导入任务。
3、等待导入结果
-- 这是失败的
[rootzuomm01 data]# curl \-u root:123456 \-H label:load_local_file \-T /root/data/loadfile.txt \http://zuomm01:8040/api/test/load_local_file_test/_stream_load
{TxnId: 1004,Label: load_local_file,TwoPhaseCommit: false,Status: Fail,Message: too many filtered rows,NumberTotalRows: 4,NumberLoadedRows: 0,NumberFilteredRows: 4,NumberUnselectedRows: 0,LoadBytes: 36,LoadTimeMs: 82,BeginTxnTimeMs: 13,StreamLoadPutTimeMs: 56,ReadDataTimeMs: 0,WriteDataTimeMs: 9,CommitAndPublishTimeMs: 0,ErrorURL: http://192.168.17.3:8040/api/_load_error_log?file__shard_0/error_log_insert_stmt_cf4aa4d10e8d5fc5-458f16b70f0f2e87_cf4aa4d10e8d5fc5_458f16b70f0f2e87
}-- 这是成功的
[rootzuomm01 data]# curl \-u root:123456 \-H label:load_local_file \-H column_separator:, \-T /root/data/loadfile.txt \http://zuomm01:8040/api/test/load_local_file_test/_stream_load
{TxnId: 1005,Label: load_local_file,TwoPhaseCommit: false,Status: Success,Message: OK,NumberTotalRows: 4,NumberLoadedRows: 4,NumberFilteredRows: 0,NumberUnselectedRows: 0,LoadBytes: 36,LoadTimeMs: 54,BeginTxnTimeMs: 0,StreamLoadPutTimeMs: 2,ReadDataTimeMs: 0,WriteDataTimeMs: 14,CommitAndPublishTimeMs: 36
}4、curl的一些可配置的参数
1、label: 导入任务的标签相同标签的数据无法多次导入。标签默认保留30分钟
2、column_separator用于指定导入文件中的列分隔符默认为\t。
3、line_delimiter用于指定导入文件中的换行符默认为\n。
4、columns用于指定文件中的列和table中列的对应关系默认一一对应 例1: 表中有3个列“c1, c2, c3”源文件中的三个列一次对应的是c3,c2,c1; 那么需要指定-H “columns: c3, c2, c1” 例2: 表中有3个列“c1, c2, c3, 源文件中前三列依次对应但是有多余1列那么需要指定-H “columns: c1, c2, c3, xxx”;最后一个列随意指定个名称占位即可 例3: 表中有3个列“year, month, day三个列源文件中只有一个时间列为”2018-06-01 01:02:03“格式那么可以指定 -H columns: col, year year(col), monthmonth(col), dayday(col)完成导入
5、where: 用来过滤导入文件中的数据
例1: 只导入大于k1列等于20180601的数据那么可以在导入时候指定-H “where: k1 20180601”
6、max_filter_ratio最大容忍可过滤数据不规范等原因的数据比例。默认零容忍。数据不规范不包括通过 where 条件过滤掉的行。
7、partitions: 用于指定这次导入所设计的partition。如果用户能够确定数据对应的partition推荐指定该项。不满足这些分区的数据将被过滤掉。 比如指定导入到p1, p2分区
-H “partitions: p1, p2”
8、timeout: 指定导入的超时时间。单位秒。默认是 600 秒。可设置范围为 1 秒 ~ 259200 秒。
9、timezone: 指定本次导入所使用的时区。默认为东八区。该参数会影响所有导入涉及的和时区有关的函数结果。
10、exec_mem_limit: 导入内存限制。默认为 2GB。单位为字节。
11、format: 指定导入数据格式默认是csv支持json格式。
12、read_json_by_line: 布尔类型为true表示支持每行读取一个json对象默认值为false。
13、merge_type: 数据的合并类型一共支持三种类型APPEND、DELETE、MERGE 其中APPEND是默认值表示这批数据全部需要追加到现有数据中DELETE 表示删除与这批数据key相同的所有行MERGE 语义 需要与delete 条件联合使用表示满足delete 条件的数据按照DELETE 语义处理其余的按照APPEND 语义处理 示例-H “merge_type: MERGE” -H “delete: flag1”
14、delete: 仅在 MERGE下有意义 表示数据的删除条件
15、function_column.sequence_col: 只适用于UNIQUE_KEYS,相同key列下保证value列按照source_sequence列进行REPLACE, source_sequence可以是数据源中的列也可以是表结构中的一列
-- 准备数据
{id:1,name:liuyan,age:18}
{id:2,name:tangyan,age:18}
{id:3,name:jinlian,age:18}
{id:4,name:dalang,age:18}
{id:5,name:qingqing,age:18}curl \-u root: \-H label:load_local_file_json_20221126 \-H columns:id,name,age \-H max_filter_ratio:0.1 \-H timeout:1000 \-H exec_mem_limit:1G \-H where:id1 \-H format:json \-H read_json_by_line:true \-H merge_type:delete \-T /root/data/json.txt \http://zuomm01:8040/api/test/load_local_file_test/_stream_load-H merge_type:append \# 会把id 3 的这条数据删除-H merge_type:MERGE \-H delete:id3导入建议
1、Stream Load 只能导入本地文件。 2、建议一个导入请求的数据量控制在 1 - 2 GB 以内。如果有大量本地文件可以分批并发提交。
4.3、外部存储数据导入(hdfs)
1、适用场景
1、源数据在 Broker 可以访问的存储系统中如 HDFS。
2、数据量在几十到百 GB 级别。
2、基本原理
1、创建提交导入的任务
2、FE生成执行计划并将执行计划分发到多个BE节点上(每个BE节点都导入一部分数据)
3、BE收到执行计划后开始执行从broker上拉取数据到自己的节点上
4、所有BE都完成后FE决定是否导入成功返回结果给客户端 | 1. user create broker loadv--------| || 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 |
----------------------------------1、新建一张表
drop table if exists load_hdfs_file_test;
CREATE TABLE IF NOT EXISTS load_hdfs_file_test
(id INT,name VARCHAR(50),age TINYINT
)
unique key(id)
DISTRIBUTED BY HASH(id) BUCKETS 3;2、将本地的数据导入到hdfs上面
hadoop fs -put ./loadfile.txt hdfs://zuomm01:8020/
hadoop fs -ls hdfs://zuomm01:8020/ 3、导入格式
语法示例
LOAD LABEL test.label_202204(
[MERGE|APPEND|DELETE] -- 不写就是append
DATA INFILE
(
file_path1[, file_path2, ...] -- 描述数据的路径 这边可以写多个 以逗号分割
)
[NEGATIVE] -- 负增长
INTO TABLE table_name -- 导入的表名字
[PARTITION (p1, p2, ...)] -- 导入到哪些分区不符合这些分区的就会被过滤掉
[COLUMNS TERMINATED BY column_separator] -- 指定分隔符
[FORMAT AS file_type] -- 指定存储的文件类型
[(column_list)] -- 指定导入哪些列 [COLUMNS FROM PATH AS (c1, c2, ...)] -- 从路劲中抽取的部分列
[SET (column_mapping)] -- 对于列可以做一些映射写一些函数
-- 这个参数要写在要写在set的后面
[PRECEDING FILTER predicate] -- 做一些过滤
[WHERE predicate] -- 做一些过滤 比如id10 [DELETE ON expr] --根据字段去做一些抵消消除的策略 需要配合MERGE
[ORDER BY source_sequence] -- 导入数据的时候保证数据顺序
[PROPERTIES (key1value1, ...)] -- 一些配置参数4、将hdfs上的数据load到表中
LOAD LABEL test.label_20221125
(
DATA INFILE(hdfs://zuomm01:8020/test.txt)
INTO TABLE load_hdfs_file_test
COLUMNS TERMINATED BY ,
(id,name,age)
)
with HDFS (
fs.defaultFShdfs://zuomm01:8020,
hadoop.usernameroot
)
PROPERTIES
(
timeout1200,
max_filter_ratio0.1
);这是一个异步的操作所以需要去查看下执行的状态
show load order by createtime desc limit 1\G;*************************** 1. row ***************************JobId: 12143Label: label_20220402State: FINISHEDProgress: ETL:100%; LOAD:100%Type: BROKEREtlInfo: unselected.rows0; dpp.abnorm.ALL0; dpp.norm.ALL4TaskInfo: cluster:N/A; timeout(s):1200; max_filter_ratio:0.1ErrorMsg: NULLCreateTime: 2022-08-31 01:36:01EtlStartTime: 2022-08-31 01:36:03EtlFinishTime: 2022-08-31 01:36:03LoadStartTime: 2022-08-31 01:36:03
LoadFinishTime: 2022-08-31 01:36:03URL: NULLJobDetails: {Unfinished backends:{702bc3732d804f60-aa4593551c6e577a:[]},ScannedRows:4,TaskNumber:1,LoadBytes:139,All backends:{702bc3732d804f60-aa4593551c6e577a:[10004]},FileNumber:1,FileSize:36}TransactionId: 1007ErrorTablets: {}
1 row in set (0.00 sec)--失败的案例会有详细的错误信息可以参考参考mysql show load order by createtime desc limit 1\G;
*************************** 1. row ***************************JobId: 12139Label: label_20220402State: CANCELLEDProgress: ETL:N/A; LOAD:N/AType: BROKEREtlInfo: NULLTaskInfo: cluster:N/A; timeout(s):1200; max_filter_ratio:0.1ErrorMsg: type:LOAD_RUN_FAIL; msg:errCode 2, detailMessage connect failed. hdfs://zuomm01CreateTime: 2022-08-31 01:32:16EtlStartTime: 2022-08-31 01:32:19EtlFinishTime: 2022-08-31 01:32:19LoadStartTime: 2022-08-31 01:32:19
LoadFinishTime: 2022-08-31 01:32:19URL: NULLJobDetails: {Unfinished backends:{4bd307c0bd564c45-b7df986d26569ffa:[]},ScannedRows:0,TaskNumber:1,LoadBytes:0,All backends:{4bd307c0bd564c45-b7df986d26569ffa:[10004]},FileNumber:1,FileSize:36}TransactionId: 1006ErrorTablets: {}
1 row in set (0.01 sec)3、load 参数说明
1、load_label导入任务的唯一 Label
2、[MERGE|APPEND|DELETE]数据合并类型。默认为 APPEND表示本次导入是普通的追加写操作。MERGE 和 DELETE 类型仅适用于 Unique Key 模型表。其中 MERGE 类型需要配合 [DELETE ON] 语句使用以标注 Delete Flag 列。而 DELETE 类型则表示本次导入的所有数据皆为删除数据
3、DATA INFILE被导入文件的路径可以为多个。
4、NEGTIVE该关键词用于表示本次导入为一批”负“导入。这种方式仅针对具有整型 SUM 聚合类型的聚合数据表。该方式会将导入数据中SUM 聚合列对应的整型数值取反。主要用于冲抵之前导入错误的数据。
5、PARTITION(p1, p2, …)可以指定仅导入表的某些分区。不再分区范围内的数据将被忽略。
6、COLUMNS TERMINATED BY指定列分隔符
7、FORMAT AS指定要导入文件的类型支持 CSV、PARQUET 和 ORC 格式。默认为 CSV。
8、column list用于指定原始文件中的列顺序。
9、COLUMNS FROM PATH AS指定从导入文件路径中抽取的列。
10、PRECEDING FILTER前置过滤条件。数据首先根据 column list 和 COLUMNS FROM PATH AS 按顺序拼接成原始数据行。然后按照前置过滤条件进行过滤。
11、SET (column_mapping)指定列的转换函数。
12、WHERE predicate根据条件对导入的数据进行过滤。
13、DELETE ON expr需配合 MEREGE 导入模式一起使用仅针对 Unique Key 模型的表。用于指定导入数据中表示 Delete Flag 的列和计算关系。
15、load_properties:指定导入的相关参数。目前支持以下参数
timeout:导入超时时间。默认为 4 小时。单位秒。max_filter_ratio:最大容忍可过滤数据不规范等原因的数据比例。默认零容忍。取值范围为0到1。exec_mem_limit:导入内存限制。默认为 2GB。单位为字节。strict_mode:是否对数据进行严格限制。默认为 false。timezone:指定某些受时区影响的函数的时区如strftime/alignment_timestamp/from_unixtime 等等具体请查阅 时区 文档。如果不指定则使用 “Asia/Shanghai” 时区
4、进阶示例
从 HDFS 导入数据使用通配符匹配两批两批文件。分别导入到两个表中
LOAD LABEL example_db.label2
(DATA INFILE(hdfs://hdfs_host:hdfs_port/input/file-10*)INTO TABLE my_table1PARTITION (p1)COLUMNS TERMINATED BY ,FORMAT AS parquet (id, tmp_salary, tmp_score) SET (salary tmp_salary 1000,score tmp_score 10),DATA INFILE(hdfs://hdfs_host:hdfs_port/input/file-20*)INTO TABLE my_table2COLUMNS TERMINATED BY ,(k1, k2, k3)
)
with HDFS (
fs.defaultFShdfs://zuomm01:8020,
hadoop.usernameroot
)导入数据并提取文件路径中的分区字段
LOAD LABEL example_db.label10
(DATA INFILE(hdfs://hdfs_host:hdfs_port/user/hive/warehouse/table_name/dt20221125/*)INTO TABLE my_tableFORMAT AS csv(k1, k2, k3)COLUMNS FROM PATH AS (dt)
)
WITH BROKER hdfs
(usernamehdfs_user,passwordhdfs_password
);对待导入数据进行过滤。
LOAD LABEL example_db.label6
(DATA INFILE(hdfs://host:port/input/file)INTO TABLE my_table(k1, k2, k3)SET (k2 k2 1)PRECEDING FILTER k1 1 》前置过滤WHERE k1 k2 》 后置过滤
)
WITH BROKER hdfs
(usernameuser,passwordpass
);只有原始数据中k1 1并且转换后k1 k2 的行才会被导入。5、取消导入
当 Broker load 作业状态不为 CANCELLED 或 FINISHED 时可以被用户手动取消。
取消时需要指定待取消导入任务的 Label 。取消导入命令语法可执行 HELP CANCEL LOAD 查看。
CANCEL LOAD [FROM db_name] WHERE LABELload_label; 4.4、通过外部表同步数据
Doris 可以创建外部表。创建完成后可以通过 SELECT 语句直接查询外部表的数据也可以通过 INSERT INTO SELECT 的方式导入外部表的数据。
Doris 外部表目前支持的数据源包括MySQLOracleHivePostgreSQLSQLServerIcebergElasticSearch
CREATE [EXTERNAL] TABLE table_name ( col_name col_type [NULL | NOT NULL] [COMMENT comment]
) ENGINEHIVE
[COMMENT comment]
PROPERTIES (
-- 我要映射的hive表在哪个库里面
-- 映射的表名是哪一张
-- hive的元数据服务地址property_nameproperty_value, ...
); 参数说明
1、外表列
列名要与 Hive 表一一对应列的顺序需要与 Hive 表一致必须包含 Hive 表中的全部列Hive 表分区列无需指定与普通列一样定义即可。
2、ENGINE 需要指定为 HIVE
3、PROPERTIES 属性
hive.metastore.urisHive Metastore 服务地址database挂载 Hive 对应的数据库名table挂载 Hive 对应的表名
1、使用示例
在Hive 中创建一个测试用表
CREATE TABLE user_info ( id int, name string, age int
) stored as orc;insert into user_info values (1,zss,18);
insert into user_info values (2,lss,20);
insert into user_info values (3,ww,25);Doris 中创建外部表
CREATE EXTERNAL TABLE hive_user_info ( id int, name varchar(10), age int
) ENGINEHIVE
PROPERTIES (
hive.metastore.uris thrift://linux01:9083,
database db1,
table user_info
);查询外部表
select * from hive_user_info;将数据从外部表导入内部表
-- 就是用sql查询从外部表中select出数据后insert到内部表即可
insert into doris_user_info
select*
from hive_user_info;Hive 表 Schema 变更不会自动同步需要在 Doris 中重建 Hive 外表。
当前 Hive 的存储格式仅支持 TextParquet 和 ORC 类型
4.5、Binlog Load
Binlog Load提供了一种使Doris增量同步用户在Mysql数据库中对数据更新操作的CDC(Change Data Capture)功能。 1、适用场景
1、INSERT/UPDATE/DELETE支持
2、过滤Query
3、暂不兼容DDL语句
2、基本原理
当前版本设计中Binlog Load需要依赖canal作为中间媒介让canal伪造成一个从节点去获取Mysql主节点上的Binlog并解析再由Doris去获取Canal上解析好的数据主要涉及Mysql端、Canal端以及Doris端总体数据流向如下
---------------------------------------------
| Mysql |
--------------------------------------------| Binlog
----------------------v----------------------
| Canal Server 数据解析 |
------------------------^-------------------Get | | Ack
-------------------|-----|-------------------
| FE | | |
| -----------------|-----|---------------- |
| | Sync Job | | | |
| | ------------v---------------- | |
| | | Canal Client | | |
| | | ----------------------- | | |
| | | | Receiver | | | |
| | | ----------------------- | | |
| | | ----------------------- | | |
| | | | Consumer | | | |
| | | ----------------------- | | |
| | ------------------------------ | |
| ------------------------------------- |
| | | | |
| ----v----- -----v---- -----v---- |
| | Channel1 | | Channel2 | | Channel3 | |
| | [Table1] | | [Table2] | | [Table3] | |
| --------- --------- --------- |
| | | | |
| --|------- ---|------ ---|------|
| ---v------| ----v-----| ----v-----||
| ----------| ----------| ----------||
| | Task | | Task | | Task | |
| ---------- ---------- ---------- |
--------------------------------------------| | |
----v-----------------v------------------v---
| Coordinator |
| BE |
------------------------------------------| | |
----v--- ---v---- ----v---
| BE | | BE | | BE |
-------- -------- --------如上图用户向FE提交一个数据同步作业。
1、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提交事务。
6、当且仅当所有channel都提交成功才会通过Ack命令通知canal并继续获取并消费数据。
7、如果有任意channel提交失败将会重新从上一次消费成功的位置获取数据并再次提交已提交成功的channel不会再次提交以保证幂等性。
8、整个数据同步作业中FE通过以上流程不断的从canal获取数据并提交到BE来完成数据同步。
3、配置Mysql端
在Mysql Cluster模式的主从同步中二进制日志文件(Binlog)记录了主节点上的所有数据变化数据在Cluster的多个节点间同步、备份都要通过Binlog日志进行从而提高集群的可用性。架构通常由一个主节点(负责写)和一个或多个从节点(负责读)构成,所有在主节点上发生的数据变更将会复制给从节点。
注意目前必须要使用Mysql 5.7及以上的版本才能支持Binlog Load功能。
1、打开mysql的二进制binlog日志功能则需要编辑my.cnf配置文件设置一下。
[rootzuomm01 sbin]# find / -name my.cnf
/etc/my.cnf
修改mysqld中的一些配置文件
[mysqld]
server_id 1
log-bin mysql-bin
binlog-format ROW#binlog-format 的三种模式
#ROW 记录每一行数据的信息
#Statement 记录sql语句
#Mixed 上面两种的混合2、重启 MySQL 使配置生效
systemctl restart mysqld 3、创建用户并授权
-- 设置这些参数可以使得mysql的密码简单化
set global validate_password_length4;
set global validate_password_policy0;
-- 新增一个canal的用户让他监听所有库中的所有表并且设置密码为canal
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO canal% IDENTIFIED BY canal ;
-- 刷新一下权限
FLUSH PRIVILEGES;4、准备测试表
CREATE TABLE user_doris (id int(11) NOT NULL AUTO_INCREMENT,name varchar(255) DEFAULT NULL,age int(11) DEFAULT NULL,gender varchar(255) DEFAULT NULL,PRIMARY KEY (id)
) ENGINEInnoDB AUTO_INCREMENT4 DEFAULT CHARSETutf85、配置 Canal 端
Canal 是属于阿里巴巴 otter 项目下的一个子项目主要用途是基于 MySQL 数据库增量日志解析提供增量数据订阅和消费用于解决跨机房同步的业务场景建议使用 canal 1.1.5及以上版本。
下载地址https://github.com/alibaba/canal/releases
上传并解压 canal deployer压缩包
mkdir /opt/app/canal
tar -zxvf canal.deployer-1.1.5.tar.gz -C /opt/app/canal在 conf 文件夹下新建目录并重命名
一个 canal 服务中可以有多个 instanceconf/下的每一个目录即是一个实例每个实例下面都有独立的配置文件
mkdir /opt/app/canel/conf/doris
拷贝配置文件模板
cp /opt/app/canal/conf/example/instance.properties /opt/app/canal/conf/doris/修改 conf/canal.properties 的配置修改 instance 配置文件
vi canal.properties 进入找到canal.destinations example
将其修改为 我们自己配置的目录
canal.destinations doris-load 启动
sh bin/startup.sh6、配置目标表
Doris 创建与 Mysql 对应的目标表
CREATE TABLE binlog_mysql ( id int(11) NOT NULL COMMENT , name VARCHAR(50) NOT NULL COMMENT , age int(11) NOT NULL COMMENT ,gender VARCHAR(50) NOT NULL COMMENT
) ENGINEOLAP
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1; 基本语法
CREATE SYNC [db.]job_name
( channel_desc, channel_desc ...
)
binlog_desc 参数说明
1、job_name是数据同步作业在当前数据库内的唯一标识
2、channel_desc 用来定义任务下的数据通道可表示 MySQL 源表到 doris 目标表的映射关系。在设置此项时如果存在多个映射关系必须满足 MySQL 源表应该与 doris 目标表是一一对应关系其他的任何映射关系如一对多关系检查语法时都被视为不合法。
3、column_mapping主要指MySQL源表和doris目标表的列之间的映射关系如果不指定FE 会默认源表和目标表的列按顺序一一对应。但是我们依然建议显式的指定列的映射关系这样当目标表的结构发生变化比如增加一个 nullable 的列数据同步作业依然可以进行。否则当发生上述变动后因为列映射关系不再一一对应导入将报错。
4、binlog_desc定义了对接远端 Binlog 地址的一些必要信息目前可支持的对接类型只有 canal 方式所有的配置项前都需要加上 canal 前缀。
canal.server.ip: canal server 的地址canal.server.port: canal server 的端口canal.destination: 前文提到的 instance 的字符串标识canal.batchSize: 每批从 canal server 处获取的 batch 大小的最大值默认 8192canal.username: instance 的用户名canal.password: instance 的密码canal.debug: 设置为 true 时会将 batch 和每一行数据的详细信息都打印出来会影响性能。
CREATE SYNC test.job20221228
( FROM test.binlog_test INTO binlog_test
)
FROM BINLOG
( type canal, canal.server.ip zuomm01, canal.server.port 11111, canal.destination doris, canal.username canal, canal.password canal
);查看作业状态
展示当前数据库的所有数据同步作业状态。
SHOW SYNC JOB;
展示数据库 test_db 下的所有数据同步作业状态。
SHOW SYNC JOB FROM test; 控制作业用户可以通过 STOP/PAUSE/RESUME 三个命令来控制作业的停止暂停和恢复
停止名称为 job_name 的数据同步作业
STOP SYNC JOB [db.]job_name 暂停名称为 job_name 的数据同步作业
PAUSE SYNC JOB [db.]job_name 恢复名称为 job_name 的数据同步作业
RESUME SYNC JOB job_name 4.6、导出数据
数据导出Export是 Doris 提供的一种将数据导出的功能。该功能可以将用户指定的表或分区的数据以文本的格式通过 Broker 进程导出到远端存储上如 HDFS / 对象存储支持S3协议 等。
EXPORT TABLE test.event_info_log1 -- 库名.表名
to hdfs://linux01:8020/event_info_log1 -- 导出到那里去
PROPERTIES
(label event_info_log1,column_separator,,exec_mem_limit2147483648,timeout 3600
)
WITH BROKER broker_name
(username root,password
);查看导出状态
mysql show EXPORT \G;
*************************** 1. row ***************************JobId: 14008State: FINISHEDProgress: 100%TaskInfo: {partitions:[*],exec mem limit:2147483648,column separator:,,line delimiter:\n,tablet num:1,broker:hdfs,coord num:1,db:default_cluster:db1,tbl:tbl3}Path: bos://bj-test-cmy/export/
CreateTime: 2019-06-25 17:08:24StartTime: 2019-06-25 17:08:28
FinishTime: 2019-06-25 17:08:34Timeout: 3600ErrorMsg: NULL
1 row in set (0.01 sec)参数说明
JobId作业的唯一 IDState作业状态 PENDING作业待调度EXPORTING数据导出中FINISHED作业成功CANCELLED作业失败 Progress作业进度。该进度以查询计划为单位。假设一共 10 个查询计划当前已完成 3 个则进度为 30%。TaskInfo以 Json 格式展示的作业信息 db数据库名tbl表名partitions指定导出的分区。* 表示所有分区。exec mem limit查询计划内存使用限制。单位字节。column separator导出文件的列分隔符。line delimiter导出文件的行分隔符。tablet num涉及的总 Tablet 数量。broker使用的 broker 的名称。coord num查询计划的个数。 Path远端存储上的导出路径。CreateTime/StartTime/FinishTime作业的创建时间、开始调度时间和结束时间。Timeout作业超时时间。单位是秒。该时间从 CreateTime 开始计算。ErrorMsg如果作业出现错误这里会显示错误原因。
注意事项
1、不建议一次性导出大量数据。一个 Export 作业建议的导出数据量最大在几十 GB。过大的导出会导致更多的垃圾文件和更高的重试成本。
2、如果表数据量过大建议按照分区导出。
3、在 Export 作业运行过程中如果 FE 发生重启或切主则 Export 作业会失败需要用户重新提交。
4、如果 Export 作业运行失败在远端存储中产生的 __doris_export_tmp_xxx 临时目录以及已经生成的文件不会被删除需要用户手动删除。
5、如果 Export 作业运行成功在远端存储中产生的 __doris_export_tmp_xxx 目录根据远端存储的文件系统语义可能会保留也可能会被清除。比如在百度对象存储BOS中通过 rename 操作将一个目录中的最后一个文件移走后该目录也会被删除。如果该目录没有被清除用户可以手动清除
6、当 Export 运行完成后成功或失败FE 发生重启或切主则 SHOW EXPORT展示的作业的部分信息会丢失无法查看。
7、Export 作业只会导出 Base 表的数据不会导出 Rollup Index 的数据。
8、Export 作业会扫描数据占用 IO 资源可能会影响系统的查询延迟
原文https://mp.weixin.qq.com/s/kjCiPfNDT27KJq4iEN0LYw
