Загрузка уведомлений S3

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

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

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

Поддерживаются только уведомления AWS S3, отправленные напрямую в SQS Standard Queue в том же регионе. Для каждой задачи требуется выделенная очередь. Исторические объекты не сканируются и не догружаются. Сквозная доставка имеет семантику 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 используется одна реплика. Для рабочей среды настройте число реплик.

Шаг 1: создание SQS Standard Queue

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

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

Добавьте следующий блок в политику доступа к очереди, заменив идентификаторы учетной записи, 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	Строка	Нет	Да	Должно быть SQS. После создания изменить нельзя.
s3.sqs.queue_url	Строка	Нет	Да	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.