Doris Streamloader

Doris Streamloader — специализированный клиентский инструмент для загрузки данных в базу данных Apache Doris. По сравнению с однопоточной загрузкой напрямую через curl этот инструмент поддерживает параллельную загрузку и значительно сокращает время загрузки больших объёмов данных.

Основные возможности

Возможность	Описание
Параллельная загрузка	Выполняет Stream Load с несколькими параллельными исполнителями. Уровень параллелизма задаётся параметром workers
Загрузка нескольких файлов	Загружает несколько файлов и каталогов в рамках одной задачи. Поддерживает подстановочные знаки и автоматически обходит все файлы в каталоге рекурсивно
Возобновление передачи	Если во время загрузки происходит частичный сбой, инструмент может возобновить её с места сбоя
Автоматические повторные попытки	После сбоя загрузки повторять её вручную не требуется. Инструмент автоматически выполняет повторные попытки в пределах заданного по умолчанию количества. Если загрузка по-прежнему не удаётся, он выводит команду для повторной попытки вручную

Сценарии использования

• Пакетная загрузка больших объёмов данных (от GB до TB) в Doris
• Пакетная загрузка нескольких файлов и каталогов
• Сценарии, чувствительные к задержке загрузки, в которых для повышения пропускной способности необходим параллелизм
• Надёжные процессы загрузки, требующие возобновления передачи и автоматического восстановления после сбоев

---

Скачивание и установка

Ресурс	Адрес
Исходный код	https://github.com/apache/doris-streamloader
Бинарный файл	https://doris.apache.org/download

примечание

Скачиваемый файл представляет собой исполняемый бинарный файл. Дополнительная компиляция или установка не требуется.

---

Использование

Базовый формат команды

doris-streamloader \
    --source_file={FILE_LIST} \
    --url={FE_OR_BE_SERVER_URL}:{PORT} \
    --header={STREAMLOAD_HEADER} \
    --db={TARGET_DATABASE} \
    --table={TARGET_TABLE}

Обязательные параметры

Параметр	Назначение
--source_file	Список файлов данных для загрузки. Поддерживает отдельный файл, каталог, подстановочные знаки и список через запятую
--url	Адрес сервиса Doris FE или BE в формате http://host:port
--header	Параметры заголовков для Stream Load. Несколько параметров разделяются символом ?
--db	Имя целевой базы данных
--table	Имя целевой таблицы

Форматы, поддерживаемые source_file

Параметр --source_file поддерживает следующие пять форматов. Выберите подходящий для вашего сценария.

1. Один файл

Например, чтобы загрузить один файл file.csv:

doris-streamloader --source_file="file.csv" --url="http://localhost:8330" --header="column_separator:|?columns:col1,col2" --db="testdb" --table="testtbl"

2. Один каталог

Например, чтобы загрузить каталог dir:

doris-streamloader --source_file="dir" --url="http://localhost:8330" --header="column_separator:|?columns:col1,col2" --db="testdb" --table="testtbl"

3. Имя файла с подстановочными знаками (обязательно в кавычках)

Например, чтобы загрузить file0.csv, file1.csv и file2.csv:

doris-streamloader --source_file="file*" --url="http://localhost:8330" --header="column_separator:|?columns:col1,col2" --db="testdb" --table="testtbl"

4. Список имён файлов через запятую

Например, чтобы загрузить file0.csv, file1.csv и file2.csv:

doris-streamloader --source_file="file0.csv,file1.csv,file2.csv" --url="http://localhost:8330" --header="column_separator:|?columns:col1,col2" --db="testdb" --table="testtbl"

5. Список каталогов через запятую

Например, чтобы загрузить dir1, dir2 и dir3:

doris-streamloader --source_file="dir1,dir2,dir3" --url="http://localhost:8330" --header="column_separator:|?columns:col1,col2" --db="testdb" --table="testtbl"

Параметр заголовков

--header поддерживает все параметры Stream Load. Несколько параметров разделяются символом ?.

Пример:

doris-streamloader --source_file="data.csv" --url="http://localhost:8330" --header="column_separator:|?columns:col1,col2" --db="testdb" --table="testtbl"

---

Необязательные параметры

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

Аутентификация и передача данных

Параметр	Назначение	Значение по умолчанию	Рекомендация
--u	Имя пользователя базы данных	root	——
--p	Пароль пользователя базы данных	Пустая строка	——
--compress	Сжимать ли данные при передаче по HTTP	false	Оставьте значение по умолчанию. Сжатие увеличивает нагрузку на CPU как на стороне инструмента, так и на стороне Doris BE из-за операций сжатия и распаковки. Включайте его, только если узким местом является пропускная способность сети на машине с исходными данными
--timeout	Тайм-аут HTTP-запросов к Doris, в секундах	60*60*10	Оставьте значение по умолчанию

Пакетная обработка и параллелизм

Параметр	Назначение	Значение по умолчанию	Рекомендация
--batch	Размер порции при чтении и отправке файлов, в строках	4096	Оставьте значение по умолчанию
--batch_byte	Размер порции при чтении и отправке файлов, в байтах	943718400 (900 MB)	Оставьте значение по умолчанию
--workers	Уровень параллелизма загрузки	0	При значении 0 инструмент работает в автоматическом режиме и вычисляет значение на основе объёма загружаемых данных, пропускной способности диска и скорости загрузки Stream Load. Значение можно задать и вручную. Для высокопроизводительных кластеров его можно увеличить, но желательно не более 10. Если при загрузке потребляется слишком много памяти (это видно по Memtracker или сообщениям Exceed в журналах), значение можно уменьшить
--disk_throughput	Пропускная способность диска, в MB/s	800	Обычно оставляйте значение по умолчанию. Этот параметр участвует в автоматическом расчёте --workers. Чтобы инструмент вычислил подходящее количество workers, задайте значение на основе фактической пропускной способности диска
--streamload_throughput	Фактическая пропускная способность загрузки Stream Load, в MB/s	100	Обычно оставляйте значение по умолчанию. Этот параметр участвует в автоматическом расчёте --workers. Значение по умолчанию получено в среде ежедневного тестирования производительности. Чтобы инструмент вычислил подходящее количество workers, задайте значение на основе измеренной пропускной способности по формуле: (LoadBytes*1000) / (LoadTimeMs*1024*1024)
--max_byte_per_task	Максимальный объём данных на одну задачу загрузки. При превышении этого предела данные выделяются в новую задачу	107374182400 (100 GB)	Рекомендуется использовать большее значение, чтобы уменьшить число версий загрузки. Однако при ошибке body exceed max size, если вы не хотите изменять streaming_load_max_mb (для этого требуется перезапуск BE), либо при ошибке -238 TOO MANY SEGMENT значение можно временно уменьшить

Проверка данных и журналирование

Параметр	Назначение	Значение по умолчанию	Рекомендация
--check_utf8	Проверять ли кодировку загружаемых данных: false пропускает проверку и загружает исходные данные; true заменяет символы, не соответствующие UTF-8, на �	true	Оставьте значение по умолчанию
--debug	Выводить ли отладочные журналы	false	Оставьте значение по умолчанию
--log_filename	Место хранения журналов	""	По умолчанию журналы выводятся в консоль. Чтобы записывать их в файл, укажите путь, например --log_filename="/var/log"

Повторные попытки при сбоях

Параметр	Назначение	Значение по умолчанию	Рекомендация
--auto_retry	Список номеров исполнителей и задач для автоматической повторной попытки	Пустая строка	Используйте только при сбое загрузки. При обычной загрузке задавать этот параметр не требуется. При сбое выводятся конкретные значения параметров. Достаточно скопировать и выполнить команду с ними. Например, --auto_retry="1,1,2,1" означает, что нужно повторить первую задачу первого исполнителя и первую задачу второго исполнителя
--auto_retry_times	Количество автоматических повторных попыток	3	Оставьте значение по умолчанию. Чтобы отключить повторные попытки, задайте 0
--auto_retry_interval	Интервал между автоматическими повторными попытками, в секундах	60	Оставьте значение по умолчанию. Если причиной сбоев является недоступность Doris, задайте интервал с учётом фактического времени перезапуска

---

Результат

Независимо от того, завершилась загрузка успешно или со сбоем, по окончании инструмент выводит итоговый результат.

Поля результата

Поле	Описание
Status	Статус загрузки. Success означает успех, а Failed означает сбой
TotalRows	Общее число строк в файлах, предназначенных для загрузки
FailLoadRows	Число строк, которые предполагалось загрузить, но загрузить не удалось
LoadedRows	Число строк, фактически загруженных в Doris
FilteredRows	Число строк, отфильтрованных Doris при загрузке
UnselectedRows	Число строк, проигнорированных Doris при загрузке
LoadBytes	Число фактически загруженных байтов
LoadTimeMs	Фактическая продолжительность загрузки, в миллисекундах
LoadFiles	Список фактически загруженных файлов

Пример успешной загрузки

При успешной загрузке вывод выглядит следующим образом:

Load Result: {
        "Status": "Success",
        "TotalRows": 120,
        "FailLoadRows": 0,
        "LoadedRows": 120,
        "FilteredRows": 0,
        "UnselectedRows": 0,
        "LoadBytes": 40632,
        "LoadTimeMs": 971,
        "LoadFiles": [
                "basic.csv",
                "basic_data1.csv",
                "basic_data2.csv",
                "dir1/basic_data.csv",
                "dir1/basic_data.csv.1",
                "dir1/basic_data1.csv"
        ]
}

Пример сбоя загрузки

Если часть данных загрузить не удалось, инструмент сначала выводит команду для повторной попытки:

load has some error, and auto retry failed, you can retry by :
./doris-streamloader --source_file /mnt/disk1/laihui/doris/tools/tpch-tools/bin/tpch-data/lineitem.tbl.1  --url="http://127.0.0.1:8239" --header="column_separator:|?columns: l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag,l_linestatus, l_shipdate,l_commitdate,l_receiptdate,l_shipinstruct,l_shipmode,l_comment,temp" --db="db" --table="lineitem1" -u root -p "" --compress=false --timeout=36000 --workers=3 --batch=4096 --batch_byte=943718400 --max_byte_per_task=1073741824 --check_utf8=true --report_duration=1 --auto_retry="2,1;1,1;0,1" --auto_retry_times=0 --auto_retry_interval=60

Скопируйте и выполните эту команду, чтобы повторить загрузку вручную. Значение auto_retry описано выше в разделе параметров. Затем выводится результат неудачной загрузки:

Load Result: {
      "Status": "Failed",
      "TotalRows": 1,
      "FailLoadRows": 1,
      "LoadedRows": 0,
      "FilteredRows": 0,
      "UnselectedRows": 0,
      "LoadBytes": 0,
      "LoadTimeMs": 104,
      "LoadFiles": [
              "/mnt/disk1/laihui/doris/tools/tpch-tools/bin/tpch-data/lineitem.tbl.1"
      ]
}

---

Рекомендации по использованию

Рекомендуемые параметры

1. Обязательные параметры: Необходимо настроить следующие параметры:

   --source_file=FILE_LIST
   --url=FE_OR_BE_SERVER_URL_WITH_PORT
   --header=STREAMLOAD_HEADER
   --db=TARGET_DATABASE
   --table=TARGET_TABLE

   Для загрузки нескольких файлов используйте source_file.

2. workers: Значение по умолчанию 0 включает автоматический режим (инструмент вычисляет значение на основе объёма загрузки, disk_throughput и streamload_throughput, обычно получая 1, 2, 4 или 8). Его можно задать и вручную. Для высокопроизводительных кластеров значение можно увеличить, но рекомендуется не превышать 10. Часто вручную задают 8.

3. max_byte_per_task: Большее значение уменьшает число версий загрузки. Однако при ошибке body exceed max size, если вы не хотите изменять streaming_load_max_mb (для этого требуется перезапуск BE), либо при ошибке -238 TOO MANY SEGMENT это значение можно временно уменьшить. Обычно подходит значение по умолчанию.

4. Два ключевых параметра, влияющих на число версий:

   Параметр	Влияние	Рекомендация
   workers	Чем больше исполнителей, тем больше версий и выше параллелизм	Обычно используйте 8
   max_byte_per_task	Чем больше значение, тем больше данных в каждой версии и тем меньше версий, однако слишком большое значение может вызвать -238 TOO MANY SEGMENT	Обычно используйте значение по умолчанию

Рекомендуемая команда

Для большинства сценариев достаточно задать обязательные параметры и установить для workers значение 8:

./doris-streamloader \
    --source_file="demo.csv,demoFile*.csv,demoDir" \
    --url="http://127.0.0.1:8030" \
    --header="column_separator:," \
    --db="demo" \
    --table="test_load" \
    --u="root" \
    --workers=8

---

Часто задаваемые вопросы

1. Что делать, если некоторые подзадачи завершаются сбоем во время загрузки?

Инструмент автоматически выполняет повторные попытки. Если они тоже завершаются сбоем, он выводит команду для повторной попытки вручную. Достаточно скопировать и выполнить её. Удалять таблицу и заново загружать данные не требуется.

2. Что делать, если объём одной загрузки превышает порог BE streaming_load_max_mb по умолчанию?

По умолчанию инструмент ограничивает объём одной загрузки значением 100 GB, которое может превышать порог BE streaming_load_max_mb. Чтобы обойтись без перезапуска BE, уменьшите значение параметра --max_byte_per_task.

Чтобы проверить значение streaming_load_max_mb:

curl "http://127.0.0.1:8040/api/show_config"

3. Что делать при ошибке -238 TOO MANY SEGMENT?

Уменьшение значения параметра --max_byte_per_task помогает устранить эту проблему.