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

Stream Load

Stream Load поддерживает загрузку локальных файлов или потоков данных в Doris по HTTP. Stream Load — это синхронный способ загрузки: результат возвращается после выполнения, поэтому по ответу можно определить, прошла ли загрузка успешно. Как правило, Stream Load можно использовать для загрузки файлов размером менее 10 GB. Более крупные файлы сначала разделите на части, а затем загрузите каждую часть с помощью Stream Load. Stream Load гарантирует атомарность пакета задач загрузки: либо все строки загружаются успешно, либо не загружается ни одна.

подсказка

Для загрузки рекомендуется использовать специализированный инструмент Doris Streamloader, который по сравнению с загрузкой в один поток напрямую через curl обеспечивает параллельную загрузку и сокращает время загрузки больших объёмов данных. Это клиентский инструмент, специально разработанный для загрузки данных в Doris. Инструкции и практические рекомендации приведены в документации Doris Streamloader.

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

Stream Load подходит для следующих типичных сценариев:

  • Загрузка локальных или удалённых файлов CSV, JSON, Parquet или ORC в Doris.
  • Непрерывная запись данных из программ (Java, Go, Python и других) через потоки HTTP.
  • Сценарии, требующие атомарности отдельного задания загрузки: либо успешно загружаются все строки, либо загрузка всех строк завершается неудачно.
  • Загрузка отдельных файлов, размер которых обычно меньше 10 GB. Более крупные файлы разделите перед загрузкой или используйте Doris Streamloader.

При загрузке файлов CSV чётко различайте значения null и пустые строки:

  • Значение null: обозначается как \N. Например, a,\N,b означает, что средний столбец содержит null.
  • Пустая строка: если между двумя разделителями нет символов, значением является пустая строка. Например, в a,,b между двумя запятыми нет символов, поэтому средний столбец содержит пустую строку.

Основные принципы​

При использовании Stream Load задание загрузки отправляется на узел FE по HTTP. FE перенаправляет запрос на узел BE по круговому алгоритму для балансировки нагрузки. Также можно отправить HTTP-запрос непосредственно на определённый узел BE. В Stream Load Doris назначает один узел координатором (Coordinator). Узел Coordinator отвечает за получение данных и их распределение между другими узлами.

На следующей схеме показан основной процесс работы Stream Load:

Основные принципы Stream Load

Порядок выполнения:

  1. Клиент отправляет на FE запрос на выполнение задания загрузки Stream Load.
  2. FE по круговому алгоритму выбирает BE в качестве узла Coordinator, отвечающего за планирование задания загрузки, и возвращает клиенту перенаправление HTTP.
  3. Клиент подключается к узлу BE, выполняющему роль Coordinator, и отправляет запрос на загрузку.
  4. BE, выполняющий роль Coordinator, распределяет данные между соответствующими узлами BE. После завершения загрузки он возвращает клиенту результат.
  5. Клиент также может напрямую назначить узел BE координатором и поручить ему распределение задания загрузки.

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

Запросы и данные Stream Load передаются по HTTP. В следующих примерах показано, как отправить задание загрузки через Stream Load с помощью инструмента curl.

Предварительные требования​

Для Stream Load требуется привилегия INSERT на целевую таблицу. Если у пользователя нет привилегии INSERT, предоставьте её с помощью команды GRANT.

Создание задания загрузки​

Загрузка данных CSV​

  1. Подготовьте данные для загрузки

    Создайте файл CSV streamload_example.csv со следующим содержимым:

    1,Emily,25
    2,Benjamin,35
    3,Olivia,28
    4,Alexander,60
    5,Ava,17
    6,William,69
    7,Sophia,32
    8,James,64
    9,Emma,37
    10,Liam,64

  2. Создайте целевую таблицу Doris

    Создайте в Doris таблицу для загрузки данных с помощью следующей инструкции:

    CREATE TABLE testdb.test_streamload(
        user_id            BIGINT       NOT NULL COMMENT "user id",
        name               VARCHAR(20)           COMMENT "name",
        age                INT                   COMMENT "age"
    )
    DUPLICATE KEY(user_id)
    DISTRIBUTED BY HASH(user_id) BUCKETS 10;

  3. Запустите задание загрузки

    Задание загрузки Stream Load можно отправить с помощью команды curl.

    curl --location-trusted -u <doris_user>:<doris_password> \
        -H "Expect:100-continue" \
        -H "column_separator:," \
        -H "columns:user_id,name,age" \
        -T streamload_example.csv \
        -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

    Stream Load — это синхронный способ загрузки, при котором результат возвращается непосредственно пользователю.

    {
        "TxnId": 3,
        "Label": "123",
        "Comment": "",
        "TwoPhaseCommit": "false",
        "Status": "Success",
        "Message": "OK",
        "NumberTotalRows": 10,
        "NumberLoadedRows": 10,
        "NumberFilteredRows": 0,
        "NumberUnselectedRows": 0,
        "LoadBytes": 118,
        "LoadTimeMs": 173,
        "BeginTxnTimeMs": 1,
        "StreamLoadPutTimeMs": 70,
        "ReadDataTimeMs": 2,
        "WriteDataTimeMs": 48,
        "CommitAndPublishTimeMs": 52
    }

  4. Просмотрите загруженные данные

    mysql> select count(*) from testdb.test_streamload;
    +----------+
    | count(*) |
    +----------+
    |       10 |
    +----------+

Загрузка данных JSON​

  1. Подготовьте данные для загрузки

    Создайте файл JSON streamload_example.json со следующим содержимым:

    [
    {"userid":1,"username":"Emily","userage":25},
    {"userid":2,"username":"Benjamin","userage":35},
    {"userid":3,"username":"Olivia","userage":28},
    {"userid":4,"username":"Alexander","userage":60},
    {"userid":5,"username":"Ava","userage":17},
    {"userid":6,"username":"William","userage":69},
    {"userid":7,"username":"Sophia","userage":32},
    {"userid":8,"username":"James","userage":64},
    {"userid":9,"username":"Emma","userage":37},
    {"userid":10,"username":"Liam","userage":64}
    ]

  2. Создайте целевую таблицу Doris

    Создайте в Doris таблицу для загрузки данных с помощью следующей инструкции:

    CREATE TABLE testdb.test_streamload(
        user_id            BIGINT       NOT NULL COMMENT "user id",
        name               VARCHAR(20)           COMMENT "name",
        age                INT                   COMMENT "age"
    )
    DUPLICATE KEY(user_id)
    DISTRIBUTED BY HASH(user_id) BUCKETS 10;

  3. Запустите задание загрузки

    Задание загрузки Stream Load можно отправить с помощью команды curl.

    curl --location-trusted -u <doris_user>:<doris_password> \
        -H "label:124" \
        -H "Expect:100-continue" \
        -H "format:json" -H "strip_outer_array:true" \
        -H "jsonpaths:[\"$.userid\", \"$.username\", \"$.userage\"]" \
        -H "columns:user_id,name,age" \
        -T streamload_example.json \
        -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load
    Примечание

    Если файл JSON содержит не массив JSON, а по одному объекту JSON в каждой строке, добавьте заголовки -H "strip_outer_array:false" и -H "read_json_by_line:true". Чтобы загрузить объект JSON из корневого узла файла JSON, задайте для jsonpaths значение $., например: -H "jsonpaths:[\"$.\"]".

    Stream Load — это синхронный способ загрузки, при котором результат возвращается непосредственно пользователю.

    {
        "TxnId": 7,
        "Label": "125",
        "Comment": "",
        "TwoPhaseCommit": "false",
        "Status": "Success",
        "Message": "OK",
        "NumberTotalRows": 10,
        "NumberLoadedRows": 10,
        "NumberFilteredRows": 0,
        "NumberUnselectedRows": 0,
        "LoadBytes": 471,
        "LoadTimeMs": 52,
        "BeginTxnTimeMs": 0,
        "StreamLoadPutTimeMs": 11,
        "ReadDataTimeMs": 0,
        "WriteDataTimeMs": 23,
        "CommitAndPublishTimeMs": 16
    }

Просмотр заданий загрузки​

По умолчанию Stream Load возвращает результат клиенту синхронно, поэтому система не сохраняет историю заданий Stream Load. Чтобы включить её сохранение, добавьте настройку enable_stream_load_record=true в be.conf. Подробнее см. в разделе Конфигурация BE.

После настройки завершённые задачи Stream Load можно просматривать с помощью команды show stream load.

mysql> show stream load from testdb;
+-------+--------+-----------------+---------------+---------+---------+------+-----------+------------+--------------+----------------+-----------+-------------------------+-------------------------+------+---------+
| Label | Db     | Table           | ClientIp      | Status  | Message | Url  | TotalRows | LoadedRows | FilteredRows | UnselectedRows | LoadBytes | StartTime               | FinishTime              | User | Comment |
+-------+--------+-----------------+---------------+---------+---------+------+-----------+------------+--------------+----------------+-----------+-------------------------+-------------------------+------+---------+
| 12356 | testdb | test_streamload | 192.168.88.31 | Success | OK      | N/A  | 10        | 10         | 0            | 0              | 118       | 2023-11-29 08:53:00.594 | 2023-11-29 08:53:00.650 | root |         |
+-------+--------+-----------------+---------------+---------+---------+------+-----------+------------+--------------+----------------+-----------+-------------------------+-------------------------+------+---------+
1 row in set (0.00 sec)

Отмена задания загрузки​

Пользователи не могут отменить Stream Load вручную. Система автоматически отменяет Stream Load при превышении времени ожидания или возникновении ошибки загрузки.

Привязка Compute Group​

Можно указать Compute Group, в которой будет выполняться Stream Load.

Режим с разделением хранения и вычислений

В режиме с разделением хранения и вычислений Compute Group можно указать следующими способами:

  1. Укажите её через параметр заголовка HTTP.

    -H "cloud_cluster:cluster1"

    Начиная с Doris 4.0.0 также можно использовать параметр compute_group:

    -H "compute_group:cluster1"

    Если заданы и compute_group, и cloud_cluster, приоритет имеет compute_group.

  2. Укажите Compute Group в свойствах пользователя, от имени которого выполняется Stream Load. Если Compute Group указана и в свойствах пользователя, и в заголовке HTTP, приоритет имеет значение из заголовка.

    set property for user1 'default_compute_group'='cluster1';

  3. Если Compute Group не указана ни в свойствах пользователя, ни в заголовке HTTP, она выбирается из Compute Group, к которым имеет доступ пользователь, от имени которого выполняется Stream Load. Если у пользователя нет доступных Compute Group, загрузка завершается неудачно.

Примечание к версиям (4.0.8 / 4.1.4)

В Doris 4.0.8 и более поздних версиях серии 4.0, а также в Doris 4.1.4 и более поздних версиях серии 4.1 исправлено несоответствие между маршрутизацией в Compute Group и планом выполнения Stream Load в режиме с разделением вычислений и хранения:

  • Заголовок compute_group имеет приоритет над cloud_cluster. Ранее FE распознавал только cloud_cluster.
  • При построении плана выполнения используется вычислительная группа BE, который фактически получает запрос, поэтому план больше не назначается вычислительной группе, отличной от той, в которой он выполняется.
  • В этом процессе также проверяются права учётной записи загрузки на доступ к целевой вычислительной группе, наличие группы и её состояние. Поэтому, если у учётной записи загрузки нет доступа к вычислительной группе, фактически обслуживающей запрос, после обновления загрузка завершается с ошибкой прав доступа и учётной записи необходимо предоставить соответствующие права.
Примечание к версии (4.1.4)

Начиная с версии 4.1.4 конечные точки Stream Load на BE (PUT /api/{db}/{table}/_stream_load, PUT /api/_http_stream и PUT /api/{db}/_stream_load_2pc) больше не выполняют предварительную проверку глобальной привилегии LOAD. Проверка прав выполняется единообразно на FE для базы данных и таблицы, в которые фактически записываются данные.

Таким образом, учётная запись, у которой есть только LOAD_PRIV на уровне базы данных или таблицы (без глобальной привилегии LOAD), теперь также может выполнять Stream Load и операции commit / abort в рамках 2PC.

Режим с совмещением хранения и вычислений

В режиме с совмещением хранения и вычислений указать Compute Group можно только в свойствах пользователя, от имени которого выполняется Stream Load. Если Compute Group не указана в свойствах пользователя, выбирается Compute Group с именем default.

set property for user1 'resource_tags.location'='group_1';

Справочная информация​

Команда загрузки​

Синтаксис загрузки Stream Load:

curl --location-trusted -u <doris_user>:<doris_password> \
  -H "Expect:100-continue" [-H ""...] \
  -T <file_path> \
  -XPUT http://fe_host:http_port/api/{db}/{table}/_stream_load

Stream Load поддерживает загрузку по HTTP как с передачей по частям (chunked), так и без неё. Для загрузки без передачи по частям необходимо указать Content-Length, задающий длину передаваемого содержимого и обеспечивающий целостность данных.

Параметры конфигурации загрузки​

Конфигурация FE​

1. stream_load_default_timeout_second

ПараметрОписание
Значение по умолчанию259200 (s)
Динамическая настройкаДа
Настройка только для FE MasterДа

Описание: время ожидания Stream Load по умолчанию. Задаёт время ожидания задачи загрузки в секундах. Если задача загрузки не завершается за настроенное время timeout, система отменяет её, и задача переходит в состояние CANCELLED. Если исходный файл невозможно загрузить за указанное время, можно задать отдельное время ожидания в запросе Stream Load или изменить параметр FE stream_load_default_timeout_second, чтобы установить глобальное время ожидания по умолчанию.

Конфигурация BE​

1. streaming_load_max_mb

ПараметрОписание
Значение по умолчанию10240 (MB)
Динамическая настройкаДа

Описание: максимальный размер загрузки Stream Load. Если размер исходного файла пользователя превышает это значение, измените параметр BE streaming_load_max_mb.

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

Параметры загрузки можно передавать через заголовок HTTP. Ниже приведено их описание:

ТегОписание параметра
labelЗадаёт метку этой загрузки. Данные с одной и той же меткой нельзя загрузить более одного раза. Если метка не указана, Doris создаёт её автоматически. Указание метки позволяет избежать повторной загрузки одних и тех же данных. По умолчанию Doris хранит метки заданий загрузки в течение трёх дней; срок хранения можно изменить с помощью label_keep_max_second. Например, чтобы задать метку 123 для этой загрузки, используйте команду -H "label:123". Использование меток предотвращает повторную загрузку одних и тех же данных. Настоятельно рекомендуется использовать для одного и того же пакета данных одну и ту же метку, чтобы повторные запросы для одного пакета принимались только один раз, обеспечивая семантику At-Most-Once. Если задание загрузки с определённой меткой находится в состоянии CANCELLED, эту метку можно использовать повторно.
column_separatorЗадаёт разделитель столбцов в загружаемом файле. Значение по умолчанию — \t. Для невидимых символов используйте префикс \x и шестнадцатеричное представление разделителя. Разделителем столбцов также может быть сочетание нескольких символов. Например, разделитель файлов Hive \x01 задаётся командой -H "column_separator:\x01".
line_delimiterЗадаёт разделитель строк в загружаемом файле. Значение по умолчанию — \n. Разделителем строк также может быть сочетание нескольких символов. Например, чтобы задать \n в качестве разделителя строк, используйте команду -H "line_delimiter:\n".
columnsЗадаёт сопоставление столбцов загружаемого файла со столбцами таблицы. Если столбцы исходного файла полностью соответствуют столбцам таблицы, это поле можно не указывать. Если исходный файл не соответствует схеме таблицы, это поле требуется для преобразования данных. Столбец можно задать двумя способами: напрямую указать имя поля загружаемого файла либо описать вычисляемый столбец с помощью синтаксиса column_name = expression. Подробные примеры приведены в разделе Преобразование данных при загрузке.
whereИспользуется для выбора части данных. Если нужно исключить ненужные данные, используйте этот параметр. Например, чтобы загрузить только данные, у которых столбец k1 равен 20180601, укажите при загрузке -H "where: k1 = 20180601".
max_filter_ratioМаксимально допустимая доля данных, которые могут быть отфильтрованы, например из-за неверного формата. По умолчанию ошибки не допускаются. Диапазон значений — от 0 до 1. Если доля ошибок загрузки превышает это значение, загрузка завершается неудачно. Строки, отфильтрованные условием where, не учитываются как некорректные данные. Например, чтобы загрузить как можно больше корректных данных (допуская 100% ошибок), укажите команду -H "max_filter_ratio:1".
partitionsЗадаёт разделы, участвующие в этой загрузке. Если можно определить, каким разделам соответствуют данные, рекомендуется указать этот параметр. Данные, не соответствующие указанным разделам, отфильтровываются. Например, для загрузки в разделы p1 и p2 укажите команду -H "partitions: p1, p2".
timeoutЗадаёт время ожидания загрузки в секундах. Значение по умолчанию — 600 секунд. Допустимый диапазон — от 1 секунды до 259200 секунд. Например, чтобы установить время ожидания загрузки 1200 s, укажите команду -H "timeout:1200".
strict_modeОпределяет, включён ли строгий режим для этой загрузки. По умолчанию отключён. Например, чтобы включить строгий режим, укажите команду -H "strict_mode:true".
timezoneЗадаёт часовой пояс для этой загрузки. По умолчанию используется текущий часовой пояс кластера. Этот параметр влияет на результаты всех связанных с часовыми поясами функций, используемых при загрузке. Например, чтобы установить часовой пояс загрузки Africa/Abidjan, укажите команду -H "timezone:Africa/Abidjan".
exec_mem_limitОграничение памяти для загрузки. Значение по умолчанию — 2 GB. Единица измерения — байты.
formatЗадаёт формат данных для этой загрузки. Значение по умолчанию — CSV. В настоящее время поддерживаются следующие форматы: CSV, JSON, arrow, csv_with_names (позволяет пропустить первую строку файла CSV), csv_with_names_and_types (позволяет пропустить первые две строки файла CSV), Parquet и ORC. Например, чтобы указать JSON в качестве формата загружаемых данных, используйте команду -H "format:json".
jsonpathsДанные JSON можно загружать двумя способами: в простом режиме, когда jsonpaths не указан и данные JSON должны иметь тип объекта; и в режиме сопоставления, когда данные JSON имеют сложную структуру и соответствующие значения нужно сопоставлять через параметр jsonpaths. В простом режиме ключи JSON должны взаимно однозначно соответствовать именам столбцов таблицы. Например, для данных JSON {"k1":1, "k2":2, "k3":"hello"} ключи k1, k2 и k3 должны соответствовать столбцам таблицы.
strip_outer_arrayЕсли strip_outer_array имеет значение true, данные JSON начинаются с массива, который разворачивается в отдельные строки. Значение по умолчанию — false. Если внешний уровень данных JSON — это массив, представленный как [], установите strip_outer_array в true. Например, для данных [{"k1" : 1, "v1" : 2},{"k1" : 3, "v1" : 4}] после установки strip_outer_array в true при загрузке в Doris создаются две строки.
json_rootjson_root — допустимая строка jsonpath, задающая корневой узел документа JSON. Значение по умолчанию — "".
merge_typeТип слияния данных. Поддерживаются три типа:
- APPEND (по умолчанию): все данные этого пакета добавляются к существующим данным.
- DELETE: удаляет все строки, ключи которых совпадают с ключами в этом пакете.
- MERGE: должен использоваться вместе с условием delete. Данные, соответствующие условию delete, обрабатываются с семантикой DELETE, остальные — с семантикой APPEND.
Например, чтобы указать режим MERGE: -H "merge_type: MERGE" -H "delete: flag=1".
deleteИмеет смысл только при использовании MERGE; задаёт условие удаления данных.
function_column.sequence_colПрименяется только к модели UNIQUE KEYS. Для одинаковых значений ключевых столбцов обеспечивает замену значений столбцов операцией REPLACE в порядке, определяемом столбцом source_sequence. source_sequence может быть столбцом источника данных или столбцом в схеме таблицы.
fuzzy_parseЛогический параметр. Если установлено true, JSON разбирается с использованием первой строки в качестве схемы. Включение этого параметра может повысить эффективность загрузки JSON, но требует, чтобы порядок ключей во всех объектах JSON совпадал с порядком в первой строке. Значение по умолчанию — false. Используется только для формата JSON.
num_as_stringЛогический параметр. Если установлено true, при разборе данных JSON числовые типы преобразуются в строки, что обеспечивает загрузку без потери точности.
read_json_by_lineЛогический параметр. Если установлено true, поддерживается чтение по одному объекту JSON в каждой строке. Значение по умолчанию — false.
send_batch_parallelismЦелочисленный параметр. Задаёт степень параллелизма при отправке пакетов данных. Если значение превышает настройку BE max_send_batch_parallelism_per_job, BE, выполняющий роль координатора, использует значение max_send_batch_parallelism_per_job.
hidden_columnsЗадаёт скрытые столбцы, включённые в загружаемые данные. Действует, когда в заголовке не указан Columns. Несколько скрытых столбцов разделяются запятыми. Система загружает данные в соответствии с указаниями пользователя. В приведённом ниже примере последний столбец загружаемых данных — __DORIS_SEQUENCE_COL__. hidden_columns: __DORIS_DELETE_SIGN__,__DORIS_SEQUENCE_COL__
load_to_single_tabletЛогический параметр. Если установлено true, данные одной задачи можно загрузить только в один tablet соответствующего раздела. Значение по умолчанию — false. Этот параметр можно задать только при загрузке данных в таблицу OLAP со случайным распределением по бакетам.
compress_typeЗадаёт формат сжатия файла. В настоящее время поддерживается сжатие только файлов CSV. Поддерживаемые форматы сжатия: gz, lzo, bz2, lz4, lzop и deflate.
trim_double_quotesЛогический параметр. Значение по умолчанию — false. Если установлено true, удаляет внешние двойные кавычки у каждого поля файла CSV.
skip_linesЦелочисленный параметр. Значение по умолчанию — 0. Пропускает несколько первых строк файла CSV. Этот параметр не действует, если для format задано значение csv_with_names или csv_with_names_and_types.
commentСтроковый параметр. Значение по умолчанию — пустая строка. Добавляет к задаче дополнительную информацию.
encloseЗадаёт символ обрамления. Если поля данных CSV содержат разделители строк или столбцов, можно указать однобайтовый символ обрамления, чтобы предотвратить непреднамеренное разделение. Например, при разделителе столбцов , и символе обрамления ' в данных "a,'b,c'" значение b,c разбирается как одно поле. Примечание: если для enclose задано значение ", для trim_double_quotes необходимо установить true.
escapeЗадаёт символ экранирования. Используется для экранирования символов в полях, совпадающих с символом обрамления. Например, для данных "a,'b,'c'" и символа обрамления ', чтобы разобрать b,'c как одно поле, укажите однобайтовый символ экранирования, например \\, и измените данные на "a,'b,\\'c'".
memtable_on_sink_nodeОпределяет, включена ли пересылка MemTable при загрузке данных. Значение по умолчанию — false.
unique_key_update_modeРежим обновления таблицы Unique. В настоящее время действует только для таблиц Unique с Merge-On-Write. Поддерживаются три типа: UPSERT, UPDATE_FIXED_COLUMNS и UPDATE_FLEXIBLE_COLUMNS. UPSERT: загружает данные с семантикой upsert. UPDATE_FIXED_COLUMNS: загружает данные с помощью частичного обновления столбцов. UPDATE_FLEXIBLE_COLUMNS: загружает данные с помощью гибкого частичного обновления столбцов.
partial_update_new_key_behaviorСпособ обработки новых строк при частичном обновлении столбцов или гибком обновлении столбцов в таблице Unique. Доступны два типа: APPEND и ERROR.
- APPEND: разрешает вставку новых строк.
- ERROR: при вставке новой строки загрузка завершается неудачно и возвращает ошибку.

Результат загрузки​

Stream Load — это синхронный способ загрузки, при котором результат возвращается пользователю в ответе на запрос загрузки, как показано ниже:

{
    "TxnId": 1003,
    "Label": "b6f3bc78-0d2c-45d9-9e4c-faa0a0149bee",
    "Status": "Success",
    "ExistingJobStatus": "FINISHED", // optional
    "Message": "OK",
    "NumberTotalRows": 1000000,
    "NumberLoadedRows": 1000000,
    "NumberFilteredRows": 1,
    "NumberUnselectedRows": 0,
    "LoadBytes": 40888898,
    "LoadTimeMs": 2144,
    "BeginTxnTimeMs": 1,
    "StreamLoadPutTimeMs": 2,
    "ReadDataTimeMs": 325,
    "WriteDataTimeMs": 1933,
    "CommitAndPublishTimeMs": 106,
    "ErrorURL": "http://192.168.1.1:8042/api/_load_error_log?file=__shard_0/error_log_insert_stmt_db18266d4d9b4ee5-abb00ddd64bdf005_db18266d4d9b4ee5_abb00ddd64bdf005"
}

Параметры результата описаны в следующей таблице:

Имя параметраОписание
TxnIdИдентификатор транзакции загрузки.
LabelМетка задания загрузки, заданная с помощью -H "label:<label_id>".
StatusИтоговое состояние загрузки:
- Success: загрузка выполнена успешно.
- Publish Timeout: загрузка завершена, но данные могут стать видимыми с задержкой; повторная попытка не требуется.
- Label Already Exists: метка уже существует; требуется новая метка.
- Fail: загрузка завершилась неудачно.
ExistingJobStatusСостояние существующего задания загрузки с этой меткой. Это поле отображается только тогда, когда Status имеет значение Label Already Exists. По нему можно определить состояние задания загрузки с существующей меткой. RUNNING означает, что задание ещё выполняется; FINISHED означает, что задание выполнено успешно.
MessageСообщение об ошибке загрузки.
NumberTotalRowsОбщее количество строк, обработанных при загрузке.
NumberLoadedRowsКоличество успешно загруженных строк.
NumberFilteredRowsКоличество строк, не прошедших проверки качества данных.
NumberUnselectedRowsКоличество строк, отфильтрованных условием where.
LoadBytesКоличество загруженных байтов.
LoadTimeMsВремя выполнения загрузки в миллисекундах.
BeginTxnTimeMsВремя, затраченное на запрос к FE для начала транзакции, в миллисекундах.
StreamLoadPutTimeMsВремя, затраченное на запрос к FE для получения плана выполнения загрузки данных, в миллисекундах.
ReadDataTimeMsВремя, затраченное на чтение данных, в миллисекундах.
WriteDataTimeMsВремя, затраченное на запись данных, в миллисекундах.
CommitAndPublishTimeMsВремя, затраченное на запрос к FE для фиксации и публикации транзакции, в миллисекундах.
ErrorURLЕсли возникли проблемы с качеством данных, перейдите по этому URL, чтобы просмотреть конкретные строки с ошибками.

С помощью ErrorURL можно просмотреть данные, которые не удалось загрузить из-за низкого качества. Используйте команду curl "<ErrorURL>", чтобы напрямую получить информацию об ошибочных данных.

Примеры загрузки​

Настройка времени ожидания и максимального размера загрузки​

Время ожидания задачи загрузки задаётся в секундах. Если задача загрузки не завершается за настроенное время timeout, система отменяет её, и задача переходит в состояние CANCELLED. Время ожидания загрузки Stream Load можно изменить, указав параметр timeout или добавив параметр stream_load_default_timeout_second в fe.conf.

Перед загрузкой рассчитайте время ожидания исходя из размера файла. Например, для файла размером 100 GB при предполагаемой скорости загрузки 50 MB/s:

Import time ~= 100GB / 50MB/s ~= 2048s

Следующая команда создаёт задачу загрузки Stream Load, в которой для timeout задано значение 3000 s:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "timeout:3000" \
    -H "column_separator:," \
    -H "columns:user_id,name,age" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Настройка максимально допустимой доли ошибок​

Задачи загрузки Doris допускают наличие некоторой доли некорректных данных. Допустимая доля задаётся параметром max_filter_ratio. Значение по умолчанию — 0: если есть хотя бы одна строка с неверными данными, вся задача загрузки завершается неудачно. Если нужно пропустить часть проблемных строк, задайте для этого параметра значение от 0 до 1, и Doris автоматически пропустит строки с неверным форматом данных. Подробнее о расчёте допустимой доли см. в документации Преобразование данных.

Следующая команда создаёт задачу загрузки Stream Load, в которой для max_filter_ratio задано значение 0.4:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "max_filter_ratio:0.4" \
    -H "column_separator:," \
    -H "columns:user_id,name,age" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Настройка условий фильтрации при загрузке​

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

Следующая команда создаёт задачу загрузки Stream Load с условием фильтрации WHERE:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "where:age>=35" \
    -H "column_separator:," \
    -H "columns:user_id,name,age" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Загрузка данных в указанные разделы​

Загрузите данные из локального файла в разделы p1 и p2 таблицы, допуская долю ошибок 20%.

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "label:123" \
    -H "Expect:100-continue" \
    -H "max_filter_ratio:0.2" \
    -H "column_separator:," \
    -H "columns:user_id,name,age" \
    -H "partitions: p1, p2" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Указание часового пояса загрузки​

Типы, связанные с DATETIME, представляют только абсолютные моменты времени и не содержат информации о часовом поясе; они не изменяются при изменении системного часового пояса Doris. Поэтому данные с часовым поясом при загрузке обрабатываются единообразно: они преобразуются в данные указанного целевого часового пояса. В системе Doris часовой пояс задаётся переменной сеанса time_zone.

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

Например, если системный часовой пояс Doris — +08:00, а столбец времени в загружаемых данных содержит две строки, 2012-01-01 01:00:00+00:00 и 2015-12-12 12:12:12-08:00, то указание -H "timezone: +08:00" для транзакции загрузки преобразует обе строки в этот часовой пояс, давая 2012-01-01 09:00:00 и 2015-12-13 04:12:12.

Подробнее о часовых поясах см. в документации Часовой пояс.

Потоковая загрузка​

Stream Load основан на протоколе HTTP и поэтому поддерживает потоковую запись данных из программ на Java, Go и Python. Именно поэтому этот способ называется Stream Load.

Следующий пример с конвейером bash демонстрирует такой способ использования: загружаемые данные генерируются программой, а не считываются из локального файла.

seq 1 10 | awk '{OFS="\t"}{print $1, $1 * 10}' | curl --location-trusted -u root -T - http://host:port/api/testDb/testTbl/_stream_load

Пропуск первой строки CSV при загрузке​

Данные файла:

 id,name,age
 1,doris,20
 2,flink,10

Чтобы пропустить первую строку при загрузке, укажите format=csv_with_names:

curl --location-trusted -u root -T test.csv  -H "label:1" -H "format:csv_with_names" -H "column_separator:," http://host:port/api/testDb/testTbl/_stream_load

Выполнение операции Delete с помощью merge_type​

В Stream Load есть три типа загрузки: APPEND, DELETE и MERGE. Их можно выбирать с помощью параметра merge_type. Чтобы удалить все строки, ключи которых совпадают с загружаемыми данными, используйте следующую команду:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "merge_type: DELETE" \
    -H "column_separator:," \
    -H "columns:user_id,name,age" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Если до загрузки таблица содержит следующие данные:

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      3 |        2 | tom      |    2 |
|      4 |        3 | bush     |    3 |
|      5 |        3 | helen    |    3 |
+--------+----------+----------+------+

Загружаемые данные:

3,2,tom,0

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

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      4 |        3 | bush     |    3 |
|      5 |        3 | helen    |    3 |
+--------+----------+----------+------+

Выполнение операции Merge с помощью merge_type​

Если для merge_type задать значение MERGE, загружаемые данные объединяются с данными таблицы. Семантику MERGE необходимо использовать вместе с условием DELETE. Строки, соответствующие условию DELETE, обрабатываются с семантикой DELETE, а остальные добавляются в таблицу с семантикой APPEND. Например, следующая операция удаляет строку, у которой siteid равен 1, и добавляет остальные данные в таблицу:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "merge_type: MERGE" \
    -H "delete: siteid=1" \
    -H "column_separator:," \
    -H "columns:user_id,name,age" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Если до загрузки данные имеют следующий вид:

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      4 |        3 | bush     |    3 |
|      5 |        3 | helen    |    3 |
|      1 |        1 | jim      |    2 |
+--------+----------+----------+------+

Загружаемые данные:

2,1,grace,2
3,2,tom,2
1,1,jim,2

После загрузки строка с siteid = 1 удаляется, а строки, у которых siteid равен 2 и 3, добавляются в таблицу:

+--------+----------+----------+------+
| siteid | citycode | username | pv   |
+--------+----------+----------+------+
|      4 |        3 | bush     |    3 |
|      2 |        1 | grace    |    2 |
|      3 |        2 | tom      |    2 |
|      5 |        3 | helen    |    3 |
+--------+----------+----------+------+

Указание столбца Sequence для Merge при загрузке​

Если в таблице Unique Key задан столбец Sequence, для одинаковых значений ключевых столбцов значение Sequence определяет порядок замены, выполняемой агрегатной функцией REPLACE: большее значение может заменить меньшее. При пометке строк на удаление в такой таблице с помощью DORIS_DELETE_SIGN необходимо обеспечить совпадение ключа и значение столбца Sequence, большее или равное текущему. Для выполнения удаления можно указать параметр function_column.sequence_col вместе с merge_type: DELETE:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "merge_type: DELETE" \
    -H "function_column.sequence_col: age" \
    -H "column_separator:," \
    -H "columns: name, gender, age" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Предположим, таблица имеет следующую структуру:

mysql> SET show_hidden_columns=true;
Query OK, 0 rows affected (0.00 sec)

mysql> DESC table1;
+------------------------+--------------+------+-------+---------+---------+
| Field                  | Type         | Null | Key   | Default | Extra   |
+------------------------+--------------+------+-------+---------+---------+
| name                   | VARCHAR(100) | No   | true  | NULL    |         |
| gender                 | VARCHAR(10)  | Yes  | false | NULL    | REPLACE |
| age                    | INT          | Yes  | false | NULL    | REPLACE |
| __DORIS_DELETE_SIGN__  | TINYINT      | No   | false | 0       | REPLACE |
| __DORIS_SEQUENCE_COL__ | INT          | Yes  | false | NULL    | REPLACE |
+------------------------+--------------+------+-------+---------+---------+
4 rows in set (0.00 sec)

Предположим, исходная таблица содержит следующие данные:

+-------+--------+------+
| name  | gender | age  |
+-------+--------+------+
| li    | male   |   10 |
| wang  | male   |   14 |
| zhang | male   |   12 |
+-------+--------+------+

1. Параметр Sequence действует: загружаемое значение столбца Sequence больше или равно существующему значению в таблице

Загружаемые данные:

li,male,10

Поскольку указан function_column.sequence_col: age, а значение age больше или равно существующему значению столбца в таблице, исходная строка удаляется, и данные в таблице принимают следующий вид:

+-------+--------+------+
| name  | gender | age  |
+-------+--------+------+
| wang  | male   |   14 |
| zhang | male   |   12 |
+-------+--------+------+

2. Параметр Sequence не действует: загружаемое значение столбца Sequence меньше или равно существующему значению в таблице

Загружаемые данные:

li,male,9

Поскольку указан function_column.sequence_col: age, но значение age меньше существующего значения столбца в таблице, операция DELETE не действует; данные в таблице остаются неизменными, а строка с первичным ключом li по-прежнему видна:

+-------+--------+------+
| name  | gender | age  |
+-------+--------+------+
| li    | male   |   10 |
| wang  | male   |   14 |
| zhang | male   |   12 |
+-------+--------+------+

Строка не удаляется, потому что во внутреннем механизме обработки система сначала проверяет строки с одинаковым ключом, делает видимой строку с большим значением столбца Sequence, а затем проверяет, равно ли значение DORIS_DELETE_SIGN этой строки 1. Если оно равно 1, строка скрывается; если равно 0, она остаётся доступной для чтения.

Загрузка данных с символами обрамления​

Если данные CSV содержат разделители или разделители столбцов, можно указать однобайтовый символ обрамления, чтобы предотвратить разделение данных.

В следующих данных столбец содержит разделитель ,:

Zhang San,30,'Shanghai City, Huangpu District, Dagu Road'

Указав символ обрамления ', можно представить "Shanghai City, Huangpu District, Dagu Road" как одно поле:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "column_separator:," \
    -H "enclose:'" \
    -H "columns:username,age,address" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Если символ обрамления встречается и внутри поля, например, если нужно обработать "Shanghai City, Huangpu District, 'Dagu Road" как одно поле, сначала экранируйте этот символ в столбце:

Zhang San,30,'Shanghai City, Huangpu District, \'Dagu Road'

Параметр escape позволяет задать однобайтовый символ экранирования, например \, как в следующем примере:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "column_separator:," \
    -H "enclose:'" \
    -H "escape:\\" \
    -H "columns:username,age,address" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Загрузка данных в таблицу с полями DEFAULT CURRENT_TIMESTAMP​

Следующий пример показывает загрузку данных в таблицу, поля которой содержат DEFAULT CURRENT_TIMESTAMP.

Схема таблицы:

CREATE TABLE testDb.testTbl (
    `id` BIGINT(30) NOT NULL,
    `order_code` VARCHAR(30) DEFAULT NULL COMMENT '',
    `create_time` DATETIMEv2(3) DEFAULT CURRENT_TIMESTAMP
)
DUPLICATE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 10;

Формат данных JSON:

{"id":1,"order_Code":"avc"}

Команда загрузки:

curl --location-trusted -u root -T test.json -H "label:1" -H "format:json" -H 'columns: id, order_code, create_time=CURRENT_TIMESTAMP()' http://host:port/api/testDb/testTbl/_stream_load

Загрузка данных JSON в простом режиме​

Если поля JSON взаимно однозначно соответствуют именам столбцов таблицы, данные JSON можно загрузить в таблицу, указав "strip_outer_array:true" и "format:json".

Например, таблица определена следующим образом:

CREATE TABLE testdb.test_streamload(
    user_id            BIGINT       NOT NULL COMMENT "user id",
    name               VARCHAR(20)           COMMENT "name",
    age                INT                   COMMENT "age"
)
DUPLICATE KEY(user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 10;

Имена полей в загружаемых данных взаимно однозначно соответствуют именам полей таблицы:

[
{"user_id":1,"name":"Emily","age":25},
{"user_id":2,"name":"Benjamin","age":35},
{"user_id":3,"name":"Olivia","age":28},
{"user_id":4,"name":"Alexander","age":60},
{"user_id":5,"name":"Ava","age":17},
{"user_id":6,"name":"William","age":69},
{"user_id":7,"name":"Sophia","age":32},
{"user_id":8,"name":"James","age":64},
{"user_id":9,"name":"Emma","age":37},
{"user_id":10,"name":"Liam","age":64}
]

Загрузить данные JSON в таблицу можно следующей командой:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "format:json" \
    -H "strip_outer_array:true" \
    -T streamload_example.json \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Загрузка сложных данных JSON в режиме сопоставления​

Если данные JSON имеют сложную структуру и не могут взаимно однозначно соответствовать именам столбцов таблицы или содержат дополнительные столбцы, можно указать параметр jsonpaths, чтобы задать сопоставление имён столбцов для загрузки в режиме сопоставления. Например, даны следующие данные:

[
{"userid":1,"hudi":"lala","username":"Emily","userage":25,"userhp":101},
{"userid":2,"hudi":"kpkp","username":"Benjamin","userage":35,"userhp":102},
{"userid":3,"hudi":"ji","username":"Olivia","userage":28,"userhp":103},
{"userid":4,"hudi":"popo","username":"Alexander","userage":60,"userhp":103},
{"userid":5,"hudi":"uio","username":"Ava","userage":17,"userhp":104},
{"userid":6,"hudi":"lkj","username":"William","userage":69,"userhp":105},
{"userid":7,"hudi":"komf","username":"Sophia","userage":32,"userhp":106},
{"userid":8,"hudi":"mki","username":"James","userage":64,"userhp":107},
{"userid":9,"hudi":"hjk","username":"Emma","userage":37,"userhp":108},
{"userid":10,"hudi":"hua","username":"Liam","userage":64,"userhp":109}
]

Указав параметр jsonpaths, можно сопоставить нужные столбцы:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "format:json" \
    -H "strip_outer_array:true" \
    -H "jsonpaths:[\"$.userid\", \"$.username\", \"$.userage\"]" \
    -H "columns:user_id,name,age" \
    -T streamload_example.json \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Указание корневого узла JSON для загрузки​

Если данные JSON содержат вложенные поля JSON, необходимо указать корневой узел JSON, из которого нужно загрузить данные. Значение по умолчанию — "".

Для следующих данных требуется загрузить в таблицу содержимое столбца comment:

[
    {"user":1,"comment":{"userid":101,"username":"Emily","userage":25}},
    {"user":2,"comment":{"userid":102,"username":"Benjamin","userage":35}},
    {"user":3,"comment":{"userid":103,"username":"Olivia","userage":28}},
    {"user":4,"comment":{"userid":104,"username":"Alexander","userage":60}},
    {"user":5,"comment":{"userid":105,"username":"Ava","userage":17}},
    {"user":6,"comment":{"userid":106,"username":"William","userage":69}},
    {"user":7,"comment":{"userid":107,"username":"Sophia","userage":32}},
    {"user":8,"comment":{"userid":108,"username":"James","userage":64}},
    {"user":9,"comment":{"userid":109,"username":"Emma","userage":37}},
    {"user":10,"comment":{"userid":110,"username":"Liam","userage":64}}
    ]

Сначала с помощью параметра json_root задайте корневой узел comment, а затем используйте параметр jsonpaths для сопоставления имён столбцов:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "format:json" \
    -H "strip_outer_array:true" \
    -H "json_root: $.comment" \
    -H "jsonpaths:[\"$.userid\", \"$.username\", \"$.userage\"]" \
    -H "columns:user_id,name,age" \
    -T streamload_example.json \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Загрузка данных типа Array​

Следующие данные содержат тип массива:

1|Emily|[1,2,3,4]
2|Benjamin|[22,45,90,12]
3|Olivia|[23,16,19,16]
4|Alexander|[123,234,456]
5|Ava|[12,15,789]
6|William|[57,68,97]
7|Sophia|[46,47,49]
8|James|[110,127,128]
9|Emma|[19,18,123,446]
10|Liam|[89,87,96,12]

Загрузите данные в таблицу со следующей схемой:

CREATE TABLE testdb.test_streamload(
    typ_id     BIGINT          NOT NULL COMMENT "ID",
    name       VARCHAR(20)     NULL     COMMENT "name",
    arr        ARRAY<int(10)>  NULL     COMMENT "array"
)
DUPLICATE KEY(typ_id)
DISTRIBUTED BY HASH(typ_id) BUCKETS 10;

С помощью задачи Stream Load данные типа ARRAY можно загрузить из текстового файла непосредственно в таблицу:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "column_separator:|" \
    -H "columns:typ_id,name,arr" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Загрузка данных типа Map​

Если загружаемые данные содержат тип Map, как в следующем примере:

[
{"user_id":1,"namemap":{"Emily":101,"age":25}},
{"user_id":2,"namemap":{"Benjamin":102,"age":35}},
{"user_id":3,"namemap":{"Olivia":103,"age":28}},
{"user_id":4,"namemap":{"Alexander":104,"age":60}},
{"user_id":5,"namemap":{"Ava":105,"age":17}},
{"user_id":6,"namemap":{"William":106,"age":69}},
{"user_id":7,"namemap":{"Sophia":107,"age":32}},
{"user_id":8,"namemap":{"James":108,"age":64}},
{"user_id":9,"namemap":{"Emma":109,"age":37}},
{"user_id":10,"namemap":{"Liam":110,"age":64}}
]

Загрузите данные в таблицу со следующей схемой:

CREATE TABLE testdb.test_streamload(
    user_id            BIGINT       NOT NULL COMMENT "ID",
    namemap            Map<STRING, INT>  NULL     COMMENT "namemap"
)
DUPLICATE KEY(user_id)
DISTRIBUTED BY HASH(user_id) BUCKETS 10;

С помощью задачи Stream Load данные типа Map можно загрузить из текстового файла непосредственно в таблицу:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "format: json" \
    -H "strip_outer_array:true" \
    -T streamload_example.json \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Загрузка данных типа Bitmap​

Если при загрузке встречаются данные типа Bitmap, можно использовать to_bitmap для преобразования данных в Bitmap или функцию bitmap_empty для заполнения Bitmap.

Например, загружаемые данные имеют следующий вид:

1|koga|17723
2|nijg|146285
3|lojn|347890
4|lofn|489871
5|jfin|545679
6|kon|676724
7|nhga|767689
8|nfubg|879878
9|huang|969798
10|buag|97997

Загрузите данные в следующую таблицу, содержащую тип Bitmap:

CREATE TABLE testdb.test_streamload(
    typ_id     BIGINT                NULL   COMMENT "ID",
    hou        VARCHAR(10)           NULL   COMMENT "one",
    arr        BITMAP  BITMAP_UNION  NOT NULL   COMMENT "two"
)
AGGREGATE KEY(typ_id,hou)
DISTRIBUTED BY HASH(typ_id,hou) BUCKETS 10;

С помощью to_bitmap можно преобразовать данные в тип Bitmap:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "Expect:100-continue" \
    -H "columns:typ_id,hou,arr,arr=to_bitmap(arr)" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Загрузка данных типа HLL​

Для преобразования данных в тип HLL можно использовать функцию hll_hash. Даны следующие данные:

1001|koga
1002|nijg
1003|lojn
1004|lofn
1005|jfin
1006|kon
1007|nhga
1008|nfubg
1009|huang
1010|buag

Загрузите их в следующую таблицу:

CREATE TABLE testdb.test_streamload(
    typ_id           BIGINT          NULL   COMMENT "ID",
    typ_name         VARCHAR(10)     NULL   COMMENT "NAME",
    pv               hll hll_union   NOT NULL   COMMENT "hll"
)
AGGREGATE KEY(typ_id,typ_name)
DISTRIBUTED BY HASH(typ_id) BUCKETS 10;

Выполните загрузку с помощью функции hll_hash:

curl --location-trusted -u <doris_user>:<doris_password> \
    -H "column_separator:|" \
    -H "columns:typ_id,typ_name,pv=hll_hash(typ_id)" \
    -T streamload_example.csv \
    -XPUT http://<fe_ip>:<fe_http_port>/api/testdb/test_streamload/_stream_load

Сопоставление столбцов, вычисляемые столбцы и фильтрация​

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

Включение строгого режима загрузки​

Свойство strict_mode определяет, выполняется ли задача загрузки в строгом режиме. Оно влияет на результаты сопоставления, преобразования и фильтрации столбцов. Подробнее о строгом режиме см. в документации Строгий режим.

Частичное обновление столбцов или гибкое частичное обновление столбцов при загрузке​

О том, как задать частичное обновление столбцов при загрузке, см. в документации Обновление столбцов.

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

Поддерживает ли Stream Load отмену вручную?​

Нет. Система автоматически отменяет Stream Load при превышении времени ожидания или возникновении ошибки загрузки; пользователи не могут вручную отменить выполняющееся задание Stream Load.

Есть ли ограничение размера отдельного файла для Stream Load?​

Рекомендуется, чтобы размер отдельного загружаемого файла не превышал 10 GB. Более крупные файлы разделите перед загрузкой. Также можно изменить параметр конфигурации BE streaming_load_max_mb, чтобы увеличить ограничение на размер одной загрузки, или использовать Doris Streamloader для параллельной загрузки с целью повышения производительности.

Как просмотреть историю заданий Stream Load?​

По умолчанию система не сохраняет историю заданий Stream Load. Задайте enable_stream_load_record=true в be.conf, выполните перезапуск, а затем используйте show stream load from <db> для просмотра истории.

Как обработать ошибку "Label Already Exists"?​

Эта ошибка означает, что такая метка уже использовалась. Состояние существующего задания можно определить по полю ExistingJobStatus в ответе:

  • RUNNING: задание ещё выполняется; дождитесь его завершения.
  • FINISHED: задание успешно завершено; повторная попытка не требуется.
  • Если состояние — CANCELLED, ту же метку можно использовать для запуска новой загрузки.

Рекомендуется всегда использовать одну и ту же метку для одного и того же пакета данных, чтобы обеспечить семантику At-Most-Once и избежать повторной загрузки.

Что делать, если загрузка завершилась неудачно из-за ошибок формата данных?​

Найти и устранить проблему можно следующим образом:

  1. Просмотрите конкретные строки с ошибками через поле ErrorURL в ответе: curl "<ErrorURL>".
  2. При необходимости увеличьте max_filter_ratio (диапазон от 0 до 1), чтобы допустить некоторое количество некорректных данных.
  3. Проверьте, правильно ли в CSV используются значения null (\N) и пустые строки (отсутствие символов между двумя разделителями).
  4. Проверьте, соответствуют ли настройки разделителя столбцов, разделителя строк, символа обрамления (enclose) и символа экранирования (escape) фактическому содержимому файла.

Что делать при превышении времени ожидания загрузки (CANCELLED)?​

Оцените подходящее время ожидания исходя из объёма данных:

Import time ~= File size / Estimated import speed (for example, 50MB/s)

Изменить время ожидания можно двумя способами:

  • Для отдельной загрузки: используйте параметр заголовка -H "timeout:3000".
  • Глобально: измените параметр FE stream_load_default_timeout_second.

Как различать значения null и пустые строки в CSV?​

ТипПредставлениеПримерЗначение среднего столбца
Значение nullИспользуйте \Na,\N,bСредний столбец содержит null
Пустая строкаМежду двумя разделителями нет символовa,,bСредний столбец содержит пустую строку
Report issue
На этой странице