Углубляемся в тонкости Apache Parquet с помощью ClickHouse - часть 1

С момента своего появления в 2013 году в качестве колоночно-ориентированного хранилища для Hadoop, Parquet стал одним из самых популярных форматов файлов, обеспечивающим эффективное хранение и поиск данных. Более того, с течением времени он превратился в базу современных Data Lake, например Apache Iceberg. В этой серии статей мы расскажем Вам о том, как можно использовать ClickHouse для чтения и записи файлов этого формата. Для более опытных пользователей Parquet мы также обсудим некоторые возможные оптимизации, которые можно осуществить при записи файлов Parquet с помощью ClickHouse, а также некоторые самые последние разработки для оптимизации производительности чтения с помощью процесса распараллеливания.

В этой статье в демонстрационных целях мы будем использовать Набор данных о стоимости жилья в Великобритании , содержащий данные начиная с 1995 года. Загрузим данный набор с помощью команды s3://datasets-documentation/uk-house-prices/parquet/.

ClickHouse - local

В рамках нашего проекта мы будем использовать локальные и размещенные на S3 файлы Parquet. Запрашиваь их будем с помощью ClickHouse- Local. ClickHouse Local - это простая в использовании версия ClickHouse, которая идеально подходит для разработчиков, которым необходимо выполнять быструю обработку локальных и удаленных файлов с помощью SQL без необходимости установки полноценного сервера баз данных. Данная система позволяет пользователям запрашивать, фильтровать и преобразовывать файлы практически любого формата, используя только SQL.

Что такое Parquet?

Официальное определение Apache Parquet звучит следующим образом: «Apache Parquet - это колоночно-ориентированный формат файлов с открытым исходным кодом, предназначенный для эффективного хранения и поиска данных».

Подобно формату MergeTree от ClickHouse, данные хранятся в формате, ориентированном на столбцы. Это означает то, что значения одного и того же столбца хранятся вместе, в отличие от форматов файлов, ориентированных на строки (Avro), где данные в строках располагаются вместе.

Симбиоз такого способа организации данных и поддержки ряда методов кодирования позволяет добиться высокой степени сжатия файлов и эффективности хранения данных. Помимо всего прочего ориентация на столбцы о минимизирует объем считываемых данных, поскольку для аналитических запросов, таких как group by, из хранилища считываются только необходимые столбцы. В сочетании с высокой степенью сжатия и внутренней статистикой, предоставляемой по каждому столбцу (хранится в виде метаданных), Parquet обеспечивает быстрый поиск необходимых данных.

Последнее свойство во многом зависит от полноты использования метаданных, уровня распараллеливания и решений, принятых при хранении данных. Ниже мы рассмотрим все эти аспекты применительно к ClickHouse.

Прежде чем мы углубимся во внутреннее устройство Parquet, давайте попробуем разобраться в том, как ClickHouse поддерживает запись и чтение файлов этого формата.

Запрос к файлам Parquet с помощью ClickHouse

В приведенном ниже примере мы предполагаем, что наши данные о ценах на жилье были экспортированы в файл 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. Это распространенное требование в случаях использования Data Lake, когда не обойтись без специального анализа. В случае запросов типа AWS Athena вышеупомянутую функцию можно совершенно спокойно заменить функцией S3:

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.)

Запрос к нескольким файлам

Обе эти функции поддерживают шаблоны glob, позволяющие выбирать подмножества файлов. Это дает преимущество распараллеливание операций чтения. Ниже мы ограничиваем наш запрос всеми файлами house_prices_, содержащими суффикс year (предполагается, что у нас есть по одному файлу на год (см. ниже).

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. Такая возможность может обеспечить значительный прирост производительности, особенно в случаях, когда необходимо прочитать много файлов (что позволяет равномерно распределить работу).

Создание файлов Parquet с помощью ClickHouse

Запись данных таблицы 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

Эти два подхода создают файл несколько большего размера по сравнению с тем, что мы получали в случае использования подхода с файловой функцией. Почему так происходит? Об этом мы поговорим во второй части нашей серии статей, а пока, для получения более оптимальных размеров пользователям рекомендуется использовать подход 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 для создания идентификатора раздела для каждой строки в наборе результатов. В свою очередь,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.)

На момент написания этой статьи предложение PARTITION BY не применимо для INTO OUTFILE.

Конвертация файлов в формат Parquet

Все подходы и шаги, описанные выше, позволяет нам конвертировать файлы в нужный формат с помощью ClickHouse-Local. В примере, приведенном ниже, для чтения локальной копии набора данных о ценах на жилье в формате CSV, содержащего все 28 млн строк мы используем ClickHouse Local. Операция чтения осуществляется непосредственно перед записью файлов в 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.

Разница в данном случае существенна, и это еще раз объясняет, почему для больших массивов данных, требующих обработки в режиме реального времени, пользователи загружают в ClickHouse файлы Parquet. Ниже мы предполагаем, что таблица 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.)

Этот подход также поддерживает шаблоны glob в том случае, если данные пользователя распределены по нескольким файлам Parquet. В качестве альтернативы файлы Parquet могут быть перенаправлены в clickhouse-client с помощью параметра –query:

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

Загрузка из S3

Поскольку возможности клиентских хранилищ часто ограничены, файлы Parquet зачастую размещаются на S3 или GCS. Давайте снова используем функцию s3 для чтения этих файлов, вставляя их данные в таблицу MergeTree с помощью предложения INSERT INTO SELECT. В приведенном ниже примере мы используем шаблон 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 (в противном случае распределяться будет только чтение, вставки будут отправляться на координирующий узел). Следующий запрос выполняется на том же кластере Cloud, который использовался в предыдущем примере:

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, способы повышения производительности, а также советы по оптимизации запросов от профессионалов в области работы с данными.

← Предыдущая статья

Создание визуализации ClickHouse с помощью Altinity и Cube

Следующая статья →

Углубляемся в тонкости Apache Parquet с помощью ClickHouse - часть 2