Интеграция Amazon Kinesis с ClickHouse Cloud

Kinesis ClickPipes можно развертывать и настраивать вручную через интерфейс ClickPipes, а также программно с помощью OpenAPI и Terraform.

Вы уже ознакомились с введением в ClickPipes и настроили учетные данные IAM или роль IAM. Сведения о том, как настроить роль для работы с ClickHouse Cloud, см. в руководстве по ролевому доступу Kinesis.

1. Откройте SQL Console для вашего сервиса ClickHouse Cloud.

2. В левом меню нажмите кнопку Data Sources, затем — "Set up a ClickPipe"

3. Выберите источник данных.

4. Заполните форму: укажите имя ClickPipe, описание (необязательно), роль IAM или учётные данные, а также другие сведения о подключении.

5. Выберите Kinesis Stream и начальное смещение. В интерфейсе отобразится образец документа из выбранного источника (топика Kafka и т. д.). Вы также можете включить Enhanced Fan-out для потоков Kinesis, чтобы повысить производительность и стабильность ClickPipe (подробнее об Enhanced Fan-out читайте здесь)

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

Вы также можете изменить расширенные настройки с помощью доступных элементов управления

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

8. Наконец, вы можете настроить разрешения для внутреннего пользователя ClickPipes.

Разрешения: ClickPipes создаст отдельного пользователя для записи данных в целевую таблицу. Для этого внутреннего пользователя можно выбрать пользовательскую роль или одну из предопределённых ролей:

• Full access: полный доступ к кластеру. Это может быть полезно, если вы используете materialized view или словарь вместе с целевой таблицей.
  • Only destination table: только разрешения INSERT для целевой таблицы.

9. Нажмите "Complete Setup", чтобы система зарегистрировала ClickPipe, после чего он появится в сводной таблице.

В сводной таблице есть элементы управления для просмотра примеров данных из источника или из целевой таблицы в ClickHouse

А также элементы управления для удаления ClickPipe и просмотра сводки задания приёма.

10. Поздравляем! Вы успешно настроили свой первый ClickPipe. Если это стриминговый ClickPipe, он будет непрерывно работать и выполнять ингестию данных в реальном времени из удалённого источника. В противном случае он обработает батч и завершит работу.

Поддерживаются следующие форматы:

• JSON

ClickPipes для Kinesis автоматически определяет и распаковывает сжатые записи. В отличие от Kafka, где декомпрессия прозрачно выполняется клиентской библиотекой, Kinesis передаёт сырые байты — ClickPipes берёт это на себя без дополнительной настройки.

Поддерживаются следующие кодеки сжатия:

• gzip
• zstd
• lz4
• snappy (фреймированный формат)

Сжатие автоматически определяется по сигнатурам magic bytes в каждой записи. Если известная сигнатура сжатия не найдена, запись считается несжатой. Обнаруженный тип сжатия также определяется при автоопределении схемы, поэтому в предпросмотре данных в интерфейсе корректно отображаются распакованные данные.

В настоящее время ClickPipes поддерживает следующие типы данных ClickHouse:

• Базовые числовые типы — [U]Int8/16/32/64, Float32/64 и BFloat16
• Целочисленные типы большой разрядности — [U]Int128/256
• Типы Decimal
• Boolean
• String
• FixedString
• Date, Date32
• DateTime, DateTime64 (только часовой пояс UTC)
• Enum8/Enum16
• UUID
• IPv4
• IPv6
• все типы ClickHouse LowCardinality
• Map с ключами и значениями, использующими любой из перечисленных выше типов (включая Nullable)
• Tuple и Array с элементами, использующими любой из перечисленных выше типов (включая Nullable; только один уровень вложенности)
• типы SimpleAggregateFunction (для пунктов назначения AggregatingMergeTree или SummingMergeTree)

Вы можете вручную указать тип варианта (например, Variant(String, Int64, DateTime)) для любого JSON-поля в потоке исходных данных. Поскольку ClickPipes определяет, какой подтип варианта следует использовать, особым образом, в определении Variant можно использовать только один целочисленный тип или один тип datetime — например, Variant(Int64, UInt32) не поддерживается.

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

Для потока Kinesis поддерживаются следующие виртуальные столбцы. При создании новой целевой таблицы их можно добавить с помощью кнопки Add Column.

Поле _raw_message можно использовать в случаях, когда нужна только полная JSON-запись Kinesis (например, при использовании функций ClickHouse JsonExtract* для заполнения материализованного представления ниже по конвейеру). Для таких пайпов удаление всех «не виртуальных» столбцов может повысить производительность ClickPipes.

• DEFAULT не поддерживается.
• По умолчанию размер отдельных сообщений ограничен 16 МБ (без сжатия) для реплик минимального размера (XS) и 32 МБ (без сжатия) для более крупных реплик. Сообщения, превышающие этот лимит, будут отклонены с ошибкой. Если вам требуются сообщения большего размера, пожалуйста, обратитесь в службу поддержки.

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

Батчи вставляются при выполнении одного из следующих условий:

• Размер батча достиг максимального значения (100 000 строк или 32 МБ на 1 ГБ памяти реплики)
• Батч оставался открытым в течение максимально допустимого времени (5 секунд)

Задержка (то есть время между отправкой сообщения Kinesis в поток и моментом, когда это сообщение становится доступным в ClickHouse) зависит от ряда факторов (например, задержки в Kinesis, сетевой задержки, размера/формата сообщения). Описанный в предыдущем разделе батчинг также влияет на задержку. Мы всегда рекомендуем протестировать ваш конкретный сценарий использования, чтобы понять, на какую задержку можно рассчитывать.

Если у вас есть особые требования к низкой задержке, свяжитесь с нами.

Мы настоятельно рекомендуем ограничивать число одновременно активных сегментов в соответствии с требованиями к пропускной способности. Для потока Kinesis типа "On Demand" AWS автоматически назначает соответствующее количество сегментов на основе пропускной способности, но для потоков "Provisioned" выделение слишком большого количества сегментов может вызывать задержки, как описано ниже, а также увеличивать затраты, поскольку для таких потоков тарификация Kinesis рассчитывается "за сегмент".

Если приложение-производитель непрерывно записывает данные в большое число активных сегментов, это может вызывать задержки, если ваш пайп масштабирован недостаточно для их эффективной обработки. С учётом ограничений Kinesis по пропускной способности ClickPipes назначает определённое количество "воркеров" на реплику для чтения данных из сегментов. Например, в минимальной конфигурации реплика ClickPipes имеет 4 таких воркера. Если производитель записывает одновременно более чем в 4 сегмента, данные из "лишних" сегментов не будут обрабатываться, пока не освободится воркер. В частности, если пайп использует "enhanced fanout", каждый воркер будет подписан на один сегмент в течение 5 минут и в это время не сможет читать данные из других сегментов. Это может вызывать "всплески" задержки с интервалом в 5 минут.

ClickPipes для Kinesis спроектирован так, чтобы масштабироваться как по горизонтали, так и по вертикали. По умолчанию создается группа потребителей с одним потребителем. Это можно настроить при создании ClickPipe или позже в разделе Настройки -> Дополнительные настройки -> Масштабирование.

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

Независимо от количества запущенных потребителей отказоустойчивость обеспечивается самой архитектурой. Если потребитель или лежащая в его основе инфраструктура выйдет из строя, ClickPipe автоматически перезапустит потребителя и продолжит обработку сообщений.

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