Последнее обновление 16 окт. 2026 г.

AWS Kinesis

Routine Load позволяет непрерывно читать данные из AWS Kinesis Data Streams и загружать записи в таблицы Doris. Поток состоит из шардов; Doris отслеживает прогресс по порядковому номеру для каждого шарда.

Версия

Kinesis Routine Load доступен в версии dev. Ветки исходного кода 4.x не включают этот источник данных; это руководство не относится к этим версиям.

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

Используйте сборку Doris с поддержкой Kinesis Routine Load. Узлы BE должны иметь доступ к конечной точке Kinesis.
Подготовьте поток Kinesis и учётные данные, позволяющие выполнять для него операции kinesis:ListShards, kinesis:GetShardIterator и kinesis:GetRecords.
Используйте учётную запись Doris с правами на создание баз данных и таблиц, а также с правом LOAD_PRIV для целевой таблицы. Для административных операций над существующим заданием также требуются соответствующие права на загрузку.
Для выполнения приведённой ниже команды отправки данных настройте AWS CLI с учётными данными, имеющими разрешение kinesis:PutRecord. Используйте отдельный тестовый поток или учитывайте записи, уже имеющиеся в потоке.

Doris поддерживает пару из ключа доступа и секретного ключа, временные учётные данные с токеном сеанса, принятие роли IAM или стандартную цепочку поиска учётных данных AWS, доступную процессу BE. Общие рекомендации по учётным данным AWS см. в разделе Аутентификация AWS.

Быстрый старт

1. Создайте базу данных и целевую таблицу

Выполните следующий SQL с помощью клиента MySQL. В примере используется одна реплика для кластера разработки с одним BE; для промышленной эксплуатации выберите подходящее количество реплик.

CREATE DATABASE IF NOT EXISTS kinesis_demo;
USE kinesis_demo;

CREATE TABLE kinesis_events (
    id BIGINT NOT NULL,
    name STRING
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1");

2. Создайте задание Routine Load

Перед выполнением этого оператора замените имя потока, регион и учётные данные. Свойство называется kinesis_stream, без префикса aws.. Значение TRIM_HORIZON включает чтение с самых старых записей, которые ещё хранятся в потоке.

CREATE ROUTINE LOAD kinesis_demo.kinesis_job ON kinesis_events
COLUMNS(id, name)
PROPERTIES (
    "format" = "json",
    "desired_concurrent_number" = "1"
)
FROM KINESIS (
    "aws.region" = "us-east-1",
    "kinesis_stream" = "my-test-stream",
    "aws.access_key" = "<access_key>",
    "aws.secret_key" = "<secret_key>",
    "property.kinesis_default_pos" = "TRIM_HORIZON"
);

Если в среде BE уже настроена подходящая стандартная цепочка поиска учётных данных AWS, не указывайте свойства ключа доступа и секретного ключа. Настройки ролей и временных учётных данных описаны в справочнике параметров.

3. Отправьте тестовую запись и проверьте её

В оболочке с AWS CLI v2, настроенным для тестового потока:

aws kinesis put-record \
  --region us-east-1 \
  --stream-name my-test-stream \
  --partition-key 1 \
  --cli-binary-format raw-in-base64-out \
  --data '{"id":1,"name":"Alice"}'

Команда возвращает ID шарда и порядковый номер записи. Routine Load работает асинхронно. После того как задание прочитает запись и зафиксирует её, выполните запрос к Doris:

SELECT id, name FROM kinesis_demo.kinesis_events WHERE id = 1;

+----+-------+
| id | name  |
+----+-------+
|  1 | Alice |
+----+-------+

Параметры

Все перечисленные ниже свойства относятся к секции FROM KINESIS (...). Если необязательное свойство не указано, используется его значение по умолчанию. Общие свойства задания, такие как format, jsonpaths и desired_concurrent_number, относятся к секции PROPERTIES (...); см. Routine Load.

Параметр	Тип / Обязательность	По умолчанию	Описание
`aws.region`	Строка; обязательный	Нет	Регион, в котором находится поток, например `us-east-1`.
`kinesis_stream`	Строка; обязательный	Нет	Имя потока. Не используйте `aws.kinesis_stream`.
`aws.endpoint`	Строка; необязательный	Конечная точка AWS для региона	Пользовательская конечная точка Kinesis, например конечная точка LocalStack. Также поддерживается устаревший псевдоним `kinesis_endpoint`; предпочтительно использовать `aws.endpoint`.
`aws.access_key`	Строка; необязательный	Стандартная цепочка поиска учётных данных AWS, если учётные данные не указаны явно	ID ключа доступа. Необходимо указывать вместе с `aws.secret_key`.
`aws.secret_key`	Строка; необязательный	Нет	Секретный ключ, соответствующий `aws.access_key`.
`aws.session_key`	Строка; необязательный	Нет	Токен сеанса для временных учётных данных.
`aws.role_arn`	Строка; необязательный	Нет	Роль IAM, которую требуется принять; субъект, использующий эти учётные данные, должен иметь право принять эту роль.
`kinesis_shards`	Строка со значениями через запятую; необязательный	Автоматическое обнаружение шардов	ID шардов для чтения.
`kinesis_shards_pos`	Строка со значениями через запятую; необязательный	`LATEST` для каждого явно указанного шарда	Позиции, по одной для каждого шарда из `kinesis_shards`. Каждым значением может быть `TRIM_HORIZON`, `LATEST` или порядковый номер. При указании порядкового номера чтение начинается после соответствующей записи.
`property.kinesis_default_pos`	Строка; необязательный	`LATEST`	Начальная позиция по умолчанию: `TRIM_HORIZON`, `LATEST` или порядковый номер. Нельзя использовать совместно с `kinesis_shards_pos`. Строки с временными метками не поддерживаются.
Другие `property.*`	Строка; необязательный	Зависит от параметра BE	Передаются на BE после удаления префикса `property.`. Например, `property.aws.external.id` задаёт внешний ID для принятия роли и требует указания ARN роли. Используйте только параметры, поддерживаемые BE.

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

SHOW ROUTINE LOAD FOR kinesis_demo.kinesis_job;

Поле	Ожидаемое значение
`State`	`RUNNING` после того, как задание запланировано и может читать поток. Если задание находится в состоянии `PAUSED`, проверьте причину приостановки.
`DataSourceType`	`KINESIS`.
`DataSourceProperties`	Регион, поток, конечная точка и настроенные шарды.
`Progress`	Прогресс по порядковым номерам для каждого шарда. Порядковые номера не являются смещениями Kafka, и их нельзя складывать для подсчёта записей.
`Lag`	Отставание каждого шарда от последней записи в миллисекундах, если эти данные доступны.

PAUSE ROUTINE LOAD FOR kinesis_demo.kinesis_job;
RESUME ROUTINE LOAD FOR kinesis_demo.kinesis_job;
STOP ROUTINE LOAD FOR kinesis_demo.kinesis_job;

PAUSE прекращает планирование новых задач и позволяет впоследствии возобновить задание. STOP окончательно завершает задание; остановленное задание нельзя возобновить.

Важные замечания и рекомендации

По умолчанию используется LATEST, при котором исторические записи не читаются. Укажите TRIM_HORIZON, чтобы читать самые старые сохранившиеся данные, или задайте явные позиции шардов для управляемого повторного чтения. Данные, срок хранения которых в Kinesis уже истёк, невозможно восстановить изменением начальной позиции.
В явно заданных списках ID шардов и позиций должно быть одинаковое количество элементов. Не используйте kinesis_shards_pos и property.kinesis_default_pos одновременно.
Если при создании задания сообщается о недопустимом свойстве, проверьте написание его имени и учитывайте различия между свойствами задания и источника. В частности, aws.kinesis_stream и строки с временными метками в качестве начальной позиции недопустимы.
Если задание приостанавливается, проверьте вывод SHOW ROUTINE LOAD и журналы BE на наличие ошибок, связанных с аутентификацией, конечной точкой, шардами и форматом данных. Записи JSON должны соответствовать целевым столбцам или настроенным сопоставлениям.
При выборе числа параллельных задач учитывайте характерную скорость передачи данных в потоке. Пропускная способность может быть ограничена лимитами чтения Kinesis и распределением шардов; увеличение числа задач не позволяет обойти ограничения AWS для каждого шарда.

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

Быстрый старт​

1. Создайте базу данных и целевую таблицу​

2. Создайте задание Routine Load​

3. Отправьте тестовую запись и проверьте её​

Параметры​

Мониторинг и управление заданием​

Важные замечания и рекомендации​