洪波涌起创建于 2025年01月12日修改于 2025年01月12日473次阅读

如何在面对巨大数据量时，有效地使用 ClickHouse 来处理和存储数据

本文经授权转自公众号 CSDN（ID：CSDNnews）

作者 | Dale McDiarmid，译者 | 明明如月，责编 | 夏萌

该文章讨论了如何在面对巨大数据量时，有效地使用 ClickHouse 来处理和存储数据。它详细介绍了 ClickHouse 的几个关键特性，包括对大规模数据加载的优化、通过使用字典来加速查询、以及如何利用 ClickHouse 处理超过一亿的唯一用户请求。文章通过具体实例，如 Admixer 的使用案例，展示了 ClickHouse 在实际应用中的性能和灵活性。

原文链接

近期，Decodable 的 Gunnar Morling 在其 LinkedIn 个人主页上发布了一个广受关注的挑战：编写可以从一个包含十亿行文本的文件中提取气温测量数据，并计算每个气象站的最低、平均和最高气温的 Java 程序。尽管我们并非 Java 领域的专家，但作为一家热衷于大数据和性能测试的公司，我们认为应该用 ClickHouse 这一官方平台来迎接这一挑战！

尽管原始挑战依旧基于 Java，但 Gunnar 在 GitHub 的讨论版块中新增了一个名为 "展示与讲述" 的专区，鼓励更多技术领域的贡献。我们也要感谢社区成员的积极参与，他们同样接受了这一挑战。

1. 遵守挑战规则

在应对这一挑战的过程中，我们努力遵循了原始挑战的宗旨和规则。因此，在我们的最终提交中，我们包含了所有处理和数据加载的时间。如果我们只报告了数据加载到表格后的查询响应时间，而刻意忽略数据插入的时间，就算作弊了。

Gunnar 在 Hetzner AX161 服务器上进行测试，限制为 8 核心。虽然我很想为了参与这个网络挑战而购买一台专用的高性能服务器，但我们最终认为这有些过头。为了确保测试结果具有可比性，我们采用了 Hetzner 的虚拟服务器实例（配备专用 CPU），型号为 CCX33，具备 8 核心和 32GB RAM。尽管它们是虚拟实例，但它采用的 AMD EPYC-Milan 处理器基于 Zen3 架构，相比 Hetzner AX161 使用的 AMD EPYC-Rome 7502P 处理器更加先进。

2. 生成（或下载）数据

用户可以根据官方指南生成一个含有 10 亿条数据记录的数据集。这需要使用 Java 21，并执行相应的命令。

在写这篇博客时，我发现了 sdkman，这是一个可以简化 Java 安装过程的工具，特别适合还未安装 Java 的用户。

然而，生成一个 13GB 大小的 measurements.txt 文件的过程比较缓慢：

bash 复制代码

# 克隆并构建数据生成工具，输出结果已省略。
git clone git@github.com:gunnarmorling/1brc.git
./mvnw clean verify
./create_measurements.sh 1000000000

结果：生成了含 1,000,000,000 条测量数据的文件，耗时 395955 毫秒。

相较之下，我们使用 ClickHouse Local 生成同样的文件速度更快，这个结果颇为吸引人。通过查看源代码，我们了解到站点列表及其平均温度已经编入代码中，并通过对一个均值和方差均为 10 的高斯分布进行采样来生成随机数据点。将原始站点数据提取为 CSV 格式，并上传到 s3，使我们能够使用 INSERT INTO FUNCTION FILE 命令来重现此逻辑。值得注意的是，在使用随机函数对结果进行采样之前，我们通过 s3 函数读取了 CTE。

sql 复制代码

INSERT INTO FUNCTION file('measurements.csv', CustomSeparated)
WITH (
    SELECT groupArray((station, avg)) FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/1brc/stations.csv')
) AS averages
SELECT
    averages[floor(randUniform(1, length(averages)))::Int64].1 as city,
    round(averages[floor(randUniform(1, length(averages)))::Int64].2 + (10 * SQRT(-2 * LOG(randCanonical(1))) * COS(2 * PI() * randCanonical(2))), 2) as temperature
FROM numbers(1_000_000_000)
SETTINGS format_custom_field_delimiter=';', format_custom_escaping_rule='Raw'

处理结果：0 行。耗时 57.856 秒。处理了 10 亿行，8.00 GB（速度为每秒 17.28 百万行，138.27 MB/s）。峰值内存使用：36.73 MiB。

以 6.8 倍的速度完成任务，非常值得分享！

熟悉 ClickHouse 的用户可能会考虑使用 randNormal 函数。但遗憾的是，目前这个函数只支持固定的均值和方差。因此，我们采用了 randCanonical 函数，并利用它通过 Box-Muller 变换对高斯分布进行采样。

或者，用户也可以选择直接从此链接下载我们生成的 gzip 压缩文件版本。:)

3. 专为 ClickHouse 本地版本而设计

尽管许多用户习惯于将 ClickHouse 部署在服务器上，作为实时数据仓库使用，但实际上，ClickHouse 还可以作为本地二进制文件运行，这种方式称为 “ClickHouse Local”，适用于临时数据分析和文件查询。自从我们一年多前在博客中介绍了这种用法以来，这已经成为 ClickHouse 的一个日益受欢迎的应用方式。

ClickHouse Local 提供了控制台模式（通过运行 clickhouse local 访问），在该模式下用户可以创建表并进行交互式查询，同时还提供了命令行界面，便于与脚本和其他外部工具集成。我们借助这一功能对 measurements.txt 文件进行了数据采样。通过设置 format_csv_delimiter=';'，可以自定义 CSV 文件的分隔符。

bash 复制代码

clickhouse local --query "SELECT city, temperature FROM file('measurements.txt', CSV, 'city String, temperature DECIMAL(8,1)') LIMIT 5 SETTINGS format_csv_delimiter=';'"

输出结果：

复制代码

Mexicali    44.8
Hat Yai    29.4
Villahermosa    27.1
Fresno    31.7
Ouahigouya    29.3

要计算每个城市的最低、最高和平均温度，我们只需要执行一个简单的 GROUP BY 查询。为了确保包含处理时间信息，我们使用了 -t 参数。挑战在于按特定格式输出结果：

复制代码

{Abha=-23.0/18.0/59.2, Abidjan=-16.2/26.0/67.3, Abéché=-10.0/29.4/69.0, Accra=-10.1/26.4/66.4, Addis Ababa=-23.7/16.0/67.0, Adelaide=-27.8/17.3/58.5, ...}

为实现此目的，可以使用 CustomSeparated 输出格式结合 format 函数。这样我们就可以避免使用像 groupArray 这样的函数，后者会将多行数据合并为单行。下面是我们使用 ClickHouse Local 控制台模式的例子。

sql 复制代码

SELECT format('{}={}/{}/{}', city, min(temperature), round(avg(temperature), 2), max(temperature))
FROM file('measurements.txt', CSV, 'city String, temperature DECIMAL(8,1)')
GROUP BY city
ORDER BY city ASC
FORMAT CustomSeparated
SETTINGS
    format_custom_result_before_delimiter = '{',
    format_custom_result_after_delimiter = '}',
    format_custom_row_between_delimiter = ', ',
    format_custom_row_after_delimiter = '',
    format_csv_delimiter = ';'

输出结果：

复制代码

{Abha=-34.6/18/70.3, Abidjan=-22.8/25.99/73.5, Abéché=-25.3/29.4/80.1, Accra=-25.6/26.4/76.8, Addis Ababa=-38.3/16/67, Adelaide=-33.4/17.31/65.5, …}

共 413 行。耗时：27.671 秒。处理了 10 亿行，13.79 GB（速度为每秒 36.14 百万行，498.46 MB/s）。峰值内存使用：47.46 MiB。

27.6 秒的处理时间为我们的基准。相较之下，同一硬件上的 Java 基准测试几乎需要 3 分钟才能完成。

bash 复制代码

./calculate_average_baseline.sh

实际耗时 2m59.364s 用户耗时 2m57.511s 系统耗时 0m3.372s

4. 提升性能

我们发现，由于 CSV 文件没有进行值转义，实际上不必使用 CSV 读取器。一种更为简单高效的做法是，直接把每一行作为字符串进行读取，接着使用分号作为分隔符来提取我们需要的子字符串。

sql 复制代码

SELECT format('{}={}/{}/{}', city, min(temperature), round(avg(temperature), 2), max(temperature))
FROM (
    SELECT
        substringIndex(line, ';', 1) AS city,
        substringIndex(line, ';', -1)::Decimal(8, 1) AS temperature
    FROM file('measurements.txt', LineAsString)
)
GROUP BY city
ORDER BY city ASC
FORMAT CustomSeparated
SETTINGS
    format_custom_result_before_delimiter = '{',
    format_custom_result_after_delimiter = '}',
    format_custom_row_between_delimiter = ', ',
    format_custom_row_after_delimiter = '',
    format_csv_delimiter = ';'

共 413 行。耗时：19.907 秒。处理了 10 亿行，13.79 GB（速度为每秒 50.23 百万行，692.86 MB/s）。峰值内存使用量：132.20 MiB。

采用这种方法后，执行时间缩减至不足 20 秒！

5. 测试替代方法

我们使用的 ClickHouse 本地方法进行了对文件的全面线性扫描。一种可能的替代方案是先将文件加载到表中，然后再对表执行查询。然而，这种方法并未显著提升性能，因为实际上相当于对数据进行了第二轮扫描。因此，整体的加载和查询时间超过了 19 秒。

sql 复制代码

CREATE TABLE weather (
    `city` String,
    `temperature` Decimal(8, 1)
) ENGINE = Memory;

INSERT INTO weather
SELECT
    city,
    temperature
FROM (
    SELECT
        splitByChar(';', line) AS vals,
        vals[1] AS city,
        CAST(vals[2], 'Decimal(8, 1)') AS temperature
    FROM file('measurements.txt', LineAsString)
);

共 0 行。耗时：21.219 秒。处理了 10 亿行，13.79 GB（每秒 47.13 百万行，650.03 MB/s）。峰值内存使用量：26.16 GiB。

sql 复制代码

SELECT
    city,
    min(temperature),
    avg(temperature),
    max(temperature)
FROM weather
GROUP BY city
ORDER BY city ASC
SETTINGS max_threads = 8;

共 413 行。耗时：2.997 秒。处理了 970.54 百万行，20.34 GB（每秒 323.82 百万行，6.79 GB/s）。峰值内存使用量：484.27 KiB。

注意，我们在此使用内存表而非传统的 MergeTree。考虑到数据集适合内存，且查询不包含过滤器（因此不会从 MergeTree 的稀疏索引中获益），我们可以使用这种引擎类型来避免 I/O。

这种方法的一个明显优点是，一旦数据加载到表中，用户就可以对数据执行任意查询。

最后，如果我们的目标查询计算最小值、最大值和平均值的性能不足，我们可以把这部分工作转移到数据插入过程中，并通过使用物化视图来实现。在这种情况下，一个物化视图 weather_mv 在数据插入时计算我们的统计数据。更具体地说，我们之前的聚合查询在数据插入时对数据块执行，其结果（实际上是聚合状态）通过 AggregatingMergeTree 表引擎存储在目标表 weather_results 中。查询该表时将利用预先计算的结果，从而显著加快执行时间。

对于我们的数据处理，我们可以采用 Null 引擎作为 weather 表的存储引擎。这种方式会导致数据行被丢弃，从而节省内存。

sql 复制代码

CREATE TABLE weather (
    `city` String,
    `temperature` Decimal(8, 1)
) ENGINE = Null;

CREATE TABLE weather_results (
    city String,
    max AggregateFunction(max, Decimal(8, 1)),
    min AggregateFunction(min, Decimal(8, 1)),
    avg AggregateFunction(avg, Decimal(8, 1))
) ENGINE = AggregatingMergeTree ORDER BY tuple();

CREATE MATERIALIZED VIEW weather_mv TO weather_results AS
SELECT
    city,
    maxState(temperature) as max,
    minState(temperature) as min,
    avgState(temperature) as avg
FROM weather
GROUP BY city;

INSERT INTO weather
SELECT
    city,
    temperature
FROM (
    SELECT
        splitByChar(';', line) AS vals,
        vals[1] AS city,
        CAST(vals[2], 'Decimal(8, 1)') AS temperature
    FROM file('measurements.txt', LineAsString)
);

共 0 行。耗时：26.569 秒。处理了 20 亿行，34.75 GB（每秒 75.27 百万行，1.31 GB/s）。

在 weather_results 表上的后续查询需要使用 merge- 函数来合并我们的聚合状态。

sql 复制代码

SELECT format('{}={}/{}/{}', city, minMerge(min), round(avgMerge(avg), 2), maxMerge(max))
FROM weather_results
GROUP BY city
ORDER BY city ASC
FORMAT CustomSeparated
SETTINGS
    format_custom_result_before_delimiter = '{',
    format_custom_result_after_delimiter = '}',
    format_custom_row_between_delimiter = ', ',
    format_custom_row_after_delimiter = '',
    format_csv_delimiter = ';';

共 413 行。耗时：0.014 秒。

这个方法给我们带来了非常短的名义执行时间，这在其他实验中也有报道。然而，当考虑到我们的 26 秒数据加载时间时，我们的整体性能仍然无法超越简单的 ClickHouse 本地查询。

6. 结论

我们成功完成了十亿行数据的挑战。我们展示了 ClickHouse Local 在符合挑战规则的硬件上解决问题大约需要 19 秒。尽管与专业解决方案相比可能略显不足，但这个方案非常简单，仅需几行 SQL 代码即可完成。在此，我们要感谢 Gunnar 对这个挑战所投入的时间和精力。

你是否有类似经验或其他高效处理大数据的方法？请在评论区分享您的见解。

本文转自公众号“CSDN”，ID：CSDNnews

预览时标签不可点

微信扫一扫
关注该公众号

× 分析：，，，，，，，，，，，，。视频小程序赞，轻点两下取消赞在看，轻点两下取消在看

分享留言收藏

end