使用 ClickHouse 深入了解 Apache Parquet （一）

云原生数据库 2023-10-19

544

自2013年作为Hadoop的列存储发布以来，Parquet几乎已经成为一种无处不在的文件交换格式，它提供了高效的存储和检索。这种采纳使其成为更近期的数据湖格式的基础，例如Apache Iceberg。在这个博客系列中，我们探讨如何使用ClickHouse读写这种格式，然后更详细地深入了解Parquet。对于更有经验的Parquet用户，我们还讨论了用户在使用ClickHouse编写Parquet文件时可以进行的一些优化，以最大化压缩，以及一些最近的开发，以使用并行化优化读取性能。

在我们的示例中，我们使用了英国房价数据集。这包含了从1995年到写作时的英格兰和威尔士的房地产价格支付数据。我们在公共s3桶s3://datasets-documentation/uk-house-prices/parquet/中以Parquet格式分发这些数据。

ClickHouse 本地版

在我们的示例中，我们使用本地和S3托管的Parquet文件，并使用ClickHouse本地版对其进行查询。ClickHouse本地版是ClickHouse的一个易于使用的版本，非常适合需要在本地和远程文件上使用SQL进行快速处理而不必安装完整的数据库服务器的开发人员。为了使用您的笔记本电脑或工作站上的本地计算资源进行数据分析而设计并优化，用户可以使用仅使用SQL且无需编写一行Python代码的方式查询、筛选和转换几乎任何格式的数据文件。我们建议查看这篇最近的博客文章，以了解这个工具的功能概述。最重要的是，ClickHouse本地版和ClickHouse服务器对于Parquet的读写使用相同的代码，因此任何解释都适用于两者。

什么是Parquet？

Apache Parquet的官方描述为我们提供了一个关于其设计和属性的出色概述：“Apache Parquet是一个开源的、面向列的数据文件格式，旨在实现高效的数据存储和检索。”

与ClickHouse的MergeTree格式类似，数据是面向列存储的。这实际上意味着同一列的值存储在一起，与面向行的文件格式（例如，Avro）形成对比，后者的行数据是共同存放的。

这种数据方向，加上支持多种适合现代处理器管道的编码技术，允许实现高压缩率和高效存储属性。列方向还最大限度地减少了读取的数据量，因为只从存储中读取必要的列以进行分组等分析查询。当与高压缩率和每列提供的内部统计数据（存储为元数据）结合时，Parquet还承诺快速检索。

这后一个属性在很大程度上取决于完全利用元数据、任何查询引擎中的并行化程度，以及存储数据时所做的决策。我们将在下面讨论与ClickHouse的关系。

在我们深入了解Parquet的内部结构之前，我们将介绍ClickHouse如何支持写入和读取这种格式。

使用ClickHouse查询Parquet

在下面的示例中，我们假设我们的房价数据已导出到一个名为house_prices.parquet的文件，并使用ClickHouse Local进行查询，除非另有说明。

读取模式

可以使用DESCRIBE语句和文件表函数来识别任何文件的模式：

DESCRIBE TABLE file('house_prices.parquet')


┌─name──────┬─type─────────────┬
│ price     │ Nullable(UInt32) │
│ date      │ Nullable(UInt16) │
│ postcode1 │ Nullable(String) │
│ postcode2 │ Nullable(String) │
│ type      │ Nullable(Int8)   │
│ is_new    │ Nullable(UInt8)  │
│ duration  │ Nullable(Int8)   │
│ addr1     │ Nullable(String) │
│ addr2     │ Nullable(String) │
│ street    │ Nullable(String) │
│ locality  │ Nullable(String) │
│ town      │ Nullable(String) │
│ district  │ Nullable(String) │
│ county    │ Nullable(String) |
└───────────┴──────────────────┴

查询本地文件

上述文件表功能可作为SELECT查询的输入，允许我们执行对Parquet文件的查询。下面，我们计算伦敦房产每年的平均价格。

SELECT
    toYear(toDate(date)) AS year,
    round(avg(price)) AS price,
    bar(price, 0, 2000000, 100)
FROM file('house_prices.parquet')
WHERE town = 'LONDON'
GROUP BY year
ORDER BY year ASC


┌─year─┬───price─┬─bar(round(avg(price)), 0, 2000000, 100)──────────────┐
│ 1995 │  109120 │ █████▍                                                │
│ 1996 │  118672 │ █████▉                                                │
│ 1997 │  136530 │ ██████▊                                               │
│ 1998 │  153014 │ ███████▋                                              │
│ 1999 │  180639 │ █████████                                             │
│ 2000 │  215860 │ ██████████▊                                           │
│ 2001 │  232998 │ ███████████▋                                          │
│ 2002 │  263690 │ █████████████▏                                        │
│ 2003 │  278423 │ █████████████▉                                        │
│ 2004 │  304666 │ ███████████████▏                                      │
│ 2005 │  322886 │ ████████████████▏                                     │
│ 2006 │  356189 │ █████████████████▊                                    │
│ 2007 │  404065 │ ████████████████████▏                                 │
│ 2008 │  420741 │ █████████████████████                                 │
│ 2009 │  427767 │ █████████████████████▍                                │
│ 2010 │  480329 │ ████████████████████████                              │
│ 2011 │  496293 │ ████████████████████████▊                             │
│ 2012 │  519473 │ █████████████████████████▉                            │
│ 2013 │  616182 │ ██████████████████████████████▊                       │
│ 2014 │  724107 │ ████████████████████████████████████▏                 │
│ 2015 │  792274 │ ███████████████████████████████████████▌              │
│ 2016 │  843685 │ ██████████████████████████████████████████▏           │
│ 2017 │  983673 │ █████████████████████████████████████████████████▏    │
│ 2018 │ 1016702 │ ██████████████████████████████████████████████████▊  │
│ 2019 │ 1041915 │ ████████████████████████████████████████████████████ │
│ 2020 │ 1060936 │ █████████████████████████████████████████████████████│
│ 2021 │  968152 │ ████████████████████████████████████████████████▍     │
│ 2022 │  967439 │ ████████████████████████████████████████████████▎     │
│ 2023 │  830317 │ █████████████████████████████████████████▌            │
└──────┴─────────┴──────────────────────────────────────────────────────┘


29 rows in set. Elapsed: 0.625 sec. Processed 28.11 million rows, 750.65 MB (44.97 million rows/s., 1.20 GB/s.)

查询S3上的文件

虽然上述文件功能可以与ClickHouse服务器实例一起使用，但它要求文件存在于配置的user_files_path目录下的服务器文件系统上。在这些情况下，Parquet文件更自然地从S3读取。这在数据湖使用案例中是一个常见的需求，需要进行即席分析。在这种情况下，上述文件功能可以用s3功能替代，用于类似AWS Athena的查询：

SELECT
    toYear(toDate(date)) AS year,
    round(avg(price)) AS price,
    bar(price, 0, 2000000, 100)
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/uk-house-prices/parquet/house_prices_all.parquet')
WHERE town = 'LONDON'
GROUP BY year
ORDER BY year ASC


┌─year─┬───price─┬─bar(round(avg(price)), 0, 2000000, 100)───────────────┐
│ 1995 │  109120 │ █████▍                                                 │
│ 1996 │  118672 │ █████▉                                                  │
│ 1997 │  136530 │ ██████▊                                                │
│ 1998 │  153014 │ ███████▋                                               │
│ 1999 │  180639 │ █████████                                              │


...


29 rows in set. Elapsed: 2.069 sec. Processed 28.11 million rows, 750.65 MB (13.59 million rows/s., 362.87 MB/s.)

查询多个文件

这两个功能都支持通配符模式，允许选择文件的子集。正如我们稍后将讨论的，这不仅仅提供了跨文件查询的优势 - 主要是读取的并行化。下面我们将查询限制在所有带有年份后缀的house_prices_文件上 - 这假设我们每年有一个文件（见下文）。

SELECT
    toYear(toDate(date)) AS year,
    round(avg(price)) AS price,
    bar(price, 0, 2000000, 100)
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/uk-house-prices/parquet/house_prices_{1..2}*.parquet')
WHERE town = 'LONDON'
GROUP BY year
ORDER BY year ASC


29 rows in set. Elapsed: 3.387 sec. Processed 28.11 million rows, 750.65 MB (8.30 million rows/s., 221.66 MB/s.)

用户还应该了解s3Cluster功能，该功能允许从集群中的多个节点并行处理文件 - 尤其与ClickHouse Cloud用户相关。在需要读取许多文件的情况下，这可以提供显著的性能优势（允许工作分布）。

用ClickHouse写Parquet

在ClickHouse中将表数据写入Parquet文件可以通过几种方式实现。这里的首选选项通常取决于你是使用ClickHouse Server还是ClickHouse Local。在下面的示例中，我们假设uk_price_paid表已经填充了数据。关于加载这些数据的详细信息，请看这里。

写本地文件

使用INTO FUNCTION子句，我们可以使用与读取相同的文件函数写parquet。这最适合于ClickHouse Local，文件可以写入本地文件系统的任何位置。ClickHouse服务器将它们写入由配置参数user_files_path指定的目录。

INSERT INTO FUNCTION file('house_prices.parquet') SELECT *
FROM uk_price_paid


0 rows in set. Elapsed: 12.490 sec. Processed 28.11 million rows, 1.32 GB (2.25 million rows/s., 105.97 MB/s.)


dalemcdiarmid@dales-mac houseprices % ls -lh house_prices.parquet
-rw-r-----  1 dalemcdiarmid  staff   243M 17 Apr 16:59 house_prices.parquet

在大多数情况下，包括 ClickHouse Cloud，本地服务器文件系统是不可访问的。在这些情况下，用户可以通过连接clickhouse-client并利用 INTO OUTFILE 子句将 parquet 文件写入客户端的文件系统。此处将根据文件扩展名自动检测所需的输出格式。

SELECT *
FROM uk_price_paid
INTO OUTFILE 'house_prices.parquet'


28113076 rows in set. Elapsed: 15.690 sec. Processed 28.11 million rows, 2.47 GB (1.79 million rows/s., 157.47 MB/s.)


clickhouse@clickhouse-mac ~ % ls -lh house_prices.parquet
-rw-r--r--  1 dalemcdiarmid  staff   291M 17 Apr 18:23 house_prices.parquet

或者，用户可以简单地发出 SELECT 查询，指定输出格式为 Parquet，并将结果重定向到文件。在这个例子中，我们将--query参数从终端传递给客户端。

clickhouse@clickhouse-mac ~ % ./clickhouse client --query "SELECT * FROM uk_price_paid FORMAT Parquet" > house_price.parquet

最后两种方法产生的文件比我们之前的文件函数方法稍大。我们将在本系列的第 2 部分中解释原因，但目前建议用户尽可能使用早期的INSERT INTO FUNCTION方法以获得更佳的尺寸。

将文件写入S3

通常，客户端存储是有限的。在这些情况下，用户可能希望将文件写入对象存储，例如 S3 和 GCS。这些都通过用于读取的相同s3 函数来支持。请注意，需要凭证 - 在下面的示例中，我们将它们作为函数参数传递，但也支持 IAM 凭证。

INSERT INTO FUNCTION s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/uk-house-prices/parquet/house_prices_sample.parquet', '<aws_access_key_id>', '<aws_secret_access_key>') SELECT *
FROM uk_price_paid
LIMIT 1000


0 rows in set. Elapsed: 0.726 sec. Processed 2.00 thousand rows, 987.86 KB (2.75 thousand rows/s., 1.36 MB/s.)

写入多个文件

最后，通常需要限制任何单个 Parquet 文件的大小。为了帮助写入文件，用户可以将 PARTITION BY 子句与该INTO FUNCTION子句结合使用。它接受任何 SQL 表达式来为结果集中的每一行创建分区 id。parition_id反过来，这可以在文件路径中使用，以确保将行分配给不同的文件。在下面的示例中，我们按年份进行分区。因此，同一年的房屋销售情况将写入同一文件。文件将以其各自的年份作为后缀，如图所示。

INSERT INTO FUNCTION file('house_prices_{_partition_id}.parquet') PARTITION BY toYear(date) SELECT * FROM uk_price_paid


0 rows in set. Elapsed: 23.281 sec. Processed 28.11 million rows, 1.32 GB (1.21 million rows/s., 56.85 MB/s.)


clickhouse@clickhouse-mac houseprices % ls house_prices_*
house_prices_1995.parquet    house_prices_2001.parquet    house_prices_2007.parquet    house_prices_2013.parquet    house_prices_2019.parquet
house_prices_1996.parquet    house_prices_2002.parquet    house_prices_2008.parquet    house_prices_2014.parquet    house_prices_2020.parquet
house_prices_1997.parquet    house_prices_2003.parquet    house_prices_2009.parquet    house_prices_2015.parquet    house_prices_2021.parquet
house_prices_1998.parquet    house_prices_2004.parquet    house_prices_2010.parquet    house_prices_2016.parquet    house_prices_2022.parquet
house_prices_1999.parquet    house_prices_2005.parquet    house_prices_2011.parquet    house_prices_2017.parquet    house_prices_2023.parquet
house_prices_2000.parquet    house_prices_2006.parquet    house_prices_2012.parquet    house_prices_2018.parquet

同样的方法可以用于 s3 函数。

INSERT INTO FUNCTION s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/uk-house-prices/parquet/house_prices_sample_{_partition_id}.parquet', '<aws_access_key_id>', '<aws_secret_access_key>') PARTITION BY toYear(date) SELECT *
FROM uk_price_paid
LIMIT 1000


0 rows in set. Elapsed: 2.247 sec. Processed 2.00 thousand rows, 987.86 KB (889.92 rows/s., 439.56 KB/s.)

在撰写本文时，INTO OUTFILE不支持这个PARTITION BY子句。

转换文件到Parquet格式

结合以上内容，我们可以使用ClickHouse Local在格式之间转换文件。在下面的示例中，我们使用ClickHouse Local和文件函数读取CSV格式的本地副本房价数据集，该数据集包含所有的2800万行，然后将其写入S3作为Parquet。正如之前所示，这些文件按年对数据进行了分区。

INSERT INTO FUNCTION s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/uk-house-prices/parquet/house_prices_sample_{_partition_id}.parquet', '<aws_access_key_id>', '<aws_secret_access_key>') PARTITION BY toYear(date) SELECT *
FROM file('house_prices.csv')


0 rows in set. Elapsed: 223.864 sec. Processed 28.11 million rows, 5.87 GB (125.58 thousand rows/s., 26.24 MB/s.)

将 Parquet 文件插入 ClickHouse

前面的所有示例都假设用户正在查询本地和 S3 托管的文件以进行临时分析，或将数据从 ClickHouse 迁移到 Parquet 进行分发。虽然 Parquet 是一种与数据存储无关的文件分发格式，但它的查询效率不如 ClickHouse MergeTree 表，后者能够利用索引和特定于格式的优化。考虑以下查询的性能，该查询使用本地 Parquet 文件和具有推荐架构的MergeTree 表（均在 Macbook Pro 2021 上运行）计算伦敦房产的每年平均价格：

SELECT
    toYear(toDate(date)) AS year,
    round(avg(price)) AS price,
    bar(price, 0, 2000000, 100)
FROM file('house_prices.parquet')
WHERE town = 'LONDON'
GROUP BY year
ORDER BY year ASC


29 rows in set. Elapsed: 0.625 sec. Processed 28.11 million rows, 750.65 MB (44.97 million rows/s., 1.20 GB/s.)


SELECT
    toYear(toDate(date)) AS year,
    round(avg(price)) AS price,
    bar(price, 0, 2000000, 100)
FROM uk_price_paid
WHERE town = 'LONDON'
GROUP BY year
ORDER BY year ASC


29 rows in set. Elapsed: 0.022 sec.

这里的差异是巨大的，并且解释了为什么对于需要实时性能的大型数据集，用户将 Parquet 文件加载到 ClickHouse 中。下面我们假设该uk_price_paid表已预先创建。

从本地文件加载

可以使用该子句从客户端计算机加载文件INFILE。clickhouse-client以下查询从本地客户端的文件系统执行并读取数据。

INSERT INTO uk_price_paid FROM INFILE 'house_price.parquet' FORMAT Parquet
28113076 rows in set. Elapsed: 15.412 sec. Processed 28.11 million rows, 1.27 GB (1.82 million rows/s., 82.61 MB/s.)

如果用户的数据分布在多个 Parquet 文件中，此方法还支持 glob 模式。clickhouse-client或者，可以使用以下参数将 Parquet 文件重定向到--query：

clickhouse@clickhouse-mac ~ % ~/clickhouse client --query "INSERT INTO uk_price_paid FORMAT Parquet" < house_price.parquet

从 S3 加载

由于客户端存储通常有限，并且随着基于对象存储的数据湖的兴起，Parquet 文件通常驻留在 S3 或 GCS 上。同样，我们可以使用 s3 函数读取这些文件，并使用INSERT INTO SELECT子句将它们的数据插入到 MergeTree 表中。在下面的示例中，我们利用 glob 模式读取按年份分区的文件，并在三节点 ClickHouse Cloud 集群上执行此查询。

INSERT INTO uk_price_paid SELECT *
FROM s3('https://datasets-documentation.s3.eu-west-3.amazonaws.com/uk-house-prices/parquet/house_prices_{1..2}*.parquet')


0 rows in set. Elapsed: 12.028 sec. Processed 28.11 million rows, 4.64 GB (2.34 million rows/s., 385.96 MB/s.)

与读取类似，可以使用s3Cluster函数来加速。为了确保插入和读取分布在整个集群中，parallel_distributed_insert_select必须启用该设置（否则，仅分布读取，而插入将发送到协调节点）。以下查询在上一个示例中使用的同一云集群上运行，显示了并行化这项工作的好处。

SET parallel_distributed_insert_select=1
INSERT INTO uk_price_paid SELECT *
FROM s3Cluster('default', 'https://datasets-documentation.s3.eu-west-3.amazonaws.com/uk-house-prices/parquet/house_prices_{1..2}*.parquet')


0 rows in set. Elapsed: 6.425 sec. Processed 28.11 million rows, 4.64 GB (4.38 million rows/s., 722.58 MB/s.)

结论

在本系列的第一部分中，我们介绍了 Parquet 格式，并展示了如何使用 ClickHouse 查询和编写该格式。在下一篇文章中，我们将更详细地探讨该格式，进一步探索 ClickHouse 集成以及最近的性能改进和优化查询的技巧。

文章转载自云原生数据库，如果涉嫌侵权，请发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。

使用 ClickHouse 深入了解 Apache Parquet （一）

评论