Импорт уведомлений S3

Экспериментальная функция

Импорт уведомлений S3 является экспериментальной функцией и поддерживается начиная с Doris 4.1.6.

Режим импорта уведомлений NOTIFICATION получает из SQS события создания объектов S3 и непрерывно загружает новые файлы в Doris. Он подходит для имён файлов UUID и файлов, поступающих в любом порядке, без зависимости от порядка имён файлов.

Поддерживаются только уведомления AWS S3, отправляемые напрямую в очередь SQS Standard в том же регионе. Для каждого задания нужна выделенная очередь. Исторические объекты не сканируются и не загружаются задним числом. Сквозная доставка выполняется с семантикой at-least-once: один объект может быть загружен несколько раз, а порядок загрузки файлов не гарантируется.

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

В примере используется s3://example-bucket/data/*.csv в us-east-1. Замените bucket, аккаунт, очередь и учётные данные своими значениями.

Подготовка целевой таблицы Doris

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

CREATE DATABASE IF NOT EXISTS db1;
CREATE TABLE db1.tbl1 (
    id INT,
    name STRING
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 1
PROPERTIES ("replication_num" = "1");

Этот пример с одним BE использует одну реплику. Для production скорректируйте количество реплик.

Шаг 1: Создание очереди SQS Standard

Создайте очередь Standard Queue в том же регионе, что и bucket S3, и сохраните её URL и ARN. Рекомендуется период хранения сообщений 14 дней, чтобы покрыть паузы и сбои. После истечения периода хранения SQS удаляет сообщения.

Шаг 2: Разрешение S3 отправлять уведомления

Объедините следующую Statement с политикой доступа очереди, заменив идентификаторы аккаунта, bucket и queue:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "AllowS3ToSendMessage",
      "Effect": "Allow",
      "Principal": {"Service": "s3.amazonaws.com"},
      "Action": "sqs:SendMessage",
      "Resource": "arn:aws:sqs:us-east-1:<account-id>:<queue-name>",
      "Condition": {
        "ArnEquals": {"aws:SourceArn": "arn:aws:s3:::example-bucket"},
        "StringEquals": {"aws:SourceAccount": "<account-id>"}
      }
    }
  ]
}

Шаг 3: Предоставление Doris доступа

Предоставьте пользователю или роли IAM задания следующие разрешения. FE требуется доступ к SQS и S3; BE должен читать объекты S3. Для AssumeRole также требуется подключение к STS.

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": "s3:ListBucket",
      "Resource": "arn:aws:s3:::example-bucket"
    },
    {
      "Effect": "Allow",
      "Action": "s3:GetObject",
      "Resource": "arn:aws:s3:::example-bucket/data/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes",
        "sqs:ChangeMessageVisibility"
      ],
      "Resource": "arn:aws:sqs:us-east-1:<account-id>:<queue-name>"
    }
  ]
}

Шаг 4: Настройка уведомлений о событиях S3

Создайте правило в разделе уведомлений о событиях bucket:

• Выберите Все события создания объектов (s3:ObjectCreated:*).
• Укажите Prefix data/ и Suffix .csv без подстановочных знаков. Правило должно охватывать шаблон URI задания.
• Выберите очередь SQS в качестве назначения. Отправляйте сообщения S3 напрямую, без оболочки SNS.

Включите правило до загрузки файлов для импорта. Создание правила уведомлений не генерирует события для существующих объектов.

Шаг 5: Создание задания Streaming Job

Для вывода схемы при создании задания S3 TVF нужны соответствующие файлы. После включения уведомлений загрузите непустой data/001.csv с содержимым:

1,Alice
2,Bob

Затем создайте задание:

CREATE JOB s3_event_job
PROPERTIES (
    "s3.ingestion_mode" = "NOTIFICATION",
    "s3.event.source" = "SQS",
    "s3.sqs.queue_url" = "https://sqs.us-east-1.amazonaws.com/<account-id>/<queue-name>",
    "max_interval" = "10"
)
ON STREAMING DO
INSERT INTO db1.tbl1
SELECT * FROM S3(
    "uri" = "s3://example-bucket/data/*.csv",
    "format" = "csv",
    "column_separator" = ",",
    "provider" = "S3",
    "s3.region" = "us-east-1",
    "s3.endpoint" = "https://s3.us-east-1.amazonaws.com",
    "s3.access_key" = "<access-key>",
    "s3.secret_key" = "<secret-key>"
);

Этот файл создаёт уведомление SQS. Объекты, загруженные до настройки уведомлений, можно использовать для вывода схемы, но они не загружаются автоматически задним числом. При создании задания хотя бы один соответствующий файл должен оставаться доступным для чтения.

Чтобы использовать роль IAM, замените параметры access-key и secret-key следующими; s3.external_id необязателен в зависимости от политики доверия:

"s3.role_arn" = "arn:aws:iam::<account-id>:role/<role-name>",
"s3.external_id" = "<external-id>"

FE вызывает AssumeRole с настроенными учётными данными AWS. Роли нужны разрешения S3 и SQS, вызывающей стороне — sts:AssumeRole, а политика доверия роли должна разрешать эту вызывающую сторону.

После обработки уведомления для data/001.csv таблица содержит соответствующие две строки. Последующие файлы загружаются через уведомления, даже если их ключи меньше ранее обработанных ключей. Повторяющиеся уведомления могут влиять на число строк в зависимости от модели целевой таблицы.

Шаг 6: Мониторинг отставания очереди

SELECT Name, Status, CurrentOffset, EndOffset, LagBytes,
       LastSourceEventTimestamp, LastTaskSuccessTime
FROM jobs("type" = "insert")
WHERE Name = 's3_event_job';

Пример CurrentOffset: {"fileName":"data/a.csv"}. Пример EndOffset: {"fileName":"data/b.csv","lagMessages":12}. Эти ключи описывают позиции пакетов, а не лексикографические водоразделы, и по ним нельзя определить, догнал ли импорт очередь.

Поле	Значение в режиме уведомлений
LagBytes	-1: размер в байтах поставленных в очередь объектов недоступен.
EndOffset.lagMessages	Приблизительное число видимых сообщений SQS, исключая сообщения в обработке и отложенные сообщения. Это не число файлов.
LastSourceEventTimestamp	Максимальная временная метка события S3 в последнем зафиксированном пакете, в секундах Unix. Она не доказывает, что все более ранние события обработаны.

FE также предоставляет метрики для каждого задания:

doris_fe_streaming_job_per_job_lag_messages
doris_fe_streaming_job_per_job_last_source_event_timestamp_seconds

Подтверждение и ограничения

• Обрабатываются только сохранённые уведомления SQS. Исторические объекты не сканируются.
• Для создания задания нужен соответствующий, доступный для чтения, непустой файл для вывода схемы S3 TVF.
• Каждая очередь должна использоваться только одним заданием. Doris пропускает и подтверждает записи вне области действия, некорректные или неподдерживаемые уведомления, а также тестовые события S3. Задания с разными шаблонами URI не должны совместно использовать очередь.
• Для допустимых файлов сообщения удаляются после фиксации транзакции INSERT. Во время обработки видимость продлевается, а неудачные пакеты можно повторить. После восстановления FE может использовать сохранённые ключи из последнего зафиксированного пакета для обработки его повторной доставки; Doris не сохраняет каждый исторический ключ для глобальной дедупликации.
• S3 и SQS могут доставлять дубликаты. Сквозная семантика — at-least-once, порядок загрузки файлов не гарантируется. Для дедупликации по бизнес-правилам используйте таблицу Unique Key, а для чувствительных к порядку обновлений при необходимости — столбец Sequence.
• Делайте ключи объектов уникальными, а содержимое — неизменяемым и доступным для чтения. Doris читает содержимое во время обработки, и это не обязательно версия, существовавшая при отправке события. Раннее удаление объектов приводит к сбоям и повторным попыткам.

Параметры

Настройте следующие параметры в PROPERTIES задания. Параметры подключения к S3, учётные данные и формат файла указываются в S3(...).

Параметр	Тип	По умолчанию	Обязателен	Описание
s3.ingestion_mode	Строка	ORDERED_LIST	Да	Установите NOTIFICATION. После создания изменить нельзя.
s3.event.source	Строка	None	Да	Должен быть SQS. После создания изменить нельзя.
s3.sqs.queue_url	Строка	None	Да	URL выделенной очереди SQS. После создания изменить нельзя.
s3.max_batch_files	Положительное целое	256	Нет	Порог количества файлов в пакете.
s3.max_batch_bytes	Целое число	10737418240	Нет	Порог количества байтов в пакете: от 104857600 до 10737418240. Единицы измерения в суффиксе не принимаются.
max_interval	Положительное целое	10	Нет	Интервал планирования при простое в секундах. Должен быть меньше 7200.

Мягкие пороги пакета не требуют ожидания полного пакета. Уже полученные сообщения остаются в том же пакете; следующий приём прекращается после достижения порога, поэтому пакет может превысить любой из порогов. offset запрещён при создании и в ALTER JOB. URI S3 изменить нельзя. Чтобы изменить путь или очередь, создайте другое задание.

Устранение неполадок и рекомендации

Симптом	Проверки и действия
Создание задания завершается ошибкой	Проверьте соответствующие непустые файлы, разрешения S3/SQS и сетевое подключение.
Задание выполняется, но файлы не загружаются	Проверьте, что уведомления были настроены до загрузки файлов, выбран параметр s3:ObjectCreated:*, фильтры охватывают URI, а очередь не читается другим потребителем.
Отсутствуют исторические файлы	Правила уведомлений не воспроизводят исторические события. Используйте отдельный префикс и однократный импорт.
После паузы отсутствуют файлы	Проверьте период хранения сообщений SQS и при необходимости отдельно загрузите пропущенные файлы.
Повторяющиеся сбои чтения	Проверьте ErrorMsg и ошибки заданий. Убедитесь, что объекты доступны, неизменяемы и совместимы с форматом.
Дубликаты строк	Проверьте бизнес-ключи и модель целевой таблицы. Не полагайтесь на глобальную дедупликацию исторических файлов.

Установите достаточный период хранения сообщений, сохраняйте объекты неизменяемыми и доступными для чтения, а также отслеживайте приблизительное отставание, время последнего успешного выполнения и ошибки заданий. Управление заданиями описано в разделе Обзор непрерывной загрузки и в разделе ALTER JOB.