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

Одноразовый импорт из 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 и замените bucket, endpoint, Region и учетные данные в SQL фактическими значениями:

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 задания. Параметры подключения к 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