概述

都說“實(shí)踐是檢驗(yàn)真理的唯一標(biāo)準(zhǔn)”，光說不練假把式，那么本文就通過實(shí)際的測試來感受一下Doris和clickhouse在讀寫方面的性能差距，看看Doris盛名之下，是否真有屠龍之技；clickhouse長鋒出鞘，是否敢縛蒼龍？

廢話不多說，上貨。

硬件配置

在這里，我使用多臺物理機(jī)搭建了clickhouse和Doris集群。

clickhouse集群

Doris集群

clickhouse集群和Doris集群共用一套物理機(jī)。

數(shù)據(jù)準(zhǔn)備

由于clickhouse與Doris共用物理機(jī)資源，為了避免互相干擾，在Doris測試時，clickhouse集群停止一切讀寫操作；同理，當(dāng)clickhouse集群測試時，Doris集群也停止一切讀寫操作。

本次測試主要針對全文檢索的場景。測試數(shù)據(jù)為clickhouse-server的日志文件，三個節(jié)點(diǎn)上的日志，共計(jì)2億條數(shù)據(jù)，采集寫入kafka后數(shù)據(jù)量為30GB。

我們將數(shù)據(jù)通過采集器組織成json格式后，采集到kafka，數(shù)據(jù)結(jié)構(gòu)如下所示：

{
    "@message": "2023.12.14 05:25:20.983533 [ 243360 ] {} <Information> aimeter.apm_span_index_trace_id (ReplicatedMergeTreePartCheckThread): Checking if anyone has a part 20231104_7476_7590_23 or covering part.",
    "@@id": "cd8946be124a4079f4f372782ee6da1f",
    "@filehashkey": "484d4e40bf93db91e25fbdfb47f084fe",
    "@collectiontime": "2023-12-18T10:56:08.125+08:00",
    "@hostname": "ck93",
    "@path": "/data01/chenyc/logs/clickhouse-server/clickhouse-server.log.4",
    "@rownumber": 6,
    "@seq": 6,
    "@ip": "192.168.101.93",
    "@topic": "log_test"
}

主要測試數(shù)據(jù)寫入和數(shù)據(jù)查詢兩個方面。

預(yù)設(shè)查詢場景有如下幾個：

• 根據(jù)ip和path維度統(tǒng)計(jì)每個ip下path的個數(shù)
• 統(tǒng)計(jì)每個ip下的Error日志的數(shù)量
• 統(tǒng)計(jì)日志中出現(xiàn)Debug 和 cdb56920-2d39-4e6d-be99-dd6ef24cc66a 的條數(shù)
• 統(tǒng)計(jì)出現(xiàn)Trace和gauge.apm_service_span出現(xiàn)的次數(shù)
• 查詢Error中出現(xiàn)READ_ONLY的日志明細(xì)
• 查詢?nèi)罩局谐霈F(xiàn)"上海"關(guān)鍵字的日志么明細(xì)

主要測試的性能指標(biāo)包括：

• 寫入性能
  • 寫入速度
  • 寫入過程的資源占用，CPU負(fù)載
  • 寫入后數(shù)據(jù)的壓縮占比

• 查詢性能
  • 查詢耗時
  • 查詢的資源占用
  • 查詢命中索引的情況

另增加測試一邊寫入一遍查詢時對讀寫的影響。

Doris建表語句

建表語句如下：

CREATE database demo;
?
use demo;
?
CREATE TABLE demo.log_test (
    `@@id` CHAR(34) NOT NULL COMMENT "每行的唯一hash標(biāo)識id",
    `@message` STRING NOT NULL COMMENT "日志內(nèi)容",
    `@filehashkey` CHAR(34) NOT NULL COMMENT "每個文件的hash值，用于標(biāo)識文件唯一性",
    `@collectiontime` DATETIME(3) COMMENT "采集時間",
    `@hostname` VARCHAR(20) NOT NULL COMMENT "主機(jī)名",
    `@path` VARCHAR(256) NOT NULL COMMENT "文件路徑",
    `@rownumber` BIGINT NOT NULL COMMENT "行號",
    `@seq` BIGINT NOT NULL COMMENT "在同一個文件內(nèi)連續(xù)的序列號",
    `@ip` CHAR(16) NOT NULL COMMENT "節(jié)點(diǎn)IP",
    `@topic` CHAR(16) NOT NULL COMMENT "所屬kafka的topic",
    
    INDEX idx_message_inv(`@message`) USING INVERTED PROPERTIES(
        "parser" = "unicode",
        "parser_mode" = "fine_grained",
        "support_phrase" = "true"
    ) COMMENT "倒排索引",
    INDEX idx_message_ngram(`@message`) USING NGRAM_BF PROPERTIES("gram_size"="5", "bf_size"="4096") COMMENT 'ngram_bf 索引'
)
DUPLICATE KEY(`@@id`)
PARTITION BY RANGE(`@collectiontime`) ()
DISTRIBUTED BY HASH(`@@id`) BUCKETS AUTO
ROLLUP (
    r1 (`@message`),
    r2 (`@ip`, `@path`)
)
PROPERTIES (
    "dynamic_partition.enable" = "true",
    "dynamic_partition.time_unit" = "DAY",
    "dynamic_partition.start" = "-7",
    "dynamic_partition.end" = "3",
    "dynamic_partition.prefix" = "p",
    "dynamic_partition.buckets" = "32",
    "compression"="zstd"
);

說明如下：

• 按@collectiontime動態(tài)分區(qū)
• 默認(rèn)三個副本
• 在@message上創(chuàng)建一個unicode倒排索引，一個NGram BloomFilter索引
• 創(chuàng)建兩個ROLLUP， 用來重建前綴索引
• 壓縮方式采用ZSTD

clickhouse建表語句

--- 本地表
create table log_test on cluster abc (
    `@@id` String NOT NULL CODEC(ZSTD(1)),
    `@message` String NOT NULL CODEC(ZSTD(1)) ,
    `@filehashkey` String NOT NULL CODEC(ZSTD(1)) ,
    `@collectiontime` DateTime64(3) CODEC(DoubleDelta, LZ4),
    `@hostname` LowCardinality(String) NOT NULL CODEC(ZSTD(1)) ,
    `@path` String NOT NULL CODEC(ZSTD(1)) ,
    `@rownumber` Int64 NOT NULL ,
    `@seq` Int64 NOT NULL ,
    `@ip` LowCardinality(String) NOT NULL CODEC(ZSTD(1)) ,
    `@topic` LowCardinality(String) NOT NULL CODEC(ZSTD(1)) ,
?
    INDEX message_idx `@message` TYPE ngrambf_v1(5, 65535, 1, 0) GRANULARITY 1,
    PROJECTION p_cnt (
    SELECT `@ip`,  `@path`, count() GROUP BY `@ip`, `@path`
  ) 
)ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMMDD(`@collectiontime`)
ORDER BY (`@collectiontime`, `@ip`, `@path`);
  
--- 分布式表
create table dist_log_test on cluster abc as log_test engine = Distributed('abc', 'default', 'log_test')

說明如下：

• string字段使用ZSTD壓縮，時間字段使用DoubleDelta壓縮
• 在@message字段是創(chuàng)建一個ngrambf_v1的二級索引
• 創(chuàng)建一個projection用于根據(jù)ip和path維度的預(yù)聚合
• 根據(jù)@collectiontime字段按天做分區(qū)

寫入性能

clickhouse

使用clickhouse_sinker向ck集群寫入數(shù)據(jù)，為公平起見，clickhouse_sinker為單實(shí)例。

配置文件如下：

{
  "clickhouse": {
      "cluster": "abc",
      "db": "default",
      "hosts": [
                  ["192.168.101.93", "192.168.101.94"],
                  ["192.168.101.96", "192.168.101.97"]
      ],
      "port": 19000,
      "username": "default",
      "password": "123456",
      "maxOpenConns": 5,
      "retryTimes": 0
  },
  "kafka": {
      "brokers": "192.168.101.94:29092,192.168.101.96:29092,192.168.101.98:29092"
  },
    "tasks": [{
      "name": "cktest",
      "topic": "log_test",
      "earliest": true,
      "consumerGroup": "abc",
      "parser": "fastjson",
      "tableName": "log_test",
      "autoSchema": true,
      "dynamicSchema":{
        "enable": false
      },
      "prometheusSchema": false,
      "bufferSize": 1000000,
      "flushInterval": 10
   }],
  "logLevel": "info"
}

測試結(jié)果：

clickhouse中數(shù)據(jù)情況：

Doris

Doris使用Routine load向Doris集群寫入數(shù)據(jù)，由于有3個backend，開啟3個并發(fā)度。

Routine Load配置如下：

CREATE ROUTINE LOAD demo.doris_test ON log_test
COLUMNS(`@message`,`@@id`,`@filehashkey`,`@collectiontime`,`@hostname`,`@path`,`@rownumber`,`@seq`,`@ip`,`@topic`)
PROPERTIES
(
    "desired_concurrent_number"="3",
    "max_error_number" = "500",
    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200",
    "strict_mode" = "false",
    "format" = "json"
)
FROM KAFKA
(
    "kafka_broker_list" = "192.168.101.94:29092,192.168.101.96:29092,192.168.101.98:29092",
    "kafka_topic" = "log_test",
    "kafka_partitions" = "0,1,2,3,4,5",
    "kafka_offsets" = "0,0,0,0,0,0"
);

這里 max_error_number 設(shè)置了500， 意思是容忍task失敗500次。之所以設(shè)置這個容忍度，是因?yàn)镈oris采用RapidJSON庫解析JSON串，這個庫解析部分亂碼數(shù)據(jù)時會報(bào)錯：

Reason: Parse json data for JsonDoc failed. code: 10, error info: The input is not valid UTF-8. src line [{"@message":"2023.12.13 18:41:17.762463 [ 143536 ] {bcf2d3d6-a68a-46a3-a008-55399f6a596f} <Error> void DB::ParallelParsingInputFormat::onBackgroundException(size_t): Code: 27. DB::ParsingException: Cannot parse input: expected '\\t' before: 'd8caa60-f8f4-45a0-9e45-ee1a9344f774\\t200.188.12.25\\t\\\\N\\tssh\\t\\\\N\\t22\\tzx2\\t1\\t1\\t2\\t180\\t0\\tIT運(yùn)維管理平臺（e海智維）\\t010336\\t季楊\\t6339\\t運(yùn)維支持部\\t1\\t廣東?': \nRow 1:\nColumn 0,   name: id,           type: Int64,            parsed text: \"4\"\nColumn 1,   name: deviceId,     type: Nullable(Int64),  parsed text: \"4\"\nERROR: garbage after Nullable(Int64): \"d8caa60-f8\"\n\n: (at row 1)\n. (CANNOT_PARSE_INPUT_ASSERTION_FAILED), Stack trace (when copying this message, always include the lines below):\n\n0. DB::Exception::Exception(DB::Exception::MessageMasked&&, int, bool) @ 0xe18d895 in /usr/bin/clickhouse\n1. ? @ 0xe1ec044 in /usr/bin/clickhouse\n2. DB::throwAtAssertionFailed(char const*, DB::ReadBuffer&) @ 0xe1ebf41 in /usr/bin/clickhouse\n3. DB::TabSeparatedFormatReader::skipFieldDelimiter() @ 0x14adec49 in /usr/bin/clickhouse\n4. DB::RowInputFormatWithNamesAndTypes::readRow(std::vector<COW<DB::IColumn>::mutable_ptr<DB::IColumn>, std::allocator<COW<DB::IColumn>::mutable_ptr<DB::IColumn>>>&, DB::RowReadExtension&) @ 0x149d531f in /usr/bin/clickhouse\n5. DB::IRowInputFormat::generate() @ 0x149b08ae in /usr/bin/clickhouse\n6. DB::ISource::tryGenerate() @ 0x14933695 in /usr/bin/clickhouse\n7. DB::ISource::work() @ 0x14933206 in /usr/bin/clickhouse\n8. DB::ParallelParsingInputFormat::parserThreadFunction(std::shared_ptr<DB::ThreadGroupStatus>, unsigned long) @ 0x14a5c341 in /usr/bin/clickhouse\n9. ThreadPoolImpl<ThreadFromGlobalPoolImpl<false>>::worker(std::__list_iterator<ThreadFromGlobalPoolImpl<false>, void*>) @ 0xe260ea5 in /usr/bin/clickhouse\n10. void std::__function::__policy_invoker<void ()>::__call_impl<std::__function::__default_alloc_func<ThreadFromGlobalPoolImpl<false>::ThreadFromGlobalPoolImpl<void ThreadPoolImpl<ThreadFromGlobalPoolImpl<false>>::scheduleImpl<void>(std::function<void ()>, long, std::optional<unsigned long>, bool)::'lambda0'()>(void&&)::'lambda'(), void ()>>(std::__function::__policy_storage const*) @ 0xe263a15 in /usr/bin/clickhouse\n11. ThreadPoolImpl<std::thread>::worker(std::__list_iterator<std::thread, void*>) @ 0xe25cc73 in /usr/bin/clickhouse\n12. ? @ 0xe2628e1 in /usr/bin/clickhouse\n13. start_thread @ 0x7ea5 in /usr/lib64/libpthread-2.17.so\n14. clone @ 0xfeb0d in /usr/lib64/libc-2.17.so\n (version 23.3.1.2823 (official build))","@@id":"b63049ee2121095314365cc38aa23472","@filehashkey":"778261a8412adda6f7e3f7297ea2d5d1","@collectiontime":"2023-12-18T11:07:21.680+08:00","@hostname":"master94","@path":"/data01/chenyc/logs/clickhouse-server/clickhouse-server.err.log.0","@rownumber":2883976,"@seq":2842107,"@ip":"192.168.101.94","@topic":"log_test"}]; 

為了屏蔽掉這個報(bào)錯導(dǎo)致任務(wù)異常PAUSED，所以將出錯容忍度設(shè)置為500。

測試結(jié)果如下：

任務(wù)情況：

mysql> show routine load\G;
*************************** 1. row ***************************
                  Id: 22719
                Name: doris_test
          CreateTime: 2023-12-18 16:08:49
           PauseTime: NULL
             EndTime: NULL
              DbName: default_cluster:demo
           TableName: log_test
        IsMultiTable: false
               State: RUNNING
      DataSourceType: KAFKA
      CurrentTaskNum: 3
       JobProperties: {"max_batch_rows":"300000","timezone":"Asia/Shanghai","send_batch_parallelism":"1","load_to_single_tablet":"false","current_concurrent_number":"3","delete":"*","partial_columns":"false","merge_type":"APPEND","exec_mem_limit":"2147483648","strict_mode":"false","jsonpaths":"","max_batch_interval":"20","max_batch_size":"209715200","fuzzy_parse":"false","partitions":"*","columnToColumnExpr":"@message,@@id,@filehashkey,@collectiontime,@hostname,@path,@rownumber,@seq,@ip,@topic","whereExpr":"*","desired_concurrent_number":"3","precedingFilter":"*","format":"json","max_error_number":"500","max_filter_ratio":"1.0","json_root":"","strip_outer_array":"false","num_as_string":"false"}
DataSourceProperties: {"topic":"log_test","currentKafkaPartitions":"0,1,2,3,4,5","brokerList":"192.168.101.94:29092,192.168.101.96:29092,192.168.101.98:29092"}
    CustomProperties: {"group.id":"doris_test_4fba1257-9291-44cc-b4ef-61dd64f58c5e"}
           Statistic: {"receivedBytes":129215158726,"runningTxns":[],"errorRows":57,"committedTaskNum":720,"loadedRows":201594590,"loadRowsRate":41912,"abortedTaskNum":0,"errorRowsAfterResumed":0,"totalRows":201594647,"unselectedRows":0,"receivedBytesRate":26864584,"taskExecuteTimeMs":4809870}
            Progress: {"0":"33599107","1":"33599106","2":"33599107","3":"33599107","4":"33599107","5":"33599107"}
                 Lag: {"0":0,"1":0,"2":0,"3":0,"4":0,"5":0}
ReasonOfStateChanged: 
        ErrorLogUrls: http://192.168.101.93:58040/api/_load_error_log?file=__shard_15/error_log_insert_stmt_6c68b52cc0f849f7-ac9a0ce97dabf7e9_6c68b52cc0f849f7_ac9a0ce97dabf7e9, http://192.168.101.96:58040/api/_load_error_log?file=__shard_16/error_log_insert_stmt_7e6449b5c1b6469f-901d55766835b101_7e6449b5c1b6469f_901d55766835b101, http://192.168.101.94:58040/api/_load_error_log?file=__shard_18/error_log_insert_stmt_44ca6b30a5394689-b34d2e33de930a0c_44ca6b30a5394689_b34d2e33de930a0c
            OtherMsg: 
                User: root
             Comment: 

這個地方的指標(biāo)數(shù)據(jù)是有BUG的：

 Statistic: {"receivedBytes":129215158726,"runningTxns":[],"errorRows":57,"committedTaskNum":720,"loadedRows":201594590,"loadRowsRate":41912,"abortedTaskNum":0,"errorRowsAfterResumed":0,"totalRows":201594647,"unselectedRows":0,"receivedBytesRate":26864584,"taskExecuteTimeMs":4809870}

我們可以看一下它的計(jì)算邏輯：

public Map<String, Object> summary() {
        Map<String, Object> summary = Maps.newHashMap();
        summary.put("totalRows", Long.valueOf(totalRows));
        summary.put("loadedRows", Long.valueOf(totalRows - this.errorRows - this.unselectedRows));
        summary.put("errorRows", Long.valueOf(this.errorRows));
        summary.put("errorRowsAfterResumed", Long.valueOf(this.errorRowsAfterResumed));
        summary.put("unselectedRows", Long.valueOf(this.unselectedRows));
        summary.put("receivedBytes", Long.valueOf(this.receivedBytes));
        summary.put("taskExecuteTimeMs", Long.valueOf(this.totalTaskExcutionTimeMs));
        summary.put("receivedBytesRate", Long.valueOf(this.receivedBytes * 1000 / this.totalTaskExcutionTimeMs));
        summary.put("loadRowsRate", Long.valueOf((this.totalRows - this.errorRows - this.unselectedRows) * 1000
                / this.totalTaskExcutionTimeMs));
        summary.put("committedTaskNum", Long.valueOf(this.committedTaskNum));
        summary.put("abortedTaskNum", Long.valueOf(this.abortedTaskNum));
        summary.put("runningTxns", runningTxnIds);
        return summary;
    }

receivedBytesRate是拿總條數(shù)/總耗時，loadRowsRate是拿總條數(shù)-出錯條數(shù)-未選中條數(shù) 再除以總時間，看起來是沒有問題的。但問題出在這個總時間taskExecuteTimeMs上：

private void updateNumOfData(long numOfTotalRows, long numOfErrorRows, long unselectedRows, long receivedBytes,
                                 long taskExecutionTime, boolean isReplay) throws UserException {
        this.jobStatistic.totalRows += numOfTotalRows;
        this.jobStatistic.errorRows += numOfErrorRows;
        this.jobStatistic.unselectedRows += unselectedRows;
        this.jobStatistic.receivedBytes += receivedBytes;
        this.jobStatistic.totalTaskExcutionTimeMs += taskExecutionTime;
        
        ...
}

此處計(jì)算總行數(shù)，去統(tǒng)計(jì)各個并發(fā)的總數(shù)是沒問題的，但是總耗時也這樣計(jì)算的話，實(shí)際上是多算了，也就是說，有幾個并發(fā)度，這個時間就多算了多少倍。因此，這個導(dǎo)入任務(wù)的真實(shí)時間應(yīng)該是 4809870/ 3 = 1603290，也就是27分鐘。

插入后表中數(shù)據(jù)：

mysql> select count(*) from log_test;
+-----------+
| count(*)  |
+-----------+
| 201594590 |
+-----------+
1 row in set (0.05 sec)

Doris接收到的數(shù)據(jù)總量為120G，遠(yuǎn)大于clickhouse壓縮前的數(shù)據(jù)量88G，猜測原因可能是寫放大導(dǎo)致。因?yàn)镽outine Load實(shí)際上是FE分配任務(wù)后在BE上執(zhí)行stream load，而stream load則是先將數(shù)據(jù)拉取到一個BE節(jié)點(diǎn)，然后廣播發(fā)送給其他節(jié)點(diǎn)。

寫入性能小結(jié)

從2億條數(shù)據(jù)的寫入性能來看，clickhouse寫入可以達(dá)到28w條每秒，Doris大約12w條/s， clickhouse的寫入性能是Doris的2倍以上。

但是從資源消耗來看，Doris的寫入由于是由Routine Load完成，占用的是BE節(jié)點(diǎn)的資源，而clickhouse使用第三方的clickhouse_sinker完成，完全可以和節(jié)點(diǎn)部署在不同機(jī)器上，從而避免對clickhouse集群資源的侵占。

Doris在寫入性能落后的情況下，CPU的消耗與clickhouse_sinker相當(dāng)，內(nèi)存稍微占用少一點(diǎn)，但是和clickhouse節(jié)點(diǎn)來比，就完全不是一個數(shù)量級了。clickhouse在數(shù)據(jù)寫入時，CPU和內(nèi)存的波動都比較小，處于正常水平。不會吃太多的資源，從而影響到查詢。更為重要的是，Doris寫入的資源占用，是每個節(jié)點(diǎn)都要占這么多，N個節(jié)點(diǎn)，這個資源消耗就是N倍，這和clickhouse_sinker之間的差距就進(jìn)一步拉大了。

clickhouse_sinker是擎創(chuàng)科技開源的一個將kafka的數(shù)據(jù)寫入clickhouse的開源項(xiàng)目。擁有著低資源消耗，高性能寫入，高容錯、穩(wěn)定的運(yùn)行能力。它通過寫本地表的方式，可以達(dá)到數(shù)據(jù)均衡寫入到各節(jié)點(diǎn)，自動按照shardingkey做hash路由，多個sinker進(jìn)程實(shí)例之間自動根據(jù)kafka的消費(fèi)lag來合理分配寫入任務(wù)等，非常適合作為clickhouse數(shù)據(jù)寫入的方案。

從壓縮性能上來看，由于clickhouse和Doris對壓縮前的數(shù)據(jù)統(tǒng)計(jì)口徑不一致，所以光看壓縮比意義不大。但kafka的數(shù)據(jù)是固定的，kafka里的數(shù)據(jù)是30G（kafka也有自己的壓縮算法，使用的是zstd），寫入到clickhouse后數(shù)據(jù)為10.17GB，這個大小沒有算入副本，如果算上副本，應(yīng)該乘以2，也就是20.34GB。

Doris的數(shù)據(jù)大小通過SHOW TABLE STATUS FROM demo LIKE '%log_test%';查詢得到，Data_length為67.71GB，除以三個副本，得到22.57GB。

不算副本，只看單一數(shù)據(jù)的大小，clickhouse的壓縮率達(dá)到了Doris的2.2倍，這個壓縮差距還是非常大的。

查詢性能

我們分上面六個預(yù)設(shè)的場景進(jìn)行查詢測試。

查詢SQL如下：

查詢結(jié)果：

備注：查詢結(jié)果取連續(xù)查詢10次的中位數(shù)。

初步分析：

從上述結(jié)果來看，clickhouse勝2負(fù)4。其中場景1是碾壓性優(yōu)勢，查詢性能是Doris的10倍多，這是因?yàn)镈oris本身不善于count類的查詢，而clickhouse依靠projection的預(yù)聚合查詢，達(dá)到了極致性能。

場景5之所以clickhouse能領(lǐng)先，有必要說明一下，按照原計(jì)劃Doris使用MATCH ALL語法去查詢，但是沒有查詢到結(jié)果（不明白為什么），改用LIKE查詢后性能比較差，達(dá)到了5秒左右，甚至比clickhouse更慢（這也是我沒有想到的）。

Doris在PK中勝4負(fù)2，比分大幅領(lǐng)先。除了場景2查詢耗時相差不大之外，其余場景3、4、6都是降維打擊，性能遙遙領(lǐng)先，達(dá)到了clickhouse的5倍左右。這自然是得益于Doris的全文檢索功能立了大功。

場景2之所以相差不大，還是因?yàn)镾QL中涉及到了count的計(jì)算，前面說過，Doris不擅長count類的查詢，因此性能比較拉胯，也就情有可原了。但即便如此，依然依然能達(dá)到clickhouse的1.3倍。

clickhouse 日志存儲優(yōu)化方案-構(gòu)建隱式列

構(gòu)建隱式列（或map列）是目前業(yè)界各大企業(yè)使用clickhouse存儲日志的通用落地方案。下面摘取了一些成熟的日志存儲的實(shí)踐方案，無不例外都用到了構(gòu)建隱式列或Map列的思想：

1. 使用 ClickHouse 構(gòu)建通用日志系統(tǒng)
2. Uber 如何使用 ClickHouse 建立快速可靠且與模式無關(guān)的日志分析平臺？
3. 還在用 ES 查日志嗎，快看看石墨文檔 Clickhouse 日志架構(gòu)玩法
4. Building an Observability Solution with ClickHouse - Part 1 - Logs
5. B站基于Clickhouse的下一代日志體系建設(shè)實(shí)踐

所謂隱式列（ Implicit columns）， 我們可以將message中常用的（有規(guī)律的）一些字段，通過正則表達(dá)式提取出來，作為一個隱式列，構(gòu)建一個大寬表，然后查詢的時候匹配該隱式列，從而達(dá)到避免走或少走全文檢索的效果。你clickhouse不是不擅長模糊查詢么，那么我就盡量不走模糊查詢，不就行了嗎？

比如本案例中，我們可以將日志中的query_id，thread_id，loglevel， timestamp等內(nèi)容提取出來。

下例為通過我們自研采集器提取字段的例子：

if $raw_event =~ /(^\d+\.\d+\.\d+\s+\d+:\d+:\d+\.\d+)\s+\[\s+(\d+)\s+\]\s+{(.*)}\s+<(\w+)>.*/  {
     $@timestamp=replace($1, ".", "-", 2);
     $@threadid=$2;
     $@queryid=$3;
     $@loglevel=$4;
}

采集到的數(shù)據(jù)樣例如下：

{
    "@message": "2023.12.07 03:41:18.775976 [ 154026 ] {} <Error> aimeter.metric_agg (ReplicatedMergeTreePartCheckThread): No replica has part covering 202312_11890_20601_1725 and a merge is impossible: we didn't find a smaller part with the same min block.",
    "@@id": "f7efeef0501a4f13f8561d2dfa18461d",
    "@filehashkey": "12d1faf9499acb0664d5dfe4af9d761c",
    "@collectiontime": "2023-12-18T17:56:26.586+08:00",
    "@hostname": "master94",
    "@path": "/data01/chenyc/logs/clickhouse-server/clickhouse-server.err.log.1",
    "@rownumber": 3,
    "@seq": 3,
    "@timestamp": "2023-12-07 03:41:18.775976",
    "@threadid": "154026",
    "@queryid": "",
    "@loglevel": "Error",
    "@ip": "192.168.101.94",
    "@topic": "log_test2"
}

建表語句如下：

CREATE TABLE default.log_test2 on cluster abc (
  `@@id` String CODEC(ZSTD(1)),
  `@message` String CODEC(ZSTD(1)),
  `@filehashkey` String CODEC(ZSTD(1)),
  `@collectiontime` DateTime64(3),
  `@hostname` String,
  `@path` String CODEC(ZSTD(1)),
  `@rownumber` Int64,
  `@seq` Int64,
  `@timestamp` DateTime64(3) CODEC(DoubleDelta, LZ4),
  `@threadid` Int32,
  `@queryid` String CODEC(ZSTD(1)),
  `@loglevel` LowCardinality(String),
  `@ip` String CODEC(ZSTD(1)),
  `@topic` LowCardinality(String),
  INDEX level_idx `@loglevel` TYPE tokenbf_v1(4096, 1, 0) GRANULARITY 1,
  INDEX ip_idx `@ip` TYPE tokenbf_v1(4096, 1, 0) GRANULARITY 1,
  INDEX query_idx `@queryid` TYPE ngrambf_v1(10, 30720, 1, 0) GRANULARITY 1,
  INDEX message_idx `@message` TYPE ngrambf_v1(5, 65535, 1, 0) GRANULARITY 1,
  PROJECTION p_cnt (
    SELECT `@ip`, `@path`, count() GROUP BY `@ip`, `@path`
  )
) ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMMDD(`@timestamp`)
ORDER BY
  (`@timestamp`, `@ip`, `@path`, `@loglevel`)
  
--- 分布式表
create table dist_log_test2 on cluster abc as log_test2 engine = Distributed('abc', 'default', 'log_test2')

由于多了4個字段，kafka中的數(shù)據(jù)膨脹了4G。

通過clickhouse_sinker將數(shù)據(jù)導(dǎo)入到clickhouse集群。

寫入clickhouse后如下所示：

針對上面6種場景，改寫SQL如下：

為了方便對比，我們將上兩次的查詢結(jié)果也貼到一塊。

先刨除掉場景1，場景6不看，因?yàn)椴樵僑QL與不加隱式列是一樣的，所以性能也差不多。

關(guān)鍵看場景2，場景3，場景4。場景2比未加隱式列之前性能提升了20倍，比Doris提升了15倍，提升非常大。而場景3和場景4也在不加隱式列的基礎(chǔ)上提升了3倍左右的性能，雖然還比不上Doris，但差距已經(jīng)追小了許多（Doris僅領(lǐng)先1.5倍）。

場景5由于仍然有根據(jù)@message字段做模糊搜索，所以性能提升不大。

總而言之，在絕大多數(shù)場景，通過隱式列的方式改寫查詢語句，可以將原有的查詢性能提升3倍左右。

通過構(gòu)建隱式列的方式存儲日志，可有效解決查詢性能的問題。但在交互上就顯得不那么友好。因?yàn)閷τ谑褂谜邅碚f，是不知道有這些列的存在的，或者說如果使用者沒有很強(qiáng)的業(yè)務(wù)感知能力，都是隨性搜索短語的話，同樣會導(dǎo)致查詢的內(nèi)容只能通過模糊匹配，那么就起不到任何加速作用。

因此，要想用好隱式列，首先需要在交互上引導(dǎo)用戶去使用隱式列進(jìn)行條件搜索，而不是隨意選擇關(guān)鍵字；

其次是要做好日志規(guī)范，否則，不僅提取有效關(guān)鍵字比較困難，而且不同的業(yè)務(wù)日志有不同的提取方法，有不同的關(guān)鍵字，導(dǎo)致提取出來的維度關(guān)鍵字五花八門，這對搜索來說也帶來了一定的困難。

大查詢對寫入的影響

我們知道，在實(shí)際生產(chǎn)環(huán)境，除非有一套相對成熟的存算分離方案，否則寫入對查詢的互相影響是不可避免的。由于數(shù)據(jù)量有限，為了盡可能模擬大查詢對寫入的影響，我們采用場景2的SQL語句，通過5個SQL并發(fā)查詢，觀察寫入數(shù)據(jù)的速度變化。

為了盡可能模擬真實(shí)并發(fā)情況，在這里使用golang分別實(shí)現(xiàn)了5并發(fā)查詢數(shù)據(jù)庫的功能。

clickhouse：

package main
?
import (
    "fmt"
    "sync"
?
    "github.com/ClickHouse/clickhouse-go/v2"
)
?
func main() {
    conn := clickhouse.OpenDB(&clickhouse.Options{
        Addr: []string{"192.168.101.93:19000","192.168.101.94:19000","192.168.101.96:19000","192.168.101.97:19000"},
        Auth: clickhouse.Auth{
            Database: "default",
            Username: "default",
            Password: "123456",
        },
    })
?
    err := conn.Ping()
    if err != nil {
        panic(err)
    }
?
    query := "SELECT `@ip`, count() FROM dist_log_test WHERE `@message` LIKE '%Error%' GROUP BY `@ip`"
    for {
        var wg sync.WaitGroup
        var lastErr error
        for i := 0; i < 5; i++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                rows, err := conn.Query(query)
                if err != nil {
                    lastErr = err
                    return
                }
                defer rows.Close()
                for rows.Next() {
                    var (
                        ip  string
                        cnt uint64
                    )
                    if err := rows.Scan(&ip, &cnt); err != nil {
                        lastErr = err
                        return
                    }
                    fmt.Printf("ip: %s, count: %d\n", ip, cnt)
                }
            }()
        }
        wg.Wait()
        if lastErr != nil {
            panic(lastErr)
        }
    }
}

查詢Doris：

package main
?
import (
    "database/sql"
    "fmt"
    "sync"
?
    _ "github.com/go-sql-driver/mysql"
)
?
func main() {
    conn, err := sql.Open("mysql", "root:@(192.168.101.94:59030)/demo")
    if err != nil {
        panic(err)
    }
?
    if err = conn.Ping(); err != nil {
        panic(err)
    }
?
    query := "SELECT `@ip`, count() FROM log_test WHERE `@message` MATCH_ANY 'Error' GROUP BY `@ip`"
    for {
        var wg sync.WaitGroup
        var lastErr error
        for i := 0; i < 5; i++ {
            wg.Add(1)
            go func() {
                defer wg.Done()
                rows, err := conn.Query(query)
                if err != nil {
                    lastErr = err
                    return
                }
                defer rows.Close()
                for rows.Next() {
                    var (
                        ip  string
                        cnt uint64
                    )
                    if err := rows.Scan(&ip, &cnt); err != nil {
                        lastErr = err
                        return
                    }
                    fmt.Printf("ip: %s, count: %d\n", ip, cnt)
                }
            }()
        }
        wg.Wait()
        if lastErr != nil {
            panic(lastErr)
        }
    }
}

clickhouse

clickhouse5個并發(fā)同時查詢，CPU飆到了接近40 core，幾乎占滿了物理機(jī)的80%的資源。

此時我們啟動clickhouse_sinker任務(wù)（sinker進(jìn)程部署在clickhouse集群以外的節(jié)點(diǎn)），觀察寫入性能如下：

上面這個測試案例比較特殊，因?yàn)樗械牟樵冋埱蠖即虻搅送粋€clickhouse節(jié)點(diǎn)，導(dǎo)致這個節(jié)點(diǎn)的CPU占用特別高，其他節(jié)點(diǎn)的CPU比較正常，整體寫入性能有所下降，約為原來的70%。但總體性能還是不錯的，單sinker進(jìn)程可以達(dá)到20w行每秒。

如果我們在clickhouse的查詢端加一層proxy，使得查詢請求比較均衡地分不到各個clickhouse節(jié)點(diǎn)，相信查詢性能還能進(jìn)一步提高。

Doris

5個并發(fā)查詢Doris，可以看到Doris的FE幾乎無資源損耗，但是BE的CPU吃滿。而且與clickhouse不同的是，clickhouse是僅請求打到這個節(jié)點(diǎn)上，這個節(jié)點(diǎn)的CPU才會占得比較高，但是Doris的各個節(jié)點(diǎn)的CPU都占得比較高。

我們同樣啟動一個Routine Load向Doris寫入數(shù)據(jù)，性能如下：

當(dāng)我們在并發(fā)執(zhí)行很多耗時的大查詢時，由于CPU占用比較滿，導(dǎo)致寫入性能下降了50%左右，這使得原本就不富裕的生活更加雪上加霜。

但是我們注意到一個有意思的現(xiàn)象，原本場景2的查詢5秒能返回結(jié)果，但是在高速寫入時，查詢速度降到了15秒（事實(shí)上所有的查詢都慢了2-3倍左右，clickhouse更夸張，查詢會慢5倍以上）。所以Doris在同時有讀寫請求的時候，是優(yōu)先保證寫請求的資源的。

大查詢寫入優(yōu)化方案-用戶資源限制

clickhouse和Doris都可以通過設(shè)置用戶權(quán)限來限制某個查詢用戶所能使用的資源。從上面的測試結(jié)果來看，我們發(fā)現(xiàn)即使clickhouse寫入時大查詢占據(jù)了80%資源，clickhouse的寫入速度（190k/s）還是高于Doris無干擾寫入的速度（126k/s）， 因此，我們主要來看clickhouse在專門設(shè)置了一個查詢用戶后的插入性能情況。

我們新增一個query的用戶：

通過profile限制其查詢最大線程數(shù)為8：

由于可用線程數(shù)減少，為了避免查詢超時，將SQL超時時間修改到1個小時。

同時通過quota設(shè)置當(dāng)1分鐘內(nèi)連續(xù)5次報(bào)錯，就禁止該用戶查詢：

我們通過這個query用戶，起5個并發(fā)，輪詢查詢副本節(jié)點(diǎn)ck94, ck97（clickhouse寫ck93, ck96）。clickhouse_sinker在集群以外的節(jié)點(diǎn)上啟動。

測試結(jié)果如下：

對比無干擾時寫入，寫入性能僅下降5-10%左右，這主要是因?yàn)閷懭氡旧硐牡墓?jié)點(diǎn)資源就比較少，當(dāng)查詢的資源被限制，clickhouse的節(jié)點(diǎn)就有足夠多的資源去保障寫入，并且我們通過配置簡單的讀寫分離的方式，讓查詢請求盡量分配到不同的副本節(jié)點(diǎn)，可以進(jìn)一步減小查詢對寫入的影響。

不過需要注意的是，用戶的最大線程數(shù)限制的是單個查詢所使用的最大線程資源，如果多個查詢語句同時請求到同一個節(jié)點(diǎn)，仍然能將該節(jié)點(diǎn)的CPU負(fù)載占滿。（測試過程中，嘗試5個并發(fā)全部請求到一個節(jié)點(diǎn)，該節(jié)點(diǎn)CPU能達(dá)到2500%）。

總結(jié)

本文重點(diǎn)比較了clickhouse和Doris在日志存儲場景下的寫入能力和查詢能力。

寫入性能上，clickhouse完勝。在寫入性能領(lǐng)先Doris 2倍的情況下，可以做到使用更少的系統(tǒng)資源，且壓縮率也達(dá)到了Doris的2倍以上。

至于查詢性能，由于Doris支持倒排索引，在模糊查詢場景，Doris對比clickhouse有5倍左右的提升，雖然clickhouse可以通過構(gòu)建隱式列的方式提升查詢效率，但Doris仍然能夠做到1.5倍左右的性能領(lǐng)先。

而在需要聚合計(jì)算count的查詢場景，Doris明顯不如clickhouse高效。

在讀寫同時進(jìn)行的場景，在大查詢比較多時，clickhouse和Doris的寫入性能都有所下降，clickhouse寫入性能下降到70%，Doris則直接腰斬。而且都對查詢影響比較大。

通過配置專門的查詢用戶限制查詢查詢資源，可有效緩解大查詢對寫入帶來的性能影響。

?本專欄知識點(diǎn)是通過<零聲教育>的系統(tǒng)學(xué)習(xí)，進(jìn)行梳理總結(jié)寫下文章，對C/C++課程感興趣的讀者，可以點(diǎn)擊鏈接，查看詳細(xì)的服務(wù)：C/C++Linux服務(wù)器開發(fā)/高級架構(gòu)師文章來源地址http://www.zghlxwxcb.cn/news/detail-813402.html

到了這里，關(guān)于ClickHouse與Doris數(shù)據(jù)庫比較的文章就介紹完了。如果您還想了解更多內(nèi)容，請?jiān)谟疑辖撬阉鱐OY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！