Использование

Ниже - весь путь от создания сервера до инкрементальной загрузки и записи, одним сквозным примером. Адреса брокеров и учётные данные подставьте свои.

1. Расширение

Один раз на базу данных, после установки коннектора на кластер (см. Установка):

CREATE EXTENSION kafka_fdw;

2. Сервер

SERVER описывает, где находится Kafka:

CREATE SERVER kafka_server
    FOREIGN DATA WRAPPER kafka_fdw
    OPTIONS (brokers 'broker1:9092,broker2:9092,broker3:9092');

Достаточно перечислить пару брокеров: остальных коннектор найдёт по метаданным кластера.

3. Учётные данные

Без аутентификации маппинг всё равно нужен, пусть и пустой:

CREATE USER MAPPING FOR CURRENT_USER SERVER kafka_server;

С SASL - с логином и паролем, см. Конфигурация.

4. Внешняя таблица

CREATE FOREIGN TABLE events (
    event_id    int,
    event_type  text,
    payload     text,
    created_at  timestamp
)
    SERVER kafka_server
    OPTIONS (topic 'events', format 'json', mpp_execute 'all segments');

Имена колонок должны совпадать с ключами JSON. Ключи, которым нет колонки, игнорируются; колонки, которым нет ключа, получают NULL.

mpp_execute 'all segments' обязателен: читают и пишут только сегменты, координатору партиции не достаются. С 1.5 таблица без этой опции не вернёт тихие ноль строк, а упадёт при первом обращении:

ERROR:  kafka_fdw: foreign table "events" must have mpp_execute 'all segments'
HINT:  ALTER FOREIGN TABLE events OPTIONS (ADD mpp_execute 'all segments');

Проверка срабатывает при SELECT или INSERT, а не при CREATE: ядро Greenplum не показывает эту опцию коннектору на этапе создания таблицы.

5. Чтение

SELECT * FROM events;

SELECT event_type, count(*) FROM events GROUP BY 1;

Каждый сегмент читает свои партиции параллельно с остальными. Партиция достаётся сегменту по остатку от деления её номера на число сегментов, поэтому топик из 12 партиций на кластере из 4 сегментов разложится ровно по три. Если партиций меньше, чем сегментов, часть сегментов не получит данных и завершит скан мгновенно - для равномерной нагрузки заводите топики с числом партиций, кратным числу сегментов.

Сколько партиций у топика, покажет мониторинг:

SELECT * FROM kafka.topic_info('kafka_server') WHERE topic = 'events';

Ограничение чтения

Без offset_storage каждый SELECT начинает с start_offset и читает, пока не истечёт batch_timeout_ms. Это удобно для разведки, но для загрузки нужен либо снимок по high watermark, либо офсеты (следующий раздел).

batch_size объём результата не ограничивает: это размер порции, которую сегмент берёт у librdkafka за один вызов, и трогать его обычно незачем. В 1.4 опция работала как лимит строк на сегмент, и batch_size '5' на двух сегментах давал 10 строк из 11. Если лимит на запрос всё же нужен, используйте WHERE __offset < ….

-- прочитать всё, что накопилось на момент старта запроса, и сразу завершиться
CREATE FOREIGN TABLE events_snapshot ( … )
    SERVER kafka_server
    OPTIONS (topic 'events', format 'json',
             end_offset 'latest', mpp_execute 'all segments');

end_offset 'latest' фиксирует high watermark в начале скана и завершает чтение сразу после последнего сообщения диапазона, не досиживая batch_timeout_ms.

Фильтры по партиции и офсету

Условия по системным колонкам __partition и __offset коннектор превращает в границы чтения, а не в фильтр после вычитки:

SELECT * FROM events WHERE __partition = 0;
SELECT * FROM events WHERE __partition IN (0, 2, 4);
SELECT * FROM events WHERE __offset >= 1000 AND __offset < 2000;

Фильтры по обычным колонкам работают как везде: данные читаются из Kafka, а отбор делает Greenplum.

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

Чтобы каждый запрос продолжал с того места, где остановился предыдущий, задайте группу и хранение офсетов:

CREATE FOREIGN TABLE events_tracked (
    event_id    int,
    event_type  text,
    payload     text,
    created_at  timestamp,
    __offset    bigint,
    __partition int
)
    SERVER kafka_server
    OPTIONS (
        topic          'events',
        format         'json',
        consumer_group 'etl_pipeline',
        offset_storage 'kafka',
        end_offset     'latest',
        mpp_execute    'all segments'
    );

INSERT INTO stage.events SELECT * FROM events_tracked;

Как это работает. Первый запрос читает с start_offset. Каждый сегмент, дочитав свои партиции, записывает новые позиции в Kafka под group.id = 'etl_pipeline', но с пометкой «незавершено». Когда транзакция фиксируется, координатор снимает пометку. Следующий запрос спрашивает у Kafka закоммиченные позиции и продолжает с них; партиции, по которым группа ещё ничего не коммитила, читаются с start_offset.

Пометка нужна, чтобы упавший запрос не терял данные. Сегмент заканчивает чтение раньше, чем известен исход всего запроса: вставка в stage.events может упасть на ограничении таблицы, другой сегмент может выдать ошибку, вы можете сделать ROLLBACK. Во всех этих случаях пометка остаётся, и следующий запрос начнёт с прежней позиции - строки перечитаются. До 1.5 сегмент коммитил окончательно, и такие строки пропадали.

Если загрузка отстала от retention и закоммиченная позиция оказалась ниже первого сохранившегося сообщения, запрос продолжит с него и напишет в WARNING, сколько сообщений потеряно: committed offset 1200 ... is below the low watermark 4800: 3600 messages were deleted by retention. Это сигнал увеличить retention или загружать чаще; восстановить удалённое уже нельзя.

Комбинация offset_storage 'kafka' + end_offset 'latest' - рабочий вариант для регулярной загрузки: запрос забирает ровно то, что накопилось с прошлого раза, и завершается, не ожидая новых сообщений.

Прогресс видно и из SQL, и из Kafka:

SELECT * FROM kafka.offsets_info('kafka_server', 'etl_pipeline', 'events');

consumer_group	topic	partition_num	last_loaded_offset	committed_offset	high_watermark	lag
etl_pipeline	events	0	1522	1523	2100	577
etl_pipeline	events	1	1486	1487	1900	413

Сдвинуть или сбросить офсеты

С 1.5 позиции группы меняются из SQL. Запросы к таблице с той же группой в этот момент выполняться не должны:

-- перечитать всё заново: каждую партицию на начало retained-лога
SELECT * FROM kafka.reset_offsets('kafka_server', 'etl_pipeline', 'events', 'earliest');

-- пропустить накопившееся: каждую партицию на high watermark
SELECT * FROM kafka.reset_offsets('kafka_server', 'etl_pipeline', 'events', 'latest');

-- по времени: на первое сообщение не раньше момента
SELECT * FROM kafka.reset_offsets('kafka_server', 'etl_pipeline', 'events', '2026-10-01 00:00:00+03');

-- одну партицию на конкретную позицию (следующий запрос прочитает партицию 0 с офсета 5000)
SELECT kafka.set_offset('kafka_server', 'etl_pipeline', 'events', 0, 5000);

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

Тот же результат даёт Kafka CLI (kafka-consumer-groups.sh --reset-offsets) или смена группы - новая начнёт с start_offset:

ALTER FOREIGN TABLE events_tracked OPTIONS (SET consumer_group 'etl_pipeline_v2');

Изменение числа сегментов кластера сбрасывать офсеты не требует: они привязаны к партициям, и новый владелец партиции продолжит с её позиции.

7. Битые сообщения

По умолчанию структурно повреждённое сообщение роняет запрос с точными координатами:

ERROR:  kafka_fdw: malformed json message in topic "events", partition 0, offset 5

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

CREATE FOREIGN TABLE events_tolerant ( … )
    SERVER kafka_server
    OPTIONS (topic 'events', format 'json',
             error_action 'skip',
             error_limit  '100',
             mpp_execute  'all segments');

Пропущенное сообщение уходит в лог как WARNING, его офсет засчитывается прочитанным - повторно оно не появится. error_limit считается на каждом сегменте отдельно: при лимите 100 и четырёх сегментах запрос переживёт до 400 пропусков.

Ошибка преобразования типа пропуском не лечится. Строка "abc" в колонке int или 29-значный идентификатор в bigint прервут запрос при любом error_action. Чините тип колонки: для больших целых - numeric.

8. Вложенные объекты и массивы

Продюсер шлёт что-то такое:

{"message": [{"a": 1, "b": "x y"}, {"a": 2, "b": "z"}], "message_id": 10}

Объявите колонку как jsonb - коннектор положит туда сырой JSON-текст структуры целиком:

CREATE FOREIGN TABLE inbox (
    message     jsonb,
    message_id  numeric
)
    SERVER kafka_server
    OPTIONS (topic 'inbox', format 'json',
             end_offset 'latest', mpp_execute 'all segments');

SELECT message_id,
       jsonb_array_length(message) AS items,
       message -> 0 ->> 'b'        AS first_b
FROM inbox;

Типы-массивы Greenplum (jsonb[], text[]) для этого не годятся: [{…},{…}] - литерал JSON-массива, а у массива PostgreSQL синтаксис {…}, его входная функция такое отвергнет.

9. Запись

INSERT INTO events (event_id, event_type, payload, created_at)
VALUES (1, 'click', '{"page": "/home"}', now());

INSERT INTO events (event_id, event_type, payload, created_at)
SELECT id, type, payload, created_at FROM local_events WHERE created_at >= current_date;

Каждый сегмент пишет свои строки напрямую в брокеры. Перед завершением INSERT коннектор дожидается подтверждения доставки всех сообщений; если за 60 секунд подтвердились не все или часть сообщений брокеры отвергли, INSERT падает. В 1.4 он в такой ситуации возвращал успех и писал WARNING в лог, а недоставленные строки терялись. Большие вставки, сотни тысяч строк на сегмент, проходят целиком: когда локальная очередь продюсера заполняется, коннектор ждёт, пока брокеры её разгребут, вместо ошибки Local: Queue full.

Партицию выбирает Kafka. Чтобы связанные сообщения попадали в одну партицию, добавьте колонку __key - Kafka гарантирует, что сообщения с одинаковым ключом окажутся вместе. Колонка __partition задаёт номер партиции явно.

CREATE FOREIGN TABLE events_out (
    event_id    int,
    event_type  text,
    payload     text,
    __key       text,
    __partition int
)
    SERVER kafka_server
    OPTIONS (topic 'events', format 'json', mpp_execute 'all segments');

INSERT INTO events_out (event_id, event_type, payload, __key)
VALUES (1, 'click', '{"page": "/home"}', 'user-123');

10. CSV

CREATE FOREIGN TABLE csv_events (
    event_id   int,
    event_type text,
    payload    text
)
    SERVER kafka_server
    OPTIONS (topic 'csv_events', format 'csv',
             csv_delimiter '|', csv_header 'true',
             mpp_execute 'all segments');

Колонки читаются по порядку, а не по именам. csv_header 'true' пропускает первую строку сообщения; строкой данных она не станет и ошибкой не считается.

JDBC и ошибка `portal "" does not exist`

Через psql этого не бывает, через DBeaver и другие JDBC-клиенты - бывает регулярно:

ERROR: portal "" does not exist   (SQLSTATE 34000)

Падает только первый запрос к внешней таблице в соединении, дальше в том же соединении всё работает; после переподключения первый запрос снова падает.

Это ограничение ядра Greenplum 6 (PostgreSQL 9.4), а не дефект коннектора. При extended query protocol (режим pgjdbc по умолчанию) вместе с autocommit = on GPDB уничтожает безымянный портал между фазами Bind и Execute для первого в сессии плана с foreign scan. Код расширения при этом не выполняется вообще: падение происходит до него. К SSL, ORCA и mpp_execute симптом отношения не имеет.

Лечится на стороне клиента, любым из трёх способов:

Перевести драйвер в simple protocol - проще всего и семантику транзакций не меняет:

jdbc:postgresql://host:5432/db?preferQueryMode=simple

В DBeaver: Edit Connection -> Driver properties -> preferQueryMode = simple.

Отключить autocommit и работать в явных транзакциях (conn.setAutoCommit(false), в DBeaver - Manual Commit).
Использовать server-side prepared statements (preferQueryMode=extendedForPrepared или именованные PreparedStatement).

Для ETL-паттерна INSERT INTO <локальная> SELECT FROM <внешняя> через JDBC действует то же самое - проще всего дописать preferQueryMode=simple в строку подключения пула или оркестратора.

Что ещё стоит знать

Прерванный или откаченный запрос не продвигает офсеты. Данные будут перечитаны при следующем запуске, в том числе если упала вставка в локальную таблицу или ошибку выдал другой сегмент. Потери нет, дубликаты возможны - если дубликаты недопустимы, дедуплицируйте по паре __partition + __offset.

Прерванный INSERT оставляет частичную запись. В Greenplum он откатывается, но то, что Kafka уже приняла до отката, откатить нечем: транзакций у продюсера нет. Планируйте повторную заливку идемпотентной.

Перебой брокера во время скана - это WARNING, а не ошибка. Смена лидера партиции или разрыв соединения пишутся в лог как transient broker error ... continuing, librdkafka переподключается, скан дочитывает до конца. Запрос упадёт только если к batch_timeout_ms конец диапазона не достигнут: scan ... timed out after 5000 ms with 3 broker error(s). Тогда повторите его, когда брокеры придут в себя, или поднимите таймаут.

Версию библиотеки сверяйте на сегментах. Сканы выполняются на сегментах, и старая .so на одном из них даёт симптомы старой версии при обновлённом расширении. SELECT gp_segment_id, kafka.version() FROM gp_dist_random('gp_id') покажет, где что стоит.

Растущий lag смотрите через мониторинг, а не по времени выполнения запросов:

SELECT sum(lag) FROM kafka.consumer_lag('kafka_server', 'etl_pipeline', 'events');