上个月，我们正式回应了 Gunnar Morling 从 Decodable 发起的非常成功的 10 亿行挑战。这项挑战要求用户编写一个 Java 程序，从包含 10 亿次测量的文本文件中计算每个城市的最低、平均和最高温度。我们对这项挑战的回应显然使用了 ClickHouse，尽管受到并非针对该问题而专门设计的解决方案的限制，但仍发布了非常可观的时间，在使用规则规定的确切硬件配置的情况下约为 19 秒。
任何有经验的 ClickHouse 用户都知道，10 亿行对于 ClickHouse 来说非常少，用户经常对数万亿的数据集执行聚合。因此，当 Dask 最近将挑战扩展到 1 万亿行 时，我们的好奇心被激起了。
在这篇博文中，我们提交了我们尝试查询 1 万亿行数据集的尝试 - 在 3 分钟内完成查询，费用仅为 0.56 美元！虽然我们最初的解决方案使用了中等硬件，但这个更大的挑战在笔记本电脑或相当于小型工作站的机器上执行起来有点棘手。
相反，我们求助于使用 AWS 中的竞价型实例，同时旨在找到成本和性能之间的折衷方案。由于 AWS 提供了很大程度上线性的定价模型，这是一项简单的任务：我们只需识别 AWS 中最具价值的实例，以最大限度地提高查询性能。使用 Pulumi 脚本，我们可以启动 ClickHouse 集群，运行查询，然后关闭资源 - 所有这些只需 0.56 美元！
对于那些好奇的人，如果将这些数据加载到 MergeTree 中，则可以在 16 秒内回答查询！但这会产生加载时间，我们认为这是作弊，但是 :)
数据集
Dask 提供的数据结构与最初的挑战一致，包含城市和温度列。正如他们在最初的博客中讨论的那样，如此规模的数据需要比 10 亿行挑战中使用的 CSV 格式提供更好的压缩。因此，他们以 Parquet 格式提供以下请求方付费存储桶中的数据，即下载该数据的用户需要提供 AWS 的访问密钥并承担传输费用。
s3://coiled-datasets-rp/1trc
但是，只要我们确保我们的 ClickHouse 集群部署在与此存储桶相同的区域（us-east-1），我们就可以避免数据传输费用，同时优化网络带宽和延迟。
完整数据集包含 2.4TiB 的 Parquet 文件，以 10 万个文件提供，每个文件包含 1000 万行，大小约为 23-24MiB。
就地查询
经常访问的数据集可以加载到像 ClickHouse 这样的分析数据库中以进行快速查询。但对于不经常使用的数据集，有时最好将它们留在像 S3 这样的“湖仓”中，并能够在其中对它们运行即席分析查询。在 AWS 生态系统中，用户可能熟悉 Amazon Athena 等技术，该技术提供了使用标准 SQL 直接在 Amazon S3 中分析数据的能力。
能够查询湖仓并作为实时分析数据库的技术可以被认为是实时数据仓库。
ClickHouse 同时提供这两种功能。就我们的挑战而言，我们只需要执行一次此查询。正如我们在 10 亿行挑战结果 中指出的那样，虽然一旦数据加载到 ClickHouse 中，最终查询会更快，但实际加载时间将抵消任何好处。因此，我们利用 ClickHouse 的 s3 函数 来“就地”查询数据。
SELECT
count() as num_rows,
uniqExact(_file) as num_files
FROM s3('https://coiled-datasets-rp.s3.us-east-1.amazonaws.com/1trc/measurements-*.parquet', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', headers('x-amz-request-payer' = 'requester'))
┌──────num_rows─┬─num_files─┐
│ 1000000000000 │ 100000 │
└───────────────┴───────────┘
作为一个简单的示例，我们可以通过一个简单的查询确认总行数和文件数。
要访问请求方付费存储桶，必须在任何请求中传递标头
'x-amz-request-payer' = 'requester'。这可以通过将参数headers('x-amz-request-payer' = 'requester')传递给 s3 函数来实现。
一个简单的查询
我们之前针对 10 亿行挑战的查询包含一些优化，以最大限度地减少字符串格式化并确保数据以正确的格式返回。感谢 Parquet 格式，我们不需要进行任何字符串解析。因此，我们的查询是一个简单的 GROUP BY。
最大限度地降低 AWS 成本
为了将查询成本降至绝对最低，我们可以利用 AWS 的竞价型实例。这些实例类型允许您使用备用的 EC2 计算资源，您可以通过 AWS API 对其进行竞价（设置了最低竞价价格）。虽然这些实例可能会被 AWS 中断和回收（例如，如果竞价价格超过出价），但它们的价格最多可享受 90% 的折扣。对于当前的竞价型实例价格，我们推荐 Vantage 的这个资源。
与按需定价不同，这些竞价型实例价格对于每种实例类型都不是线性的。例如，在撰写本文时，具有四个 vCPU 的 c7g.xlarge 每小时 0.0693 美元（比按需价格便宜 > 50%），这比具有八个内核的 c7g.2xlarge 的价格（0.1157 美元）高出不到 ½。EC2 备用容量供需的长期趋势最终决定了价格。
下一个挑战是最大限度地缩短任何实例处于活动状态的时间。我们需要
- 声明我们的实例并等待它们可用 - 通常约为 30 秒
- 安装 ClickHouse 并配置以形成集群
- 运行查询
- 立即停止实例。
这需要完成的工作的简化，并且没有考虑配置支持性 AWS 资源的需求，包括 VPC、子网、互联网网关和适当的安全组，以便实例可以相互通信。
为了执行 AWS 编排工作并配置 ClickHouse 集群，我们选择了 Pulumi，主要是因为它允许使用 Python 编写基础设施配置代码，使其非常易读，并且允许在一个工具中完成所有工作。我们使用 Dynamic ResourceProvider 扩展了这一点，以支持在集群准备就绪后查询集群。这使我们能够在单个脚本中完成上述所有工作，此处提供。
要重现这些测试，用户只需克隆上述存储库并修改下面显示的 Pulumi 堆栈配置文件，指定区域、可用区、实例数量及其各自的类型，然后在运行 ./run.sh 之前进行配置。
config:
aws:region: us-east-1
1trc:aws_zone: us-east-1d
1trc:instance_type: "c7g.4xlarge"
1trc:number_instances: 16
# this must exist as a key-pair in AWS
1trc:key_name: "<key-pair-name>"
# change as needed
1trc:cluster_password: "clickhouse_admin"
# AMD ami (us-east-1)
1trc:ami: "ami-05d47d29a4c2d19e1"
# Intel AMI (us-east-1)
# 1trc:ami: "ami-0c7217cdde317cfec"
# query to run on the cluster
1trc:query: "SELECT 1"
此脚本将调用 Pulumi up。这将导致实例被配置和配置，并将执行定义的查询作为最后一步。查询完成后，将发出 Pulumi down 以拆除基础设施，并报告最终计时。
我们应该强调，此脚本绝不能替代生产 ClickHouse 集群的正确配置代码。考虑到实例的短暂性（少于 5 分钟），我们接受了一些折衷方案，但这将在生产集群中受到严重限制。具体而言
- 我们不配置任何存储，因为我们会在 S3 中就地查询数据。
- 虽然访问仅限于运行 Pulumi 代码的 IP，但集群通过端口 8213 上的 HTTP 公开，因此我们可以运行所需的查询。在集群存活期间，这代表着最小的风险，但对于生产环境，不建议使用缺少 SSL 的情况。我们的实例间通信网络设置也过于宽松。
- 集群的配置非常有限，没有考虑修改、扩展或升级集群的需求。我们的 ClickHouse 配置也非常少，并且不可自定义。
- 未执行任何测试，也没有考虑可扩展性。
- 我们不处理 AWS 上竞价型实例始终可能发生的潜在竞价型实例中断。我们还假设一组同构的实例类型（这与 AWS 最佳实践 相悖），因为这使事情更容易推理和测试。
虽然功能有限，但上面的脚本确实允许我们廉价地启动 ClickHouse 集群，运行查询并立即销毁基础设施。这种临时集群部署可用于运行对于笔记本电脑而言在计算方面可能遥不可及的查询。例如，要运行 32 个 c7g.4xlarge（16 核，32GiB 内存）实例 5 分钟，我们将支付大约 0.2441 * (5/60) * 32 = 0.65 美元 - ClickHouse 可以使用如此多的资源在 S3 中查询大量数据！
选择 AWS 实例类型
在执行任何测试和查询调整之前，我们想要确定实例类型，主要是为了避免进行详尽的测试，这将证明是耗时的。我们基于简单的推理，旨在在不花费大量资源来寻找完美的硬件配置的情况下产生有竞争力的结果。
作为一个简单的聚合查询，其中站点具有相对较低的基数（数千个），我们怀疑此查询将受 CPU 或网络限制。除了比 AWS 中 Intel/AMD 同类产品更便宜的每个内核价格外，ARM 处理器在我们的公共基准测试（参见 c7g）和测试中表现良好。例如，如果我们考虑大小为 4xlarge（16 个 vCPU）的最新 c7 实例系列
这里的 c7g 代表我们最具成本效益的选择，特别是由于我们不需要 c7gd 提供的额外存储空间。c7gn 虽然是相同的 ARM 处理器，但确实提供了额外的网络带宽，这在非常大的实例尺寸上可能是需要的，以避免网络成为瓶颈。然而，这种额外的网络带宽是有代价的 - 竞价成本的 1.5 倍！
因此，我们对实例类型的决定归结为 c7g 或 c7gn。只要我们水平扩展并且不使用网络可能成为瓶颈的较大实例，c7g 在经济上就是最有意义的。
我们还考虑了稍微便宜一点的 c6g 实例。根据我们的经验，这些实例更难获得，并且网络接口速度较慢，但在详尽的评估过程中，它们可能值得考虑。
作为经验丰富的 ClickHouse 用户，我们相当确信我们可以垂直扩展性能。因此，我们选择特定 c7g 实例的决定归结为每个 vCPU 的价格和可用性。对 c7g 的快速分析表明，较大的实例类型通常在竞价市场上更具价格效率。
| 实例类型 | 每小时平均价格（美元） | vCPU | 每个 vCPU 的成本 |
|---|---|---|---|
| c7g.xlarge | 0.0696 | 4 | 0.0174 |
| c7g.2xlarge | 0.116 | 8 | 0.0145 |
| c7g.4xlarge | 0.2436 | 16 | 0.015225 |
| c7g.8xlarge | 0.522 | 32 | 0.0163125 |
| c7g.12xlarge | 0.7134 | 48 | 0.0148625 |
| c7g.16xlarge | 0.928 | 64 | 0.0145 |
这些价格代表平均值。价格因可用区而异。对于我们的最终测试，我们使用当前最便宜的区域。这可以从 AWS 控制台 确定。
虽然 c7g.16xlarge 是最具成本效益的选择，但我们在测试期间很难获得大量这些实例 - 尽管它们看起来具有 类似的可用性。我们发现 c7g.12xlarge 更容易配置，并且价格差异最小。
优化查询设置
选择了 c7g.12xlarge 作为我们的实例类型后，我们希望确保查询设置是最佳的，以最大限度地减少查询运行时间。对于我们的性能测试，我们使用了单个实例，并使用模式 measurements-1*.parquet 查询了部分数据。如下所示，这针对大约 1110 亿行，并提供 454 秒的初始响应时间。
SELECT station, min(measure), max(measure), round(avg(measure), 2)
FROM s3('https://coiled-datasets-rp.s3.us-east-1.amazonaws.com/1trc/measurements-1*.parquet', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', headers('x-amz-request-payer' = 'requester'))
GROUP BY station
ORDER BY station ASC
FORMAT `Null`
0 rows in set. Elapsed: 454.770 sec. Processed 111.11 billion rows, 279.18 GB (244.32 million rows/s., 613.88 MB/s.)
Peak memory usage: 476.72 MiB.
上面使用了
FORMAT Null，以便不打印结果，并且我们可以轻松识别实际查询时间。由于响应很小，我们认为与完整的查询处理相比，这种开销可以忽略不计。
您可以立即看到，在约 615 MB/秒 的速度下，我们没有受到网络限制（实例支持 22.5Gbps）。我们使用 ClickHouse 仪表板评估了集群的资源利用率。这里突出了三个直接的可视化：用于查询的 CPU 核心数、来自 S3 的读取性能以及 S3 读取等待时间。
此查询使用的 CPU 核心数约为 14-15 个（与 ClickHouse 客户端在执行期间报告的一致），远低于可用的 48 个。在进行任何线程调整之前，我们花了一些时间评估为什么会出现这种情况。S3 读取等待时间表明我们发出了过多的读取请求（这已通过查询跟踪得到证实）。快速浏览我们的文档指出
“对于 s3 函数和表，并行下载由值 max_download_threads 和 max_download_buffer_size 确定。只有当文件大小大于所有线程组合的总缓冲区大小时，才会并行下载文件。这仅在版本 > 22.3.1 上可用。”
Parquet 文件大约为 25MB，max_download_buffer_size 默认设置为 10MiB。考虑到此查询的简单性（它只是读取并进行简单的聚合），我们知道我们可以安全地将此缓冲区大小增加到 50 MB (max_download_buffer_size=52428800)，目的是确保每个文件都由单个线程下载。这将减少每个线程花费在发出 S3 调用上的时间，从而也减少 S3 等待时间。
将缓冲区大小增加到 50MiB
我们可以看到，这使我们的 S3 读取等待时间减少了约 20%，同时将我们的读取吞吐量提高到超过 920 MB/秒。由于等待时间减少，我们更有效地利用了我们的线程，将 CPU 利用率提高到约 23 个核心。所有这些都将我们的执行时间从 486 秒减少到 303 秒。
0 rows in set. Elapsed: 303.462 sec. Processed 111.11 billion rows, 279.18 GB (366.14 million rows/s., 919.97 MB/s.)
Peak memory usage: 534.87 MiB.
在确认进一步增加缓冲区大小没有带来任何好处后，正如预期的那样，并且考虑到在 920MB/秒 的速度下，我们的实例仍然绝对没有受到网络限制，我们考虑了如何进一步提高 CPU 利用率。
默认情况下，此查询的大多数阶段将在每个节点上使用 48 个线程运行，这已通过 EXPLAIN PIPELINE 确认
EXPLAIN PIPELINE
SELECT
station,
min(measure),
max(measure),
round(avg(measure), 2)
FROM s3('https://coiled-datasets-rp.s3.us-east-1.amazonaws.com/1trc/measurements-1*.parquet', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', headers('x-amz-request-payer' = 'requester'))
GROUP BY station
ORDER BY station ASC
┌─explain─────────────────────────────────────┐
│ (Expression) │
│ ExpressionTransform │
│ (Sorting) │
│ MergingSortedTransform 48 → 1 │
│ MergeSortingTransform × 48 │
│ LimitsCheckingTransform × 48 │
│ PartialSortingTransform × 48 │
│ (Expression) │
│ ExpressionTransform × 48 │
│ (Aggregating) │
│ Resize 48 → 48 │
│ AggregatingTransform × 48 │
│ StrictResize 48 → 48 │
│ (Expression) │
│ ExpressionTransform × 48 │
│ (ReadFromStorageS3Step) │
│ S3 × 48 0 → 1 │
└─────────────────────────────────────────────┘
17 rows in set. Elapsed: 0.305 sec.
默认情况下，这默认为每个节点的 vCPU 数量，并且对于大多数查询来说是一个合理的默认值。在我们的例子中，考虑到任务的读取密集型性质，增加此值是有意义的。
理想情况下，我们希望能够在此测试中仅增加下载线程的数量。目前，这在 ClickHouse 中不可用作设置，但它是我们正在考虑的事情。
虽然我们没有进行详尽的测试，但一些简单的测试表明增加 max_threads 到 128 的好处
以上代表 3 次执行的平均值。
这也带来了不错的改进，将总运行时间减少到 138 秒。
0 rows in set. Elapsed: 138.820 sec. Processed 111.11 billion rows, 279.18 GB (800.39 million rows/s., 2.01 GB/s.)
Peak memory usage: 1.36 GiB.
返回我们的高级仪表板，我们可以看到这使我们的读取吞吐量增加到约 2 GiB/秒（仍然没有受到网络限制），并利用了 48 个核心中的 46 个。进一步增加超过 128 的 max_threads 会导致 CPU 争用和 CPU 等待时间 - 从而降低整体吞吐量。
此时，我们确信我们受到 CPU 限制，并且不相信进一步的调整会带来显着收益；我们的最终查询时间为 138 秒，即大约 2 分钟来查询所有 1110 亿行。
单个实例显然不足以在几乎大 10 倍的数据集上为我们提供合理的运行时间。但是，确信我们的单个 c7g.12xlarge 不会受到网络限制，并且使用合理的查询设置，我们水平扩展并解决我们更大的 1 万亿行问题。
当我们在集群中运行查询时，这些相同的设置可以传递到每个节点，以及要处理的数据子集，从而优化每个节点的性能。
使用集群
要查询集群并使用所有节点，我们必须稍微更改我们的查询，使用 s3Cluster 而不是 s3 函数。这使我们能够使用集群中的所有节点来处理查询。
与仅在接收节点上执行查询的标准 s3 函数不同，此函数 ① 首先执行存储桶的列表。协调节点使用此列表 ② 将要处理的文件发送到集群中的每个节点，从而允许工作并行化 ③。然后，协调节点 ④ 整理来自集群中每个节点的中间结果，以 ⑤ 提供最终响应。我们在下面可视化了这一点
我们的最终查询
SELECT station,
min(measure),
max(measure),
round(avg(measure), 2)
FROM s3Cluster('default', 'https://coiled-datasets-rp.s3.us-east-1.amazonaws.com/1trc/measurements-*.parquet', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', headers('x-amz-request-payer' = 'requester'))
GROUP BY station
ORDER BY station ASC
FORMAT `Null`
SETTINGS max_download_buffer_size = 52428800, max_threads = 110
在上面的示例中，与所有 ClickHouse Cloud 集群一样，我们 将我们的集群命名为“default”。
最终计时和成本
虽然查询时间应该随着我们部署的实例数量线性减少，从而保持成本不变，但我们也受到可以实际配置的实例数量的限制。 AWS 通常建议混合实例类型和 EC2 扩展组，而不是我们同构的扩展方法。 我们当前的脚本也限制我们使用单个可用区（虽然这是一个明显的改进，但它使代码稍微复杂化）。 少量测试表明可以可靠地获得八个实例，总共提供 384 个 vCPU。 因此，我们的最终配置如下
config:
aws:region: us-east-1
1trc:aws_zone: us-east-1b
1trc:instance_type: "c7g.12xlarge"
1trc:number_instances: 8
1trc:key_name: "dalem"
1trc:cluster_password: "a_super_password"
# AMD ami (us-east-1)
1trc:ami: "ami-05d47d29a4c2d19e1"
1trc:query: "SELECT station, min(measure), max(measure), round(avg(measure), 2) FROM s3Cluster('default','https://coiled-datasets-rp.s3.us-east-1.amazonaws.com/1trc/measurements-*.parquet', 'AWS_ACCESS_KEY_ID', 'AWS_SECRET_ACCESS_KEY', headers('x-amz-request-payer' = 'requester')) GROUP BY station ORDER BY station ASC SETTINGS max_download_buffer_size = 52428800, max_threads=128"
请注意，我们使用 us-east-1b 作为我们的可用区。 在撰写本文时，这是最便宜的区域，价格为每实例每小时 0.7162 美元（比按需实例节省 58.84%）。
通过我们的最佳设置，我们可以运行我们的运行脚本
./run.sh
(venv) dalemcdiarmid@PY aws-starter % ./run.sh
Updating (dev)
View in Browser (Ctrl+O): https://app.pulumi.com/clickhouse/aws-starter/dev/updates/88
Type Name Status Info
+ ├─ aws:ec2:Instance 1trc_node_4 created (17s)
+ ├─ aws:ec2:Instance 1trc_node_2 created (17s)
+ ├─ aws:ec2:Instance 1trc_node_7 created (18s)
…
+ └─ pulumi-python:dynamic:Resource 1trc-clickhouse-query created (181s)
Diagnostics:
pulumi:pulumi:Stack (aws-starter-dev):
info: checking cluster is ready...
info: cluster is ready!
info: running query...
info: query took 178.94s ← query time!
Resources:
+ 80 created
Duration: 5m10s ← startup time + query time!
…
Destroying (dev)
View in Browser (Ctrl+O): https://app.pulumi.com/clickhouse/aws-starter/dev/updates/89
Type Name Status
pulumi:pulumi:Stack aws-starter-dev
- ├─ command:remote:Command restart_node_3_clickhouse deleting (26s)...
- ├─ command:remote:Command restart_node_2_clickhouse deleting (26s)...
- ├─ command:remote:Command restart_node_4_clickhouse deleting (26s)...
- ├─ command:remote:Command restart_node_5_clickhouse deleting (26s)...
…
Resources:
- 80 deleted
Duration: 2m58s ← time to remove resources
The resources in the stack have been deleted, but the history and configuration associated with the stack are still maintained.
If you want to remove the stack completely, run `pulumi stack rm dev`.
Total time: 499 seconds
从输出中，我们可以看到我们的查询在 180 秒或 3 分钟内完成。 但是，我们的集群花费了大约 2 分 10 秒来配置并可用于查询，然后又花费了 3 分钟来终止。 因此，我们的总运行时间为 499 秒。
因此，我们可以使用它来计算我们的最终成本。
每个实例的成本：$0.7162
实例数量：8
活动时间：499 秒
成本：(499/3600) * 8 * 0.7162 = $0.79
这仅代表成本的估计值。 有几个因素可能导致实际成本有所不同 - 主要是并非所有实例都在整个 499 秒内处于活动状态。 通过订阅 Spot 实例数据馈送，可以获得确切的成本。 这会将与 Spot 实例相关的成本发送到 S3 存储桶，以便稍后检索。 检查我们测试运行的统计数据，我们可以看到每个实例的实际产生费用
将这些加起来就是我们查询 1 万亿行的总价
(0.0708400000+0.0700529676+0.0704461296+0.0702499028+0.0702499028+0.0696590972+0.0706430648+0.0706430648) = $0.56
最终想法
值得注意的是，虽然更多实例将减少查询时间，但它们将需要更长的时间来配置，可能会产生更多成本。 因此，在水平扩展之前进行垂直扩展是有意义的，因为这会主导运行时间。 但是，如果您要运行多个查询，则此时间将被抵消，因为它是一次性成本！
作为一个不声称自己是 Pulumi 专家的人，我也怀疑可以改进配置（和终止代码），以最大限度地缩短启动和关闭时间。 和以往一样，我们欢迎贡献！
在开发这种方法时，我们并没有严格遵守 AWS 关于 Spot 实例的最佳实践。 改进后的代码版本可以使用异构实例类型，甚至利用 EC2 Auto Scaling 组，以允许用户简单地指定所需的核心数！ 预装 ClickHouse 的 AMI 可以进一步缩短启动时间，而 AWS 区域的灵活性可能会使扩展更容易。
精明的读者会注意到我们没有使用竞价来配置我们的 spot 实例。 我们只是接受了当前小时的 spot 价格。 如果您真的想节省几美分，还有降低价格的空间！
使用 MergeTree
最后，我们很好奇在 MergeTree 表引擎中查询这些数据的速度有多快。 虽然将数据加载到表中也需要一些时间（在 300 核 ClickHouse Cloud 集群上花费了 385 秒），但使用全表扫描查询结果表仅需 16.5 秒，并且比查询 parquet 文件的任何结果都快得多（同一集群在 48 秒内完成查询）！ 请注意，可以通过 ClickHouse Cloud API 轻松部署此类集群，用于运行查询，然后立即终止。 感谢 Alexey Milovidov 提供这些时间数据。
结论
在十亿行挑战的基础上，我们展示了 ClickHouse 如何在 3 分钟内以 0.56 美元的价格查询更大的 1 万亿数据集！！
我们欢迎任何关于更快、更经济高效地查询此数据集的建议或替代方案。
