安装Clickhous进行基准测试

测试环境

使用Docker容器运行Clickhouse, 预计会有10%的性能损耗
CPU: Intel® Core™ i7-6700K CPU @ 4.00GHz
内存: 4G

测试结果

硬盘类型	写入速度	写1千行	写1万行	写10万行	读1千行	读1万行	读10万行	读40万行
固态硬盘	900MB/s	5ms	25ms	230ms	5ms	13ms	90ms	450ms
机械硬盘	300MB/s	5ms	23ms	205ms	5ms	11ms	90ms	400ms

写入速度使用以下命令测得

dd if=/dev/zero  of=./test.img bs=100M count=30 oflag=dsync

什么是Clickhouse?

https://clickhouse.com/docs/zh

ClickHouse是一个用于联机分析（OLAP）的列式数据库管理系统（DBMS）。它以列式存储方式存储数据，适用于读取大量行但只提取部分列的查询场景。ClickHouse通过减少I/O消耗、压缩数据、提高系统缓存利用率和执行向量化操作等方式提供高效的数据处理能力。它在处理分析查询方面比传统的行式数据库系统更加适用，并且具有高吞吐量和低延迟的特点。

下面是ClickHouse和MySQL之间的一些主要差异的表格列举：

特性	ClickHouse	MySQL
存储方式	列式存储	行式存储
查询性能	高吞吐量、低延迟	低吞吐量、较高延迟
数据压缩	支持高效的数据压缩	有限的数据压缩支持
数据更新	适用于批量插入和追加，不支持实时数据更新	支持实时数据更新和事务处理
数据一致性	低一致性要求，不支持数据修改	高一致性要求，支持数据修改
数据类型	支持广泛的数据类型，包括日期、时间、IP地址等	支持常见的数据类型
查询语言	支持SQL查询语言	支持SQL查询语言
高级功能	支持复杂的分布式查询和聚合操作	提供较为基本的查询和聚合功能
扩展性	易于水平扩展，可处理大规模数据	相对较易于垂直扩展，处理中小规模数据
适用场景	适用于大规模数据分析和OLAP场景	适用于事务处理和常规的OLTP场景

请注意，上述表格列举的差异是一般性的概括，并不涵盖所有方面。具体的差异可能因版本、配置和使用情况而有所不同。

运行Docker容器

docker run -d --restart always -p 8123:8123 -p 9000:9000 --name clickhouse --ulimit nofile=262144:262144 \
    -e CLICKHOUSE_USER=root -e CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1 -e CLICKHOUSE_PASSWORD=password \
    -v /path/data:/var/lib/clickhouse/ \
    -v /path/logs:/var/log/clickhouse-server/ \
 clickhouse/clickhouse-server

数据库管理工具

使用DBeaver连接数据库

DBeaver

测试所需代码

创建数据库和表

CREATE DATABASE IF NOT EXISTS my_database;

CREATE TABLE my_database.`test_table` (
  `hash` String,
  `from` String,
  `to` String,
  `value` Float,
  `token` String
) ENGINE=MergeTree() ORDER BY hash;

写入测试 Python

# pip install clickhouse-driver
from clickhouse_driver import connect
import time
import random

conn = connect(
    host="host",
    port=9000,
    user="root",
    password="password",
    database="my_database",
)

cursor = conn.cursor()

cursor.execute("show tables;")
print(cursor.fetchall())

s = "1234567890abcdef"
addresses = []
for _ in range(100000):
    addresses.append("0x" + "".join(random.choices(s, k=40)))
aln = len(addresses) - 1


def range_data():
    d = []
    od = "0x" + "".join(random.choices(s, k=64))
    d.append(od)
    od = addresses[random.randint(0, aln)]
    d.append(od)
    od = addresses[random.randint(0, aln)]
    d.append(od)
    od = random.randint(1, 100) + random.random()
    d.append(od)
    od = addresses[random.randint(0, aln)]
    d.append(od)
    return d


def write_data(data):
    cursor.executemany("INSERT INTO test_table VALUES", data)
    print(cursor.rowcount)


def read_data(line):
    cursor.execute("SELECT * FROM test_table LIMIT {}".format(line))
    print(cursor.rowcount)


def write_test(cnt=10000):
    for i in range(1, 500):
        dts = [range_data() for _ in range(cnt)]
        st = time.time_ns()
        write_data(dts)
        et = time.time_ns()
        print(f"写入总行数: {i*cnt}, 用时: {(et - st)/10**6:.2f}", "ms")


def read_test(cnt=10000):
    for i in range(1, 500):
        st = time.time_ns()
        read_data(cnt)
        et = time.time_ns()
        print(f"读取行数: {i*cnt}, 用时: {(et - st)/10**6:.2f}", "ms")


if __name__ == "__main__":
    # write_test(100000)
    read_test(500000)