Skip to content

Возможности

Чтение Kafka -> Greenplum

SELECT из foreign table читают все сегменты параллельно. Партиции топика делятся между сегментами по остатку от деления номера партиции на число сегментов, без Kafka rebalancing: назначение статическое и повторяемое, одна и та же партиция всегда достаётся одному и тому же сегменту. Координатор в чтении не участвует - он строит план и собирает результат.

  • Формат сообщения - JSON или CSV (RFC 4180, разделители настраиваются).
  • Условия по __partition и __offset уходят в чтение как границы: коннектор подписывается только на нужные партиции и стартует с нужного офсета, а не фильтрует лишнее после вычитки.
  • end_offset 'latest' фиксирует high watermark на старте скана и завершает чтение сразу после последнего сообщения диапазона. Для ETL-загрузок это рабочий режим: запрос не ждёт batch_timeout_ms впустую.
  • batch_size задаёт размер порции, которую сегмент запрашивает у librdkafka за раз, и на объём результата не влияет. До 1.5 опция обрезала выдачу до batch_size строк на сегмент, и этого никто не ожидал.
  • Перебои брокеров (смена лидера партиции, разрыв соединения, переезд координатора) скан переживает: librdkafka переподключается сам, в лог уходит WARNING. Запрос падает, только если к batch_timeout_ms конец диапазона так и не достигнут - частичный результат коннектор не отдаёт.
  • Метаданные сообщения доступны через системные колонки __offset, __partition, __timestamp, __key, __topic - их достаточно добавить в определение таблицы.

Запись Greenplum -> Kafka

INSERT тоже идёт со всех сегментов: каждый пишет свои строки напрямую в брокеры через собственный продюсер. Строка сериализуется в JSON или CSV, партицию выбирает Kafka - либо по ключу из колонки __key, либо явным номером в __partition, либо своим алгоритмом, если ни того ни другого нет. Перед завершением INSERT коннектор дожидается подтверждения доставки всех сообщений; если за 60 секунд подтвердились не все, INSERT падает, а не делает вид, что всё записано. Когда локальная очередь продюсера заполняется быстрее, чем брокеры отвечают, коннектор ждёт освобождения места, так что большие вставки (сотни тысяч строк на сегмент) проходят целиком.

Инкрементальная загрузка

Офсеты хранятся в Kafka, в __consumer_offsets, под group.id, равным опции consumer_group. Каждый успешный запрос коммитит позиции своих партиций, следующий продолжает с них. Партиции без закоммиченного офсета читаются с start_offset.

Коммит двухфазный. Сегмент, дочитав партиции, сразу записывает новую позицию, но с пометкой «незавершено»; снимает пометку координатор при фиксации транзакции, то есть только когда весь запрос удался. Если INSERT ... SELECT упал на стороне вставки, другой сегмент выдал ошибку или вы сделали ROLLBACK, пометка остаётся, и следующий запрос начнёт с прежней позиции. Строки перечитаются, но не пропадут. До 1.5 сегмент коммитил окончательно, и упавший запрос мог навсегда потерять его строки.

Если закоммиченная позиция оказалась ниже начала retained-лога (retention удалил сообщения раньше, чем их успели загрузить), коннектор продолжит с первого доступного сообщения и напишет в WARNING, сколько потеряно. Молчаливого прыжка на конец партиции, который librdkafka делает по умолчанию, больше нет.

Отсюда два практических следствия. Прогресс загрузки видно обычными инструментами Kafka, снаружи Greenplum. И изменение числа сегментов кластера не требует сброса офсетов: они привязаны к партициям, поэтому новый владелец партиции продолжит с той же позиции.

Управление офсетами из SQL

До 1.3 офсеты лежали в таблице Greenplum, и подвинуть их мог любой UPDATE. С переездом в Kafka эта возможность пропала, в 1.5 она вернулась в виде функций:

  • kafka.set_offset(server, group, topic, partition, offset) - выставить позицию одной партиции;
  • kafka.reset_offsets(server, group, topic, 'earliest' | 'latest' | '<timestamp>') - перевести все партиции топика на начало retained-лога, на его конец или на первое сообщение не раньше указанного момента.

Офсет здесь - номер следующего сообщения для чтения, как committed_offset в kafka.offsets_info(). Значения за пределами [low_watermark, high_watermark] функции отвергают. Типовой сценарий - переключение на другой Kafka-контур с реплицированным топиком: сменили brokers у сервера, выставили позиции, ETL продолжает работу без доступа к Kafka CLI.

Обработка битых сообщений

Продюсеры иногда шлют не JSON: незакавыченные строки, обрезанные сообщения. Поведение задаёт опция error_action:

  • fail (по умолчанию) - запрос падает с ошибкой, в которой указаны топик, партиция и офсет сообщения. Испорченную запись легко найти в Kafka и посмотреть глазами.
  • skip - сообщение пропускается с WARNING в лог, скан продолжается, офсет пропущенного засчитывается как прочитанный. Число пропусков ограничивает error_limit.

Ошибка преобразования типа - это другое. Если JSON валиден, а значение не влезает в тип колонки (строка в int, 29-значный идентификатор в bigint), запрос прерывается всегда, независимо от error_action. Такие вещи лечатся типом колонки, а не пропуском данных.

Вложенные структуры

Значение ключа может быть вложенным объектом или массивом. Объявите колонку как jsonb (или json, text) - коннектор положит туда сырой JSON-текст структуры целиком, дальше работают штатные операторы Greenplum. Типы-массивы (jsonb[], text[]) для этого не годятся: [{…},{…}] - литерал JSON-массива, а не массива PostgreSQL.

В колонку jsonb или json любое значение попадает как JSON-литерал: строка "abc" остаётся jsonb-строкой с кавычками, пустая строка "" - тоже, число - числом. В 1.4 строка сначала раскавычивалась, и {"message": ""} ломал запрос ошибкой invalid input syntax for type json.

Безопасность

Поддерживаются PLAINTEXT, SSL, SASL_PLAINTEXT и SASL_SSL. Из механизмов SASL - PLAIN, SCRAM-SHA-256, SCRAM-SHA-512, GSSAPI (Kerberos). Для mTLS задаются клиентский сертификат и ключ. Учётные данные лучше держать в USER MAPPING, а не в опциях сервера: доступ к ним ограничен владельцем маппинга. Пароли маскируются в логах и затираются в памяти при освобождении.

Мониторинг

SQL-функции в схеме kafka читают состояние вживую из Kafka:

  • kafka.topic_info() - топики, партиции, лидеры, число реплик и ISR;
  • kafka.consumer_lag() - закоммиченный офсет, high watermark и отставание по каждой партиции;
  • kafka.offsets_info() - то же плюс last_loaded_offset, то есть офсет последнего сообщения, которое реально попало в Greenplum;
  • kafka.version() - версия загруженной библиотеки на том хосте, где выполняется вызов. Через gp_dist_random('gp_id') она показывает, доехала ли новая .so до каждого сегмента.

Ограничения версии 1.5.0

  • Только INSERT и SELECT. UPDATE и DELETE невозможны: Kafka - append-only лог.
  • Чтение даёт at-least-once. Прерванный или откаченный запрос не продвигает офсеты, поэтому данные будут перечитаны - дубликаты возможны, потери нет.
  • Запись даёт at-least-once. Прерванный INSERT откатывается в Greenplum, но то, что Kafka уже приняла до отката, остаётся в топике: транзакций у продюсера нет, Kafka Transactions API не поддерживается. Повторную заливку планируйте идемпотентной.
  • Форматы - JSON и CSV. Avro, Protobuf и Schema Registry не поддерживаются.
  • Расширение написано под Greenplum 6 (PostgreSQL 9.4), совместимость с GP7 не гарантируется.
  • Чтение и запись идут только на сегментах. Таблица без mpp_execute 'all segments' (или с 'master') при первом же SELECT или INSERT падает с подсказкой, что добавить. До 1.5 такая таблица молча возвращала ноль строк.
  • JDBC-клиенты (DBeaver и другие) требуют одной настройки подключения из-за ограничения ядра GP6 - см. Использование.

Разбор ограничений с примерами - на страницах Использование и Конфигурация.