Clickhouse

ClickHouse更快查询的原因

clickhouse官网介绍

1. 支持列式存储，更方便地读取所需要的列
2. 支持创建索引，支持基于内存的索引不仅可以存储需要的列，还可以存储一定范围内的列
3. 数据压缩，除了通用的压缩，ClickHouse支持专门的编解码器，可以使数据更加紧凑。
4. 向量化的查询
5. 查询时支持语句级别的多线程，最大程度的使用节点的资源，物尽其用

项目选用clickhouse原因

1. 使用clickhouse存储dwd轻度汇总后的dws数据，使用flink写出clickhouse
2. 优点
   • 语句级别多线程 clickhouse在写入数据时会使用cpu的多核同时处理写入各个表分区数据，插入快
   • 支持sql，方便后期数据再次汇总
   • 插件式的表引擎，类型丰富，支持场景多
   • 支持表的TTL
   • clickhouse支持列式存储

CK实际数据存储结构

博客园ck数据储存优质文章

shell复制自动换行：关放大阅读展开代码
[root@hadoop162 20221202_1_2_1]# pwd
/var/lib/clickhouse/data/gmall2022/dws_trade_cart_add_uu_window/20221202_1_2_1
[root@hadoop162 20221202_1_2_1]# ll
total 36
-rw-r-----. 1 clickhouse clickhouse 252 Dec 13 08:31 checksums.txt
-rw-r-----. 1 clickhouse clickhouse 103 Dec 13 08:31 columns.txt
-rw-r-----. 1 clickhouse clickhouse   1 Dec 13 08:31 count.txt
-rw-r-----. 1 clickhouse clickhouse 151 Dec 13 08:31 data.bin
-rw-r-----. 1 clickhouse clickhouse 144 Dec 13 08:31 data.mrk3
-rw-r-----. 1 clickhouse clickhouse  10 Dec 13 08:31 default_compression_codec.txt
-rw-r-----. 1 clickhouse clickhouse   8 Dec 13 08:31 minmax_stt.idx
-rw-r-----. 1 clickhouse clickhouse   4 Dec 13 08:31 partition.dat
-rw-r-----. 1 clickhouse clickhouse  16 Dec 13 08:31 primary.idx

主要的文件：

1. data.bin 存储表中每列的数据
2. primary.idx 表的主键，没有设置primary key，默认使用order by字段作为主键
3. data.mrk3 标记文件，用于联系primary.idx中的稀疏索引和列式存储数据
   

项目CK使用ReplacingMergeTree

1. 使用ReplacingMergeTree原因
   在做DWS层数据汇总的时候，将汇总数据写出到ClickHouse，需要将数据按照维度分析并聚合，根据Flink动态表的概念，可能会对刚刚聚合的结果做修改

java复制自动换行：关放大阅读展开代码
// 统计5s内的词频
Table result = tEnv.sqlQuery("select " +
        " date_format(window_start, 'yyyy-MM-dd HH:mm:ss') stt, " +
        " date_format(window_end, 'yyyy-MM-dd HH:mm:ss') edt, " +
        " kw keyword, " +
        " count(*) keyword_count, " +
        " unix_timestamp()*1000 as ts " +   // et as ...
        "from  table(TUMBLE(TABLE ana_table, DESCRIPTOR(et), interval '5' second )) " +
        "group by kw,window_start,window_end");
// 将Flink中的动态表转化为Stream，才能写入ClickHouse
SingleOutputStreamOperator<KeywordBean> map = tEnv.toRetractStream(result, KeywordBean.class)
    // 过滤到回撤的数据，即出现聚合结果更新，RetractStream会有更新前和更新后概念；这里只需要更新后的数据
        // Tuple<Boolean,KeywordBean>,Boolean为false，标识更新前的数据
        .filter(w -> w.f0)
        //过滤掉Boolean标识
        .map(w -> w.f1);

clickhouse规模

1. 建表时，ReplicatedReplacingMergeTree，具有表副本和数据去重功能
2. 2台（副本），使用zookeeper组合使用，dws层每天大概20G数据，每年大概20T
   那么通过过滤，Ck中可能存在相同维度下的出现更新的聚合结果，此时就需要去重！！
   可以借助ReplacingMergeTree引擎实现后台的数据去重。 依托zookeeper实现数据副本功能，大概原理：
   
3. ReplacingMergeTree去重原理
   主要通过建表时指定的order by字段和指定的版本字段实现，ReplacingMergeTree() 填入的参数为版本字段，重复数据保留版本字段值最大的。如果不填版本字段，默认保留最后一条

sql复制自动换行：关放大阅读展开代码
create table if not exists dws_traffic_keyword_page_view_window
(
    stt           DateTime COMMENT '窗口起始时间',
    edt           DateTime COMMENT '窗口结束时间',
    keyword       String COMMENT '关键词',
    keyword_count UInt64 COMMENT '关键词出现频次',
    ts            UInt64 COMMENT '时间戳'
) engine = ReplacingMergeTree(ts)
      partition by toYYYYMMDD(stt)
      order by (stt, edt, keyword);

常见OLAP数据库的比较

ClickHouse,ElasticSearch,Drios

1. ClickHouse 适合处理大量的列式数据，特别是对于需要快速查询和数据处理的应用场景
2. Elasticsearch 更适合处理大量的文档数据，特别是对于需要实时搜索和分析的应用场景
3. Drios 需要处理大规模的列式数据，特别是针对 OLAP 和 BI 场景

常见的ClickHouse表引擎

1. MergeTree家族中的MergeTree，ReplicatedReplacingMergeTree,ReplacingMergeTree,SummingMergeTree
2. Log家族的TinyLog，Log
3. 和其他存储结合的引擎，比如Kafka，Hudi，MySQL，Hive等
4. 一些特殊场景下的引擎Distributed，Buffer

稀疏索引和跳数索引

官方文档相关介绍

sql复制自动换行：关放大阅读展开代码
-- 包含稀疏索引（其中的最大最小值索引）的建表语句
CREATE TABLE default.mt  
(  
    `a` Int32,  
    `b` Int32,  
    `c` Int32,  
    INDEX `idx_c` (c) TYPE minmax GRANULARITY 1   -- 为c字段建立maxmin索引，加速c的查询
)  
ENGINE = MergeTree  -- 使用的表引擎
PARTITION BY a   --分区字段，建表可选
ORDER BY b  -- 排序字段
SETTINGS index_granularity=3 -- 主键索引（无则为order by字段）的稀疏索引粒度，默认为8192

稀疏索引

plaintext复制自动换行：关放大阅读展开代码
默认为表的主键，没有主键则为order by字段建立稀疏索引，即默认每8192条记录，取一个主键，此处对比稠密索引

跳数索引

1. 建表

sql复制自动换行：关放大阅读展开代码
CREATE TABLE skip_table
(
  my_key UInt64,
  my_value UInt64
)
ENGINE MergeTree primary key my_key
SETTINGS index_granularity=8192;
INSERT INTO skip_table SELECT number, intDiv(number,4096) FROM numbers(100000000);

2. 查询

sql复制自动换行：关放大阅读展开代码
SELECT * FROM skip_table WHERE my_value IN (125, 700)
┌─my_key─┬─my_value─┐
│ 512000 │      125 │
│ 512001 │      125 │
│    ... |      ... |
└────────┴──────────┘
8192 rows in set. Elapsed: 0.079 sec. Processed 100.00 million rows, 800.10 MB (1.26 billion rows/s., 10.10 GB/s.
-- 可以看出进行全表扫描

3. 解决手段

• 跳数索引
• 创建具有不同主键的新表，但需要自己维护原表改变带来的数据一致性问题
• 基于原表创建materialized view，clickhouse会自动维护新表和原表的数据一致
• 基于原表添加projection ，还是实验性的功能，相当于在materialized view基础上，根据SQL语句自动选择索引使用顺序 此处考虑建立跳数索引的方式

sql复制自动换行：关放大阅读展开代码
-- 为原表添加名为vix  的跳数索引，set(100)表示这个跳数索引最多存储100个值，索引粒度为2个GRANULARITY 
ALTER TABLE skip_table ADD INDEX vix my_value TYPE set(100) GRANULARITY 2;
-- 使得新添加的跳数索引作用所有新老数据
ALTER TABLE skip_table MATERIALIZE INDEX vix;
-- 再次查询
SELECT * FROM skip_table WHERE my_value IN (125, 700)
┌─my_key─┬─my_value─┐
│ 512000 │      125 │
│ 512001 │      125 │
│    ... |      ... |
└────────┴──────────┘
8192 rows in set. Elapsed: 0.051 sec. Processed 32.77 thousand rows, 360.45 KB (643.75 thousand rows/s., 7.08 MB/s.)
-- 明显是范围查询了

4. 跳数索引大概原理图

mergeTree

sql复制自动换行：关放大阅读展开代码
drop table if exists dws_trade_order_window;
create table if not exists dws_trade_order_window
(
    stt                     DateTime COMMENT '窗口起始时间',
    edt                     DateTime COMMENT '窗口结束时间',
    order_unique_user_count UInt64 COMMENT '下单独立用户数',
    order_new_user_count    UInt64 COMMENT '下单新用户数',
    ts                      UInt64 COMMENT '时间戳'
)-- # ReplacingMergeTree 和  ReplacingMergeTree 组合
engine = ReplicatedReplacingMergeTree(ts)
        partition by toYYYYMMDD(stt)
        order by (stt, edt);

其他的引擎：

1. SummingMergeTree 对指定的列写入时预聚合
2. Log
3. Merge 纯内存读写

clickhouse优化

1. 单台内存（128G）分配，ck会尽量占用全部的内存，可以设置其使用上限，比如100G

xml复制自动换行：关放大阅读展开代码
-- 默认使用节点90%的内存资源，可以适当调整到70%
<max_server_memory_usage_to_ram_ratio>0.9</max_server_memory_usage_to_ram_ratio>

2. 并发处理条数，虽然ck并不支持高并发，但可以适当提高并发处理能力

xml复制自动换行：关放大阅读展开代码
// 配置文件修改并发处理能力，每秒处理的条数 ，比如提高到300
<max_concurrent_queries>100</max_concurrent_queries>

3. ck保证最终的数据一致性，可能导致中间过程数据的不一致，查询前可以使用optmize将处在临时分区的
   刚插入数据合并到主分区；也可使用final关键字查询；或者手动group by ，聚合等

sql复制自动换行：关放大阅读展开代码
optimize dws_user_user_login_window;
select * from dws_user_user_login_window;

4. 写入时攒批，避免写入过快导致 too many parts

java复制自动换行：关放大阅读展开代码
 // 自定义clickhouse sinker 写数据到clickhouse
JdbcSink.sink(sql.toString(),  
                  new JdbcStatementBuilder<T>() {  
                      @Override  
                      public void accept(PreparedStatement ps,  
                                         T t) throws SQLException { 
                         // sql preparement设置                 
                  },  
                  new JdbcExecutionOptions.Builder() 
                      // 1M
                      .withBatchSize(1024 * 1024)  
                      // 5 seconds
                      .withBatchIntervalMs(5000)
                      // 重试次数  
                      .withMaxRetries(3)  
                      .build(),  
                  new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()  
                      .withDriverName(driver)  
                      .withUrl(url)  
                      .withUsername(user)  
                      .withPassword(password)  
                      .build()  
);