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

Упорядоченная загрузка из 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 и все ключи, лексикографически меньшие его, загружая только соответствующие ключи после него. Во избежание пересечения с примером быстрого старта используйте отдельную целевую таблицу.

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

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

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

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

Report issue