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

CREATE ROUTINE LOAD

Описание​

Функция Routine Load позволяет пользователям создавать постоянно действующую задачу загрузки, которая непрерывно считывает данные из указанного источника и загружает их в Doris.

В настоящее время поддерживается только загрузка данных в формате CSV или JSON из Kafka без аутентификации или с аутентификацией по SSL. Пример загрузки данных в формате JSON

Синтаксис​

CREATE ROUTINE LOAD [<db>.]<job_name> [ON <tbl_name>]
[<merge_type>]
[<load_properties>]
[<job_properties>]
FROM <data_source> [<data_source_properties>]
[COMMENT "<comment>"]

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

1. [<db>.]<job_name>

Имя задания загрузки. В одной базе данных одновременно может выполняться только одно задание с таким именем.

2. FROM <data_source>

Тип источника данных. В настоящее время поддерживается: KAFKA

3. <data_source_properties>

  1. <kafka_broker_list>

    Данные для подключения к брокерам Kafka. Формат: ip:host. Несколько брокеров разделяются запятыми.

    "kafka_broker_list" = "broker1:9092,broker2:9092"

  2. <kafka_topic>

    Задает топик Kafka для подписки.

    "kafka_topic" = "my_topic"

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

1. <tbl_name>

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

В настоящее время имя таблицы можно получить только из Value сообщения Kafka. Данные должны соответствовать следующему формату, например для JSON: table_name|{"col1": "val1", "col2": "val2"}, где tbl_name — имя таблицы, а | — разделитель между именем таблицы и ее данными.

Для данных в формате CSV используется аналогичный формат: table_name|val1,val2,val3. Обратите внимание: table_name здесь должно совпадать с именем таблицы в Doris, иначе загрузка завершится ошибкой.

Совет: динамические таблицы не поддерживают параметр columns_mapping. Если структура ваших таблиц соответствует структуре таблиц в Doris и требуется загрузить данные для большого количества таблиц, этот способ будет наиболее подходящим.

2. <merge_type>

Тип слияния данных. По умолчанию используется APPEND: загружаемые данные записываются обычными операциями добавления. Типы MERGE и DELETE доступны только для таблиц модели Unique Key. Тип MERGE необходимо использовать вместе с предложением [DELETE ON], чтобы задать столбец Delete Flag. Тип DELETE означает, что все загружаемые данные обозначают удаление.

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

3. <load_properties>

Описывает загружаемые данные. Состав параметра:

[column_separator],
[columns_mapping],
[preceding_filter],
[where_predicates],
[partitions],
[DELETE ON],
[ORDER BY]

  1. <column_separator>

    Задает разделитель столбцов. Значение по умолчанию: \t

    COLUMNS TERMINATED BY ","

  2. <columns_mapping>

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

    (k1, k2, tmpk1, k3 = tmpk1 + 1)

    Совет: динамические таблицы не поддерживают этот параметр.

  3. <preceding_filter>

    Фильтрует исходные данные. Подробное описание приведено в документе [Сопоставление, преобразование и фильтрация столбцов].

    WHERE k1 > 100 and k2 = 1000

    Совет: динамические таблицы не поддерживают этот параметр.

  4. <where_predicates>

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

    WHERE k1 > 100 and k2 = 1000

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

  5. <partitions>

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

    PARTITION(p1, p2, p3)

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

  6. <DELETE ON>

    Должен использоваться с режимом загрузки MERGE и применим только к таблицам модели Unique Key. Задает столбец Delete Flag и выражение для его вычисления в загружаемых данных.

    DELETE ON v3 >100

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

  7. <ORDER BY>

    Применим только к таблицам модели Unique Key. Задает столбец Sequence Col в загружаемых данных. В основном используется для обеспечения порядка данных при загрузке.

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

4. <job_properties>

Задает общие параметры заданий Routine Load.

PROPERTIES (
    "key1" = "val1",
    "key2" = "val2"
)

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

  1. <desired_concurrent_number>

    Желаемое число параллельно выполняемых задач. Задание Routine Load разделяется на несколько подзадач. Этот параметр задает, сколько задач одного задания могут выполняться одновременно. Значение должно быть больше 0. Значение по умолчанию: 5.

    Это число не является фактическим числом параллельно выполняемых задач. Фактическое значение определяется с учетом количества узлов кластера, нагрузки и состояния источника данных.

    "desired_concurrent_number" = "3"

  2. <max_batch_interval>/<max_batch_rows>/<max_batch_size>

    Эти три параметра задают:

    1. Максимальное время выполнения каждой подзадачи в секундах. Значение должно быть больше или равно 1. Значение по умолчанию: 10.
    2. Максимальное количество строк, считываемых каждой подзадачей. Значение должно быть больше или равно 200000. Значение по умолчанию: 20000000.
    3. Максимальное количество байтов, считываемых каждой подзадачей. Единица измерения — байты, диапазон — от 100MB до 10GB. Значение по умолчанию: 1G.

    Эти три параметра управляют временем выполнения и объемом обработки подзадачи. Когда любой из них достигает порогового значения, задача завершается.

    "max_batch_interval" = "20",
    "max_batch_rows" = "300000",
    "max_batch_size" = "209715200"

  3. <max_error_number>

    Максимальное допустимое количество ошибочных строк в окне выборки. Значение должно быть больше или равно 0. Значение по умолчанию: 0, то есть ошибочные строки не допускаются.

    Размер окна выборки составляет max_batch_rows * 10. Если количество ошибочных строк в окне выборки превышает max_error_number, задание Routine Load приостанавливается и требует ручной проверки проблем с качеством данных.

    Строки, отфильтрованные условиями where, не считаются ошибочными.

  4. <strict_mode>

    Определяет, включен ли строгий режим. По умолчанию он отключен. Если режим включен и преобразование типа столбца превращает исходное значение, отличное от NULL, в NULL, такая строка отфильтровывается. Задается следующим образом:

    "strict_mode" = "true"

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

    1. Если строгий режим установлен в true, при преобразовании типов столбцов ошибочные данные отфильтровываются. Здесь под ошибочными понимаются исходные данные, которые не равны NULL, но после преобразования типа столбца дают значение NULL.
    2. Строгий режим не влияет на столбцы, значения которых формируются с помощью функций преобразования при загрузке.
    3. Для столбцов с ограничениями диапазона строгий режим не влияет на данные, которые проходят преобразование типа, но не соответствуют ограничениям диапазона. Например, если тип — decimal(1,0), а исходное значение — 10, оно проходит преобразование типа, но выходит за объявленный диапазон столбца. Строгий режим на такие данные не влияет.

    Связь строгого режима с загрузкой исходных данных

    Пример для столбца типа TinyInt

    Примечание: столбцы таблицы допускают значения NULL

    исходные данныепример исходных данныхstring в intstrict_modeрезультат
    null\NN/Atrue или falseNULL
    not nullaaa или 2000NULLtrueнекорректные данные (отфильтрованы)
    not nullaaaNULLfalseNULL
    not null11true или falseкорректные данные

    Пример для столбца типа Decimal(1,0)

    Примечание: столбцы таблицы допускают значения NULL

    исходные данныепример исходных данныхstring в intstrict_modeрезультат
    null\NN/Atrue или falseNULL
    not nullaaaNULLtrueнекорректные данные (отфильтрованы)
    not nullaaaNULLfalseNULL
    not null1 или 101true или falseкорректные данные

    Примечание: хотя значение 10 выходит за допустимый диапазон, его тип соответствует требованиям decimal, поэтому строгий режим на него не влияет. Значение 10 в итоге будет отфильтровано на других этапах обработки ETL, но не в строгом режиме.

  5. <timezone>

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

    "timezone" = "Asia/Shanghai"

  6. <format>

    Задает формат загружаемых данных. По умолчанию используется CSV, также поддерживается JSON.

    "format" = "json"

  7. <jsonpaths>

    При загрузке данных в формате JSON параметр jsonpaths позволяет указать поля, которые нужно извлечь из данных JSON.

    -H "jsonpaths: [\"$.k2\", \"$.k1\"]"

  8. <strip_outer_array>

    При загрузке данных в формате JSON значение true параметра strip_outer_array указывает, что данные JSON представлены массивом, каждый элемент которого обрабатывается как отдельная строка. Значение по умолчанию: false.

    -H "strip_outer_array: true"

  9. <json_root>

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

    -H "json_root: $.RECORDS"

  10. <send_batch_parallelism>

    Целочисленный параметр, задающий степень параллелизма при отправке пакетов данных. Если значение превышает max_send_batch_parallelism_per_job в конфигурации BE, узел BE, выполняющий роль координатора, использует значение max_send_batch_parallelism_per_job.

    "send_batch_parallelism" = "10"

  11. <load_to_single_tablet>

    Логический параметр. Значение true позволяет задаче загружать данные только в один tablet соответствующего раздела. Значение по умолчанию: false. Этот параметр разрешено задавать только при загрузке данных в таблицы olap со случайным распределением по бакетам.

    "load_to_single_tablet" = "true"

  12. <partial_columns>

    Логический параметр. Значение true включает частичное обновление столбцов. Значение по умолчанию: false. Этот параметр разрешено задавать только для таблиц модели Unique, использующих Merge on Write. Динамическая загрузка в несколько таблиц не поддерживает этот параметр.

    "partial_columns" = "true"

  13. <max_filter_ratio>

    Максимальная допустимая доля отфильтрованных строк в окне выборки. Значение должно быть больше или равно 0 и меньше или равно 1. Значение по умолчанию: 0.

    Размер окна выборки составляет max_batch_rows * 10. Если в окне выборки отношение количества ошибочных строк к общему количеству строк превышает max_filter_ratio, задание Routine Load приостанавливается и требует ручной проверки проблем с качеством данных.

    Строки, отфильтрованные условиями where, не считаются ошибочными.

  14. <enclose>

    Символ обрамления. Если поля данных CSV содержат разделители строк или столбцов, для защиты от случайного разделения можно задать однобайтовый символ обрамления. Например, если разделитель столбцов — ",", а символ обрамления — "'", то в данных "a,'b,c'" значение "b,c" будет разобрано как одно поле.

    Примечание: если enclose установлен в ", trim_double_quotes должен быть установлен в true.

  15. <escape>

    Символ экранирования. Используется для экранирования символов в полях CSV, совпадающих с символом обрамления. Например, если данные имеют вид "a,'b,'c'", символ обрамления — "'", и требуется разобрать "b,'c" как одно поле, нужно указать однобайтовый символ экранирования, например \, и изменить данные на a,'b,\'c'.

5. Необязательные свойства в data_source_properties

  1. <kafka_partitions>/<kafka_offsets>

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

    В качестве offset можно указать конкретное смещение, большее или равное 0, либо:

    • OFFSET_BEGINNING: начать подписку с начала имеющихся данных.
    • OFFSET_END: начать подписку с конца.
    • Время, например: "2021-05-22 11:00:00"

    Если параметр не указан, по умолчанию выполняется подписка на все разделы топика с OFFSET_END.

    "kafka_partitions" = "0,1,2,3",
    "kafka_offsets" = "101,0,OFFSET_BEGINNING,OFFSET_END"
    "kafka_partitions" = "0,1,2,3",
    "kafka_offsets" = "2021-05-22 11:00:00,2021-05-22 11:00:00,2021-05-22 11:00:00"

    Примечание: формат времени нельзя смешивать с форматом OFFSET.

  2. <property>

    Задает пользовательские параметры Kafka. Работает так же, как параметр "--property" в оболочке Kafka.

    Если значением параметра является файл, перед значением нужно добавить ключевое слово "FILE:".

    О создании файлов см. документацию команды CREATE FILE.

    Другие поддерживаемые пользовательские параметры приведены среди настроек клиента в официальной документации CONFIGURATION библиотеки librdkafka. Например:

    "property.client.id" = "12345",
    "property.ssl.ca.location" = "FILE:ca.pem"

    2.1 При подключении к Kafka по SSL необходимо указать следующие параметры:

    ```text
    "property.security.protocol" = "ssl",
    "property.ssl.ca.location" = "FILE:ca.pem",
    "property.ssl.certificate.location" = "FILE:client.pem",
    "property.ssl.key.location" = "FILE:client.key",
    "property.ssl.key.password" = "abcdefg"
    ```

    Among them:

    `property.security.protocol` and `property.ssl.ca.location` are required, used to specify the connection method as SSL and the location of the CA certificate.

    If client authentication is enabled on the Kafka server side, the following also need to be set:

    ```text
    "property.ssl.certificate.location"
    "property.ssl.key.location"
    "property.ssl.key.password"
    ```

    Used to specify the client's public key, private key, and private key password respectively.

    2.2 Задание начального смещения по умолчанию для разделов Kafka

    Если <kafka_partitions>/<kafka_offsets> не указан, по умолчанию данные считываются из всех разделов.

    В этом случае начальное смещение можно задать с помощью <kafka_default_offsets>. Значение по умолчанию — OFFSET_END, то есть подписка начинается с конца.

    Пример:

    "property.kafka_default_offsets" = "OFFSET_BEGINNING"

6. COMMENT

Comment information for the routine load task.

Требования к правам доступа​

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

ПравоОбъектПримечания
LOAD_PRIVТаблицаCREATE ROUTINE LOAD относится к операции LOAD для таблицы

Примечания по использованию​

  • Динамические таблицы не поддерживают параметр columns_mapping
  • При динамической загрузке в несколько таблиц такие параметры, как merge_type, where_predicates и другие, должны соответствовать требованиям каждой динамической таблицы
  • Формат времени нельзя смешивать с форматом OFFSET
  • kafka_partitions и kafka_offsets должны соответствовать друг другу один к одному
  • Если enclose установлен в ", trim_double_quotes должен быть установлен в true.

Примеры​

Полные примеры загрузки CSV/JSON из Kafka, аутентификации по SSL, режима Merge, столбцов последовательности и других возможностей приведены в разделе Routine Load руководства по загрузке данных.

Report issue