Интеграция Google Cloud Pub/Sub с ClickHouse Cloud

Легко подключайте топики Google Cloud Pub/Sub к ClickHouse Cloud.

Вы можете записаться в лист ожидания закрытой предварительной версии здесь.

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

​

Предварительные требования

Вы уже ознакомились с введением в ClickPipes, имеете доступ к проекту GCP, содержащему топик, из которого нужно настроить приём данных, и создали сервисный аккаунт с соответствующими разрешениями Pub/Sub. Точный набор разрешений, необходимых ClickPipes, см. в руководстве по разрешениям IAM для Pub/Sub.

​

Создание первого ClickPipe

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

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

3. Выберите GCP Pub/Sub в качестве источника данных.

4. Заполните форму: укажите имя ClickPipe, ваш GCP Project ID и JSON‑файл сервисного аккаунта для сервисного аккаунта, которому выдан доступ к Pub/Sub. Project ID должен содержать 6–30 символов, может включать строчные буквы, цифры и дефисы, должен начинаться с буквы и не может заканчиваться дефисом.

5. Выберите Pub/Sub topic, из которого нужно получать данные. После успешной проверки учётных данных раскрывающийся список автоматически заполняется топиками из вашего проекта GCP (в алфавитном порядке).

   • Формат данных. При выборе топика ClickPipes обращается к реестру схем Pub/Sub. Если к топику привязана нативная схема Avro или Protobuf, Data format и Schema определяются автоматически, а селекторы фиксируются на последней схеме топика. Для топиков без нативной схемы по умолчанию используется JSONEachRow.
   • Начальное смещение. Выберите, с какого места начинать чтение. Доступны варианты Latest (только новые сообщения), Earliest (самые старые сохранённые сообщения) и Seek to Timestamp (с выбором даты и времени в UTC).
   • Filter expression (необязательно). Фильтр subscription filter Pub/Sub по атрибутам сообщений — например, attributes.type = "telemetry". Фильтры применяются только к атрибутам сообщений, а не к полезной нагрузке, и не могут быть изменены после создания пайпа (для изменения фильтра пайп нужно пересоздать).
   • В интерфейсе будет показан пример сообщения из выбранного топика, а также переключатель Flatten object, с помощью которого можно предварительно посмотреть, как вложенный JSON будет развёрнут на стороне назначения.

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

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

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

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

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

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

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

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

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

10. Поздравляем! Вы успешно настроили свой первый Pub/Sub ClickPipe. Он будет непрерывно работать, выполняя ингестию данных в реальном времени из вашего топика Pub/Sub в ваш сервис ClickHouse Cloud.

​

Управляемые подписки

Сообщения Pub/Sub потребляются через подписки, а не напрямую из топиков. ClickPipes создает и управляет отдельной подпиской для каждого пайпа — вам достаточно выбрать только топик.

• Управляемая подписка называется clickpipes-{pipeID} и создается в топике при запуске пайпа.
• Для нее задаются срок подтверждения 60 секунд, хранение сообщений в течение 7 дней и включается упорядочение сообщений.
• Создание подписки идемпотентно — при перезапуске пайпа и перепланировании реплики используется существующая подписка, если она уже привязана к настроенному топику.
• Во время обнаружения топиков и сэмплирования сообщений ClickPipes также создает кратковременные эфемерные подписки (clickpipes-discovery-{uuid}), которые удаляются сразу после завершения сэмплирования.
• При удалении пайпа ClickPipes удаляет управляемую подписку в рамках очистки ресурсов.

Поэтому предоставляемый вами сервисный аккаунт должен иметь разрешение на создание и удаление подписок в проекте, помимо их использования для получения сообщений. Полный список см. в руководстве по разрешениям IAM для Pub/Sub.

​

Поддерживаемые форматы данных

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

• JSON
• Avro — через встроенные схемы Pub/Sub (кодирование BINARY)
• Protobuf — через встроенные схемы Pub/Sub (кодирование BINARY)

Для Avro и Protobuf схема берётся из реестра схем Pub/Sub для топика. Пайп всегда использует последнюю ревизию схемы топика; селектор схемы в интерфейсе сделан только для чтения.

​

Сжатие

ClickPipes для Pub/Sub автоматически определяет и распаковывает сжатые сообщения. Клиент Pub/Sub передаёт сырые байты — ClickPipes сам выполняет распаковку, без дополнительной настройки.

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

• gzip
• zstd
• lz4
• snappy (framed format)

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

Автоопределение безопасно для текстовых форматов, таких как JSON, поскольку печатаемые символы ASCII не могут совпасть с сигнатурами magic bytes сжатия. Размер распакованной полезной нагрузки ограничен 64 МБ.

​

Поддерживаемые типы данных

​

Поддержка стандартных типов

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

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

​

Поддержка типа Variant

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

​

Поддержка типа JSON

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

​

Виртуальные столбцы Pub/Sub

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

Имя	Описание	Рекомендуемый тип данных
_message_id	Идентификатор сообщения Pub/Sub, назначенный брокером	String
_publish_time	Временная метка публикации Pub/Sub (с точностью до миллисекунд, UTC)	DateTime64(3)
_ordering_key	Ключ упорядочивания Pub/Sub (пустая строка, если для сообщения ключ не задан)	String
_attributes	Пользовательские атрибуты сообщения Pub/Sub	Map(String, String)
_raw_message	Полная полезная нагрузка сообщения Pub/Sub (по умолчанию отключена)	String

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

​

Ограничения

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

​

Производительность

​

Формирование батчей

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

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

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

​

Задержка

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

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

​

Ключи упорядочивания

Pub/Sub гарантирует, что сообщения с одним и тем же ключом упорядочивания доставляются одному подписчику в порядке публикации. ClickPipes по умолчанию включает упорядочивание для своих управляемых подписок: если сообщения содержат ключи упорядочивания, подписчики получают их по порядку; если сообщения не содержат ключей упорядочивания, поведение не меняется.

Если ваш продюсер публикует все сообщения с небольшим количеством ключей упорядочивания (или с одним ключом), Pub/Sub будет направлять эти сообщения небольшому числу подписчиков, что может ограничить горизонтальную пропускную способность. Мы рекомендуем либо не использовать ключи упорядочивания, если упорядочивание не требуется, либо использовать ключ упорядочивания с высокой мощностью.

​

Масштабирование

ClickPipes для Pub/Sub поддерживает как горизонтальное, так и вертикальное масштабирование. Каждый пайп использует одну управляемую подписку Pub/Sub — изменить это нельзя. По умолчанию сообщения из этой подписки получает один консьюмер; вы можете увеличить число консьюмеров при создании ClickPipe или в любой другой момент в разделе Settings -> Advanced Settings -> Scaling. ClickPipes автоматически распределяет сообщения из подписки между запущенными консьюмерами — дополнительная координация не требуется.

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

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

​

Семантика доставки

ClickPipes для Pub/Sub обеспечивает доставку at-least-once. Сообщение Pub/Sub подтверждается только после того, как соответствующая строка была вставлена в ClickHouse (или записана в таблицу ошибок в случае некорректных записей); после обработки подтверждаются все сообщения — включая ошибочные записи, направленные в таблицу ошибок, — чтобы предотвратить бесконечную повторную доставку. Если реплика аварийно завершится после вставки, но до того, как подтверждение дойдёт до Pub/Sub, сообщение будет повторно доставлено по истечении срока подтверждения и вставлено снова, поэтому нижестоящие потребители должны быть устойчивы к дубликатам. Если вам нужна семантика «ровно один раз», выполняйте дедупликацию на стороне потребителя с помощью виртуального столбца _message_id (каждый идентификатор сообщения Pub/Sub уникален в пределах topic).

​

Аутентификация

ClickPipes для Pub/Sub аутентифицируется в GCP с помощью JSON-ключа сервисного аккаунта. Вы загружаете файл ключа при создании пайпа; ClickPipes шифрует его при хранении и использует во время работы для получения сообщений и управления жизненным циклом управляемой подписки.

Точный список требуемых разрешений IAM и описание рекомендуемой пользовательской роли см. в руководстве по разрешениям IAM для Pub/Sub.