Перейти к содержимому
Последнее обновление

Упорядоченная загрузка из S3

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

Непрерывная загрузка Doris является экспериментальной и поддерживается начиная с версии Doris 4.1.5.

Упорядоченная загрузка использует ORDERED_LIST для непрерывного обнаружения новых объектов S3 в лексикографическом порядке ключей, включая путь и имя файла. Она подходит для журналов и ETL-файлов, названных с возрастающими временными метками или порядковыми номерами. Этот порядок определяет обнаружение файлов и не гарантирует порядок строк в целевой таблице.

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

Принцип работы​

Doris ищет соответствующие файлы после последнего зафиксированного ключа. Новый ключ должен быть лексикографически больше зафиксированного ключа. После загрузки file1, file2 и file3 вновь загруженный file0 прочитан не будет. Если новых файлов нет, задание выполняет опрос с интервалом max_interval.

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

Шаг 1: Подготовьте целевую таблицу и файлы​

Учетной записи, от имени которой выполняется задание, требуется разрешение LOAD на целевую таблицу, а также соответствующие права для создания базы данных и таблицы. Учетные данные S3 должны иметь разрешения s3:ListBucket и s3:GetObject. Узлы Doris должны иметь доступ к конечной точке S3.

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 используется одна реплика. Установите число реплик в соответствии с вашим производственным кластером. Загрузите следующие данные как s3://bucket/001.csv и замените bucket, endpoint, Region и учетные данные в SQL своими значениями:

1,Alice
2,Bob

Шаг 2: Создайте задание​

Не указывайте s3.ingestion_mode, чтобы использовать значение по умолчанию, либо явно задайте ORDERED_LIST в PROPERTIES задания.

CREATE JOB my_job
ON STREAMING
DO
INSERT INTO db1.tbl1
select * from S3(
    "uri" = "s3://bucket/*.csv",
    "s3.access_key" = "<s3_access_key>",
    "s3.secret_key" = "<s3_secret_key>",
    "s3.region" = "<s3_region>",
    "s3.endpoint" = "<s3_endpoint>",
    "format" = "csv",
    "column_separator" = ","
);

После загрузки выполните SELECT * FROM db1.tbl1 ORDER BY id;. Ожидаемый результат:

+----+-------+
| id | name  |
+----+-------+
|  1 | Alice |
|  2 | Bob   |
+----+-------+

Шаг 3: Отслеживайте ход выполнения​

SELECT Name, Status, CurrentOffset, EndOffset, ErrorMsg
FROM jobs("type" = "insert")
WHERE Name = 'my_job';

После фиксации 001.csv ожидается, что Status будет RUNNING, а CurrentOffset — {"fileName":"001.csv"}. Загрузите файл с большим ключом, например 002.csv, чтобы продолжить загрузку. Если новых файлов нет, задание остается выполняющимся.

Параметры​

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

ПараметрТипЗначение по умолчаниюОбязательныйОписание
s3.ingestion_modeСтрокаORDERED_LISTНетЗадайте ORDERED_LIST. После создания изменить нельзя.
offsetСтрока JSONНетНетЧитать строго после этого ключа, например '{"fileName":"data/001.csv"}'. Без этого параметра чтение начинается с начала соответствующего пути.
s3.max_batch_filesПоложительное целое число256НетПорог количества файлов в каждой партии.
s3.max_batch_bytesЦелое число10737418240НетПорог в байтах для каждой партии, от 104857600 до 10737418240. Суффиксы единиц не принимаются.
max_intervalПоложительное целое число10НетИнтервал планирования при простое, в секундах.

Пороговые значения партии не требуют ожидания полного пакета перед вставкой.

Начало после определенного файла​

Если data/001.csv и более ранние файлы уже обработаны, начните чтение с data/002.csv с помощью:

CREATE JOB s3_after_offset_job
PROPERTIES (
    "s3.ingestion_mode" = "ORDERED_LIST",
    "offset" = "{\"fileName\":\"data/001.csv\"}"
)
ON STREAMING DO
INSERT INTO db1.tbl1
SELECT * FROM S3(
    "uri" = "s3://bucket/data/*.csv",
    "format" = "csv",
    "column_separator" = ",",
    "s3.region" = "<s3_region>",
    "s3.endpoint" = "<s3_endpoint>",
    "s3.access_key" = "<s3_access_key>",
    "s3.secret_key" = "<s3_secret_key>"
);

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

Режим загрузки и S3 URI нельзя изменить после создания. Чтобы изменить режим или путь, создайте другое задание.

Устранение неполадок и ограничения​

  • Новые файлы не загружаются: проверьте, что ключи соответствуют URI и больше CurrentOffset. Файлы, добавленные до курсора, не будут автоматически дозагружены. Для дозагрузки используйте отдельный префикс и однократное задание либо настройте загрузку по уведомлениям до загрузки файлов.
  • Неожиданный порядок: используйте временные метки фиксированной ширины или счетчики с ведущими нулями, например 000001.csv. Иначе file10 сортируется перед file2.
  • Объекты перезаписываются или удаляются: каждый ключ должен быть уникальным, а его содержимое — неизменным и доступным для чтения во время загрузки. Перезапись уже зафиксированного объекта не вызывает повторную загрузку.
  • Создание завершается ошибкой для пустого пути: при создании задания S3 TVF требуется соответствующий доступный для чтения непустой файл, чтобы вывести схему.

Для приостановки, возобновления и проверки задач см. Обзор непрерывной загрузки. Для изменения задания см. ALTER JOB.

Report issue