Шаблон Dataflow для передачи из Pub/Sub в ClickHouse

Шаблон Pub/Sub to ClickHouse — это стриминговый конвейер, который читает сообщения в формате JSON из подписки Pub/Sub и записывает их в таблицу ClickHouse. Сообщения, которые не удалось разобрать или сопоставить с целевой схемой, направляются в пункт назначения dead-letter: таблицу ClickHouse, тему Pub/Sub или в оба сразу.

• Исходная подписка Pub/Sub должна существовать.
• Сообщения, публикуемые в подписку, должны быть корректным JSON.
• Целевая таблица ClickHouse должна существовать, а имена её столбцов должны совпадать с именами полей в полезной нагрузке JSON.
• Хост ClickHouse должен быть доступен с машин воркеров Dataflow.
• Должен быть указан как минимум один пункт назначения dead-letter (clickHouseDeadLetterTable или deadLetterTopic). Если указаны оба, сообщения с ошибками направляются в оба пункта назначения одновременно.
• Если задан clickHouseDeadLetterTable, таблица dead-letter уже должна существовать в ClickHouse со схемой, показанной в разделе Обработка dead-letter.
• Если задан deadLetterTopic, топик Pub/Sub уже должен существовать.

Сообщения Pub/Sub должны быть объектами JSON, в которых имена полей верхнего уровня в точности совпадают с именами столбцов целевой таблицы ClickHouse.

Чтобы сопоставить входящие сообщения с целевой таблицей, конвейер при запуске выполняет следующие действия:

1. Получает схему целевой таблицы ClickHouse.
2. Создает схему Beam Row на основе схемы ClickHouse.
3. Для каждого входящего сообщения Pub/Sub разбирает полезную нагрузку JSON и формирует строку, считывая поля с именами из схемы ClickHouse.

Значения JSON приводятся к соответствующему типу столбца ClickHouse:

Поскольку конвейер работает в режиме стриминга, входящие строки накапливаются в окнах перед записью в ClickHouse. Стратегия оконной обработки выбирается на основе указанных вами параметров:

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

Сообщения, для которых не удалось выполнить парсинг JSON, сопоставление со схемой или приведение типов, направляются в настроенные пункты назначения dead-letter. Необходимо указать как минимум один из параметров: clickHouseDeadLetterTable или deadLetterTopic; если заданы оба, сообщения с ошибками будут отправлены в оба.

Если задан параметр clickHouseDeadLetterTable, таблица dead-letter уже должна существовать со следующей фиксированной схемой:

Минимальное определение для одновузлового развертывания:

CREATE TABLE my_table_dead_letter (
    raw_message   String,
    error_message String,
    stack_trace   String,
    failed_at     DateTime
) ENGINE = MergeTree()
ORDER BY failed_at;

Если задан deadLetterTopic, каждое сообщение, обработка которого завершилась ошибкой, повторно публикуется в топик со следующим содержимым:

• Полезная нагрузка: исходные байты сообщения.
• Атрибут errorMessage: сообщение исключения, зафиксированное в момент сбоя.
• Атрибут failedAt: временная метка времени обработки, соответствующая моменту сбоя строки.

Это позволяет удобно повторно обработать сообщения с ошибками после устранения проблемы в схеме или у продьюсера.

Шаблон Pub/Sub to ClickHouse доступен в Google Cloud Console.

Войдите в Google Cloud Console и найдите Dataflow.

1. Нажмите кнопку CREATE JOB FROM TEMPLATE.

   

2. Когда откроется форма шаблона, введите имя задачи и выберите нужный регион.

3. В поле Dataflow Template введите ClickHouse или Pub/Sub и выберите шаблон Pub/Sub to ClickHouse.

4. После выбора форма развернётся. Заполните:

   • входную подписку Pub/Sub в формате projects/<PROJECT_ID>/subscriptions/<SUBSCRIPTION_NAME>.
   • URL конечной точки ClickHouse — для ClickHouse Cloud используйте https://<HOST>:8443.
   • базу данных ClickHouse, целевую таблицу, имя пользователя и пароль.
   • как минимум один пункт назначения dead-letter: таблицу ClickHouse или топик Pub/Sub (или оба варианта).

5. При необходимости настройте батчинг (windowSeconds, batchRowCount) и параметры тонкой настройки ClickHouseIO, как подробно описано в разделе Параметры шаблона.

Перейдите на вкладку задач Dataflow в Google Cloud Console, чтобы отслеживать состояние задачи. Там вы найдете сведения о задаче, включая ход выполнения и возможные ошибки:

Шаблон также отправляет следующие пользовательские метрики в пространстве имен PubSubToClickHouse; их можно просмотреть на странице задачи Dataflow:

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

• Увеличьте ресурсы инстанса: переведите ClickHouse server на более крупный инстанс с большим объёмом памяти, чтобы он справлялся с нагрузкой при обработке данных.
• Уменьшите размер батча: сократите batchRowCount (и/или maxInsertBlockSize) в конфигурации задачи Dataflow, чтобы отправлять в ClickHouse меньшие фрагменты данных и снизить потребление памяти на батч.

Наиболее распространённые причины:

• Имена JSON-полей не совпадают в точности с именами столбцов ClickHouse (сопоставление чувствительно к регистру).
• Значение JSON невозможно привести к типу столбца (например, строку не в формате ISO-8601 в столбце DateTime).
• Схема целевой таблицы изменилась после запуска конвейера — схема загружается один раз при запуске. Перезапустите задачу после внесения изменений в схему.

Проверьте столбцы error_message и stack_trace в таблице dead-letter ClickHouse (или атрибут errorMessage в сообщениях Pub/Sub dead-letter), чтобы определить первопричину.

• Убедитесь, что подписка получает сообщения — проверьте метрику messages-received на странице задачи Dataflow.
• В режиме по времени (только windowSeconds) строки сбрасываются на диск только на границах окна. Уменьшите windowSeconds, чтобы проверить, происходят ли сбросы.
• Проверьте сетевую доступность между воркерами Dataflow и конечной точкой ClickHouse (брандмауэр, пиринг VPC или Private Service Connect).

Исходный код шаблона доступен в следующих репозиториях:

• GoogleCloudPlatform/DataflowTemplates — основной репозиторий Google Cloud Platform.
• ClickHouse/DataflowTemplates — форк ClickHouse.