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

Одноразовый импорт из S3

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

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

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

Подготовьте все файлы до создания задания и поддерживайте стабильность пути источника во время импорта. Завершенное задание не ожидает появления новых файлов. Для непрерывного импорта выберите упорядоченный импорт или импорт по уведомлениям.

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

Шаг 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 и замените в SQL значения bucket, endpoint, Region и учетных данных на свои:

1,Alice
2,Bob

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

CREATE JOB s3_one_time_job
PROPERTIES (
    "s3.ingestion_mode" = "ONE_TIME",
    "s3.max_batch_files" = "256",
    "s3.max_batch_bytes" = "10737418240"
)
ON STREAMING DO
INSERT INTO db1.tbl1
SELECT * FROM S3(
    "uri" = "s3://bucket/*.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>"
);

Проверка завершения​

Запросите состояние задания:

SELECT Name, ExecuteType, Status
FROM jobs("type" = "insert")
WHERE Name = 's3_one_time_job';

Ожидаемый результат:

Name             ExecuteType  Status
s3_one_time_job  STREAMING    FINISHED

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

id  name
1   Alice
2   Bob

Ограничения и восстановление​

ONE_TIME не создает снимок каталога при создании. Он перечисляет файлы пакетами и останавливается после фиксации последнего пакета. Файлы, добавленные до зафиксированного курсора, не будут загружены задним числом. Подготовьте все исходные файлы до создания задания и прекратите запись или перезапись объектов по этому пути во время импорта. После перезапуска FE задание возобновляет работу с зафиксированного прогресса и переходит в FINISHED после завершения обработки.

Параметры​

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

ПараметрТипЗначение по умолчаниюОбязательныйОписание
s3.ingestion_modeСтрокаORDERED_LISTДаУстановите значение ONE_TIME. После создания изменить его нельзя.
s3.max_batch_filesПоложительное целое число256НетПорог количества файлов в одном пакете.
s3.max_batch_bytesЦелое число10737418240НетПорог в байтах для одного пакета: от 104857600 до 10737418240. Суффиксы единиц измерения не поддерживаются.
max_intervalПоложительное целое число10НетИнтервал планирования при простое, в секундах.

Не задавайте offset, s3.event.source или s3.sqs.queue_url. Параметр max_interval не заставляет завершенное задание ждать новые файлы. Параметры сеанса см. в разделе INSERT INTO SELECT.

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

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

  • Не удается создать задание: убедитесь, что для определения схемы S3 TVF существует соответствующий, доступный для чтения и непустой файл.
  • Задание не завершается: проверьте Status, ErrorMsg и задачи с ошибками. Нечитаемые объекты и ошибки формата могут помешать фиксации пакета.
  • Новые файлы не читаются после завершения: FINISHED — конечное состояние. Поместите следующий пакет в отдельный префикс и создайте новое задание, чтобы избежать повторной загрузки старых данных.
  • Сочетание с импортом по уведомлениям: используйте непересекающиеся префиксы для исторических и поступающих данных либо обрабатывайте пересечения и дубликаты в приложении.

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

Report issue