Заполнение данных

Независимо от того, являетесь ли вы новым пользователем ClickHouse или отвечаете за существующее развертывание, пользователям неизбежно потребуется заполнить таблицы историческими данными. В некоторых случаях это относительно просто, но может стать более сложным, когда необходимо заполнить материализованные представления. Этот гид документирует некоторые процессы для этой задачи, которые пользователи могут применить к своему случаю использования.

примечание

В этом руководстве предполагается, что пользователи уже знакомы с концепцией Инкрементных материализованных представлений и загрузки данных с помощью функций таблиц, таких как s3 и gcs. Мы также рекомендуем пользователям прочитать наше руководство по оптимизации производительности вставки из объектного хранилища, советы из которого можно применить к вставкам на протяжении всего этого руководства.

Пример набора данных

Протягом этого руководства мы используем набор данных PyPI. Каждая строка в этом наборе данных представляет собой загрузку пакета Python с использованием такого инструмента, как pip.

Например, этот поднабор охватывает один день - 2024-12-17 и доступен публично по адресу https://datasets-documentation.s3.eu-west-3.amazonaws.com/pypi/2024-12-17/. Пользователи могут выполнить запрос с помощью:

Полный набор данных для этого бакета содержит более 320 ГБ файлов в формате parquet. В приведенных ниже примерах мы намеренно нацеливаемся на поднаборы, используя шаблоны glob.

Мы предполагаем, что пользователь потребляет поток этих данных, например, из Kafka или объектного хранилища, для данных после этой даты. Схема этих данных показана ниже:

примечание

Полный набор данных PyPI, состоящий более чем из 1 триллиона строк, доступен в нашей публичной демонстрационной среде clickpy.clickhouse.com. Для получения дополнительной информации об этом наборе данных, включая то, как демонстрация использует материализованные представления для повышения производительности и как данные заполняются ежедневно, см. здесь.

Сценарии заполнения

Заполнение данных обычно необходимо, когда поток данных поступает из определенного момента во времени. Эти данные вставляются в таблицы ClickHouse с инкрементными материализованными представлениями, что инициируется по блокам по мере их вставки. Эти представления могут выполнять трансформацию данных перед вставкой или вычислять агрегаты и отправлять результаты в целевые таблицы для дальнейшего использования в приложениях нижнего потока.

Мы постараемся охватить следующие сценарии:

1. Заполнение данных с уже существующей загрузкой данных - Загружаются новые данные, и необходимо заполнить исторические данные. Эти исторические данные были идентифицированы.
2. Добавление материализованных представлений к существующим таблицам - Необходимо добавить новые материализованные представления к настройке, где исторические данные были заполнены, и данные уже стримятся.

Мы предполагаем, что данные будут заполняться из объектного хранилища. В каждом случае мы стремимся избежать пауз в вставке данных.

Мы рекомендуем заполнять исторические данные из объектного хранилища. Данные должны экспортироваться в формат Parquet, где это возможно, для оптимальной производительности чтения и сжатия (уменьшение сетевой передачи). Размер файла около 150 МБ обычно предпочтителен, но ClickHouse поддерживает более 70 форматов файлов и способен обрабатывать файлы любого размера.

Использование дублирующих таблиц и представлений

Для всех сценариев мы используем концепцию "дублирующих таблиц и представлений". Эти таблицы и представления представляют собой копии тех, которые используются для потоковых данных в реальном времени и позволяют выполнять заполнение в изолированном состоянии с легким способом восстановления в случае сбоя. Например, у нас есть следующая основная таблица pypi и материализированное представление, которое вычисляет количество загрузок для каждого проекта Python:

Мы заполняем основную таблицу и связанное представление с поднабором данных:

Предположим, мы желаем загрузить еще один поднабор {101..200}. Хотя мы могли бы вставить прямо в pypi, мы можем выполнить это заполнение в изоляции, создав дублирующие таблицы.

Если заполнение завершается неудачно, мы не повлияли на наши основные таблицы и можем просто обрезать наши дублирующие таблицы и повторить процесс.

Чтобы создать новые копии этих представлений, мы можем использовать предложение CREATE TABLE AS с суффиксом _v2:

Мы заполняем это нашим вторым поднабором примерно того же размера и подтверждаем успешную загрузку.

Если мы столкнулись с неудачей в любой момент во время этой второй загрузки, мы можем просто обрезать наши pypi_v2 и pypi_downloads_v2 и повторить загрузку данных.

Когда загрузка данных завершена, мы можем переместить данные из наших дублирующих таблиц в основные таблицы с помощью предложения ALTER TABLE MOVE PARTITION.

Имена партиций

В приведенном выше вызове MOVE PARTITION используется имя партиции (). Это представляет собой единственную партицию для этой таблицы (которая не разбита на партиции). Для таблиц, которые разбиты на партиции, пользователям необходимо будет вызвать несколько MOVE PARTITION - по одной для каждой партиции. Название текущих партиций можно установить из таблицы system.parts, например, SELECT DISTINCT partition FROM system.parts WHERE (table = 'pypi_v2').

Теперь мы можем подтвердить, что pypi и pypi_downloads содержат полные данные. pypi_downloads_v2 и pypi_v2 можно безопасно удалить.

Важно отметить, что операция MOVE PARTITION является как легковесной (используя жесткие ссылки), так и атомарной, т.е. она либо завершается неудачей, либо успехом без промежуточного состояния.

Мы активно используем этот процесс в наших сценариях заполнения ниже.

Обратите внимание, как этот процесс требует от пользователей выбирать размер каждой операции вставки.

Большие вставки, т.е. большее количество строк, означают, что потребуется меньше операций MOVE PARTITION. Однако это должно быть сбалансировано с затратами в случае сбоя вставки, например, из-за сетевого прерывания, для восстановления. Пользователи могут дополнить этот процесс пакетированием файлов, чтобы уменьшить риск. Это можно выполнить с помощью диапазонных запросов, например, WHERE timestamp BETWEEN 2024-12-17 09:00:00 AND 2024-12-17 10:00:00 или шаблонов glob. Например,

примечание

ClickPipes использует этот подход при загрузке данных из объектного хранилища, автоматически создавая дубликаты целевой таблицы и ее материализованных представлений и избегая необходимости пользователю выполнять вышеуказанные шаги. Используя также несколько потоков-работников, каждый из которых обрабатывает разные поднаборы (через шаблоны glob) и имея свои дублирующие таблицы, данные могут быть загружены быстро с семантикой exactly-once. Для заинтересованных пользователей дополнительные сведения можно найти в этом блоге.

Сценарий 1: Заполнение данных с уже существующей загрузкой данных

В этом сценарии мы предполагаем, что данные для заполнения не находятся в изолированном бакете, и поэтому требуется фильтрация. Данные уже вставляются, и можно идентифицировать отметку времени или монотонно возрастающий столбец, из которого нужно заполнить исторические данные.

Этот процесс следует следующим шагам:

1. Определить контрольную точку - либо отметку времени, либо значение столбца, из которого необходимо восстановить исторические данные.
2. Создать дубликаты основной таблицы и целевых таблиц для материализованных представлений.
3. Создать копии любых материализованных представлений, указывающих на целевые таблицы, созданные на шаге (2).
4. Вставить данные в нашу дубликатную основную таблицу, созданную на шаге (2).
5. Переместить все партиции из дублирующих таблиц в их оригинальные версии. Удалить дублирующие таблицы.

Например, в данных PyPI предположим, что у нас уже загружены данные. Мы можем определить минимальную отметку времени, а, следовательно, и нашу "контрольную точку".

Из этого мы знаем, что необходимо загрузить данные до 2024-12-17 09:00:00. Используя наш прежний процесс, мы создаем дублирующие таблицы и представления и загружаем поднабор, используя фильтр по отметке времени.

примечание

Фильтрация по столбцам отметок времени в Parquet может быть очень эффективной. ClickHouse будет читать только столбец отметки времени, чтобы определить полный диапазон данных для загрузки, минимизируя сетевой трафик. Индексы Parquet, такие как min-max, также могут быть использованы движком запросов ClickHouse.

После завершения этой вставки мы можем переместить связанные партиции.

Если исторические данные находятся в изолированном бакете, то вышеуказанный временной фильтр не требуется. Если временной или монотонный столбец недоступен, изолируйте ваши исторические данные.

Просто используйте ClickPipes в ClickHouse Cloud

Пользователи ClickHouse Cloud должны использовать ClickPipes для восстановления исторических резервных копий, если данные могут быть изолированы в своем собственном бакете (и фильтрация не требует). Кроме параллелизации загрузки с несколькими работниками, что сокращает время загрузки, ClickPipes автоматизирует вышеуказанный процесс - создавая дублирующие таблицы как для основной таблицы, так и для материализованных представлений.

Сценарий 2: Добавление материализованных представлений к существующим таблицам

Не редко появляется необходимость добавления новых материализованных представлений в настройку, для которой были заполнены значительные данные, и данные продолжают вставляться. В этом случае полезными являются отметка времени или монотонный увеличенный столбец, которые могут быть использованы для определения момента в потоке и избегания пауз в вставке данных. В приведенных ниже примерах мы предполагаем оба случая, предпочитая методы, которые избегают пауз в вставке.

Избегайте команды POPULATE

Мы не рекомендуем использовать команду POPULATE для заполнения материализованных представлений для чего-либо кроме небольших наборов данных, когда вставка приостановлена. Эта операция может пропустить строки, вставленные в ее исходную таблицу, если материализованное представление создано после корректировки хеша заполнения. Кроме того, это заполнение выполняется для всех данных и уязвимо к прерываниям или ограничениям по памяти на больших наборах данных.

Доступна отметка времени или монотонно увеличенный столбец

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

Наш самый простой подход включает следующие шаги:

1. Создать материализованное представление с фильтром, который учитывает только строки больше неопределенного времени в ближайшем будущем.
2. Выполнить запрос INSERT INTO SELECT, который вставляет в целевую таблицу нашего материализованного представления, считывая из исходной таблицы с запросом агрегации представления.

Это можно дополнительно улучшить, чтобы нацелиться на поднаборы данных на шаге (2) и/или использовать дублирующую целевую таблицу для материализованного представления (присоединить партиции к оригиналу, как только вставка завершена) для более легкого восстановления после сбоя.

Рассмотрим следующее материализованное представление, которое вычисляет самые популярные проекты за час.

При добавлении целевой таблицы, перед добавлением материализованного представления, мы модифицируем его SELECT-предложение, чтобы включить фильтр, который учитывает только строки больше определенного времени в ближайшем будущем - в этом случае предполагаем, что 2024-12-17 09:00:00 находится через несколько минут.

После добавления этого представления мы можем заполнить все данные для материализованного представления до этой даты.

Самый простой способ сделать это - просто выполнить запрос из материализованного представления на основной таблице с фильтром, игнорирующим недавно добавленные данные, вставляя результаты в целевую таблицу нашего представления через INSERT INTO SELECT. Например, для вышеуказанного представления:

примечание

В приведенном выше примере наша целевая таблица является SummingMergeTree. В этом случае мы можем просто использовать наш оригинальный запрос агрегации. Для более сложных случаев, которые используют AggregatingMergeTree, пользователи будут использовать функции -State для агрегатов. Пример этого можно найти здесь.

В нашем случае это относительно легкая агрегация, которая завершилась менее чем за 3 секунды и использовала менее 600 MiB памяти. Для более сложных или длительных агрегаций пользователи могут сделать этот процесс более устойчивым, используя ранее описанный подход с дублирующей таблицей, т.е. создать теневую целевую таблицу, например, pypi_downloads_per_day_v2, вставить данные в нее и прикрепить ее результирующие партиции к pypi_downloads_per_day.

Часто запрос материализованного представления может быть более сложным (что не редкость, иначе пользователи не использовали бы представления!) и потреблять ресурсы. В редких случаях ресурсы для запроса превышают мощность сервера. Это подчеркивает одно из преимуществ материализованных представлений ClickHouse - они инкрементальные и не обрабатывают весь набор данных за один раз!

В этом случае у пользователей есть несколько вариантов:

1. Модифицировать ваш запрос для заполнения диапазонов, например, WHERE timestamp BETWEEN 2024-12-17 08:00:00 AND 2024-12-17 09:00:00, WHERE timestamp BETWEEN 2024-12-17 07:00:00 AND 2024-12-17 08:00:00 и т.д.
2. Использовать Null table engine для заполнения материализованного представления. Это реплицирует типичное инкрементальное заполнение материализованного представления, выполняя его запрос по блокам данных (настраиваемого размера).

(1) представляет собой самый простой подход и часто бывает достаточным. Мы не включаем примеры для краткости.

Мы исследуем (2) далее.

Использование Null table engine для заполнения материализованных представлений

Null table engine предоставляет движок хранилища, который не сохраняет данные (можно думать о нем как о /dev/null в мире движков таблиц). Несмотря на то, что это кажется противоречивым, материализованные представления все равно будут выполняться для данных, вставленных в этот движок таблицы. Это позволяет создавать материализованные представления без сохранения оригинальных данных - избегая ввода-вывода и связанного с ним хранения.

Важно, что любые материализованные представления, прикрепленные к движку таблицы, все равно выполняются по блокам данных по мере их вставки - отправляя свои результаты в целевую таблицу. Эти блоки имеют настраиваемый размер. Хотя более крупные блоки могут быть потенциально более эффективными (и быстрее обрабатываемыми), они требуют больше ресурсов (в основном памяти). Использование этого движка таблицы означает, что мы можем строить наше материализованное представление инкрементально, т.е. по одному блоку за раз, избегая необходимости удерживать всю агрегацию в памяти.

Рассмотрим следующий пример:

Здесь мы создаем таблицу Null, pypi_v2, для получения строк, которые будут использоваться для построения нашего материализованного представления. Обратите внимание, как мы ограничиваем схему только теми столбцами, которые нам нужны. Наше материализованное представление выполняет агрегацию над строками, вставленными в эту таблицу (по одному блоку за раз), отправляя результаты в нашу целевую таблицу pypi_downloads_per_day.

примечание

Мы использовали pypi_downloads_per_day в качестве нашей целевой таблицы здесь. Для дополнительной надежности пользователи могут создать дублирующую таблицу, pypi_downloads_per_day_v2, и использовать ее в качестве целевой таблицы представления, как показывалось в предыдущих примерах. По завершении вставки партиции в pypi_downloads_per_day_v2 могут, в свою очередь, быть перемещены в pypi_downloads_per_day. Это обеспечит восстановление в случае сбоя вставки из-за проблем с памятью или прерывания сервера, т.е. нам просто нужно обрезать pypi_downloads_per_day_v2, настроить параметры и повторить попытку.

Чтобы заполнить это материализованное представление, мы просто вставим соответствующие данные для заполнения в pypi_v2 из pypi.

Обратите внимание на наше использование памяти здесь: 639.47 MiB.

Настройка производительности и ресурсов

Несколько факторов будут определять производительность и ресурсы, использованные в вышеописанном сценарии. Мы рекомендуем читателям ознакомиться с механикой вставки, подробно описанной здесь, перед попытками настройки. Вкратце:

• Параллелизм чтения - количество потоков, используемых для чтения. Контролируется через max_threads. В ClickHouse Cloud значение этого параметра определяется размером экземпляра, по умолчанию оно соответствует количеству vCPU. Увеличение этого значения может улучшить производительность чтения за счет больших затрат памяти.
• Параллелизм вставки - количество потоков вставки, используемых для вставки. Контролируется через max_insert_threads. В ClickHouse Cloud это значение зависит от размера экземпляра (от 2 до 4) и установлено на 1 в OSS. Увеличение этого значения может улучшить производительность за счет больших затрат памяти.
• Размер блока вставки - данные обрабатываются в цикле, где они извлекаются, анализируются и формируются в модули вставки в памяти в зависимости от ключа партиционирования. Эти блоки сортируются, оптимизируются, сжимаются и записываются в хранилище в качестве новых частей данных. Размер блока вставки, контролируемый параметрами min_insert_block_size_rows и min_insert_block_size_bytes (несжатый), влияет на использование памяти и ввод-вывод на диске. Более крупные блоки используют больше памяти, но создают меньше частей, уменьшая ввод-вывод и фоновую агрегацию. Эти параметры представляют собой минимальные пороги (первый достигнутый порог приводит к сбросу).
• Размер блока материализованного представления - помимо вышеуказанных механик для основной вставки, перед вставкой в материализованные представления блоки также уменьшаются для более эффективной обработки. Размер этих блоков определяется параметрами min_insert_block_size_bytes_for_materialized_views и min_insert_block_size_rows_for_materialized_views. Более крупные блоки позволяют более эффективную обработку за счет большего потребления ресурсов памяти. По умолчанию эти параметры возвращаются к значениям параметров исходной таблицы min_insert_block_size_rows и min_insert_block_size_bytes соответственно.

Для улучшения производительности пользователи могут следовать рекомендациям, изложенным здесь. В большинстве случаев не должно быть необходимости также изменять min_insert_block_size_bytes_for_materialized_views и min_insert_block_size_rows_for_materialized_views, чтобы улучшить производительность. Если эти значения будут изменены, используйте те же лучшие практики, как обсуждалось для min_insert_block_size_rows и min_insert_block_size_bytes.

Чтобы минимизировать использование памяти, пользователи могут экспериментировать с этими настройками. Это неизбежно снизит производительность. Используя ранее приведенный запрос, мы показываем примеры ниже.

Снижение max_insert_threads до 1 снижает наши накладные расходы по памяти.

Мы можем еще больше снизить использование памяти, установив значение max_threads на 1.

Наконец, мы можем еще больше снизить использование памяти, установив min_insert_block_size_rows на 0 (это отключает его как определяющий фактор для размера блока) и min_insert_block_size_bytes на 10485760 (10 MiB).

Наконец, имейте в виду, что снижение размеров блоков создает больше частей и вызывает большее давление на слияние. Как обсуждалось здесь, эти настройки следует изменять осторожно.

Нет временной метки или колонка, повышающаяся монотонно

Указанные выше процессы зависят от наличия временной метки или монотонно повышающейся колонки у пользователя. В некоторых случаях это просто недоступно. В этом случае мы рекомендуем следующий процесс, который использует многие из ранее упомянутых шагов, но требует от пользователей приостановить загрузку данных.

1. Приостановите вставки в вашу основную таблицу.
2. Создайте дубликат вашей основной целевой таблицы, используя синтаксис CREATE AS.
3. Присоедините партиции из оригинальной целевой таблицы к дубликату, используя ALTER TABLE ATTACH. Примечание: Эта операция присоединения отличается от ранее используемого перемещения. Хотя она опирается на жесткие ссылки, данные в оригинальной таблице сохраняются.
4. Создайте новые материализованные представления.
5. Перезапустите вставки. Примечание: Вставки будут обновлять только целевую таблицу, а не дубликат, который будет ссылаться только на оригинальные данные.
6. Заполните материализованное представление, применяя тот же процесс, который использовался выше для данных с временными метками, используя дублирующую таблицу в качестве источника.

Рассмотрим следующий пример, используя PyPI и наше предыдущее новое материализованное представление pypi_downloads_per_day (предположим, что мы не можем использовать временную метку):

На предпоследнем шаге мы заполняем pypi_downloads_per_day, используя наш простой подход INSERT INTO SELECT, описанный ранее. Это можно также улучшить, используя подход с таблицей Null, документированный выше, с опциональным использованием дублирующей таблицы для большей надежности.

Хотя эта операция требует приостановить вставки, промежуточные операции обычно могут быть выполнены быстро - минимизируя любое прерывание данных.