SHOW ROUTINE LOAD

SHOW ROUTINE LOAD показывает информацию о выполнении задач Routine Load.

Вы можете управлять задачами Routine Load, которые выполняются на таблицах Selena, только как пользователь, имеющий привилегию INSERT на эти таблицы Selena. Если у вас нет привилегии INSERT, следуйте инструкциям в разделе GRANT, чтобы предоставить привилегию INSERT пользователю, которого вы используете для подключения к вашему cluster Selena.

Синтаксис

SHOW [ALL] ROUTINE LOAD [ FOR [<db_name>.]<job_name> | FROM <db_name> ]
[ WHERE [ STATE = { "NEED_SCHEDULE" | "RUNNING" | "PAUSED" | "UNSTABLE" | "STOPPED" | "CANCELLED"  } ] ]
[ ORDER BY field_name [ ASC | DESC ] ]
[ LIMIT { [offset, ] limit | limit OFFSET offset } ]

подсказка

Вы можете добавить опцию \G к оператору (например, SHOW ROUTINE LOAD FOR <job_name>\G), чтобы вертикально отобразить результат вместо обычного горизонтального табличного формата.

Параметры

Параметр	Обязательный	Описание
db_name	Нет	Имя базы данных, к которой принадлежит задача загрузки. Обратите внимание, что этот параметр обязателен, если используется предложение FROM.
job_name	Нет	Имя задачи загрузки. Обратите внимание, что этот параметр обязателен, если используется предложение FOR.
ALL	Нет	Отображает все задачи загрузки, включая те, которые находятся в состояниях STOPPED или CANCELLED.
STATE	Нет	Статус задачи загрузки.
ORDER BY field_name [ASC | DESC]	Нет	Сортирует результат в порядке возрастания или убывания на основе указанного поля. Поддерживаются следующие поля: Id, Name, CreateTime, PauseTime, EndTime, TableName, State и CurrentTaskNum. Чтобы отсортировать результат в порядке возрастания, укажите ORDER BY field_name ASC. Чтобы отсортировать результат в порядке убывания, укажите ORDER BY field_name DESC. Если вы не укажете поле или порядок сортировки, результат по умолчанию сортируется в порядке возрастания Id.
LIMIT limit	Нет	Количество задач загрузки, которые будут возвращены. Например, если указано LIMIT 10, будет возвращена информация только о 10 задачах загрузки, соответствующих условиям фильтрации. Если этот параметр не указан, отображается информация обо всех задачах загрузки, соответствующих условиям фильтрации.
OFFSET offset	Нет	Параметр offset определяет количество задач загрузки, которые будут пропущены. Например, OFFSET 5 пропускает первые пять задач загрузки и возвращает остальные. Значение параметра offset по умолчанию равно 0.

Результат

Параметр	Описание
Id	Глобально уникальный ID задачи загрузки, генерируется автоматически Selena.
Name	Имя задачи загрузки.
CreateTime	Дата и время создания задачи загрузки.
PauseTime	Дата и время, когда задача загрузки перешла в состояние PAUSED.
EndTime	Дата и время, когда задача загрузки перешла в состояние STOPPED.
DbName	База данных, к которой принадлежит целевая таблица задачи загрузки.
TableName	Целевая таблица задачи загрузки.
State	Статус задачи загрузки, включая: NEED_SCHEDULE: Задача загрузки ожидает планирования. После использования CREATE ROUTINE LOAD или RESUME ROUTINE LOAD для создания или возобновления задачи Routine Load, задача загрузки сначала переходит в состояние NEED_SCHEDULE. RUNNING: Задача загрузки выполняется. Вы можете просмотреть прогресс потребления задачи Routine Load через Statistic и Progress. PAUSED: Задача загрузки приостановлена. Вы можете обратиться к ReasonOfStateChanged и ErrorLogUrls для устранения неполадок. После исправления ошибки вы можете использовать RESUME ROUTINE LOAD для возобновления задачи Routine Load. CANCELLED: Задача загрузки отменена. Вы можете обратиться к ReasonOfStateChanged и ErrorLogUrls для устранения неполадок. Однако после исправления ошибки вы не сможете восстановить задачу загрузки в этом состоянии. STOPPED: Задача загрузки остановлена. Вы не можете восстановить задачу загрузки в этом состоянии. UNSTABLE: Задача загрузки нестабильна. Задача Routine Load переводится в состояние UNSTABLE, если какая-либо подзадача в рамках задачи Routine Load отстаёт (а именно, разница между временной меткой потребляемого сообщения и текущим временем превышает этот параметр FE routine_load_unstable_threshold_second, и в источнике данных существуют непотреблённые сообщения).
DataSourceType	Тип источника данных. Фиксированное значение: KAFKA.
CurrentTaskNum	Текущее количество подзадач в задаче загрузки.
JobProperties	Свойства задачи загрузки, такие как потребляемые партиции и сопоставление столбцов.
DataSourceProperties	Свойства источника данных, такие как топик и список адресов и портов брокеров в Kafka cluster.
CustomProperties	Дополнительные свойства, связанные с источником данных, определённые в задаче загрузки.
Statistic	Статистика загрузки данных, такая как успешно загруженные строки, общее количество строк и объём полученных данных.
Progress	Прогресс (измеряемый смещением offset) потребления сообщений в партициях топика.
TimestampProgress	Прогресс (измеряемый временной меткой timestamp) потребления сообщений в партициях топика.
ReasonOfStateChanged	Причины нахождения задачи загрузки в состоянии CANCELLED или PAUSED.
ErrorLogUrls	URL журналов ошибок. Вы можете использовать команду curl или wget для доступа к URL.
TrackingSQL	SQL-команда, которую вы можете напрямую выполнить для запроса информации журнала ошибок, записанной в базу данных information_schema.
OtherMsg	Информация обо всех неудачных подзадачах загрузки задачи Routine Load.
LatestSourcePosition	Позиция последнего сообщения в каждой партиции топика, что помогает проверить задержки загрузки данных.

Примеры

Если задача загрузки успешно запущена и находится в состоянии RUNNING, возвращённый результат может быть следующим:

MySQL [example_db]> SHOW ROUTINE LOAD FOR example_tbl_ordertest1\G
*************************** 1. row ***************************
                  Id: 10204
                Name: example_tbl_ordertest1
          CreateTime: 2023-12-21 21:01:31
           PauseTime: NULL
             EndTime: NULL
              DbName: example_db
           TableName: example_tbl
               State: RUNNING
      DataSourceType: KAFKA
      CurrentTaskNum: 1
       JobProperties: {"partitions":"*","rowDelimiter":"\t","partial_update":"false","columnToColumnExpr":"order_id,pay_dt,customer_name,nationality,temp_gender,price","maxBatchIntervalS":"10","partial_update_mode":"null","whereExpr":"*","timezone":"Asia/Shanghai","format":"csv","columnSeparator":"','","log_rejected_record_num":"0","taskTimeoutSecond":"60","json_root":"","maxFilterRatio":"1.0","strict_mode":"false","jsonpaths":"","taskConsumeSecond":"15","desireTaskConcurrentNum":"5","maxErrorNum":"0","strip_outer_array":"false","currentTaskConcurrentNum":"1","maxBatchRows":"200000"}
DataSourceProperties: {"topic":"lilyliuyitest4csv","currentKafkaPartitions":"0","brokerList":"xxx.xx.xx.xxx:9092"}
    CustomProperties: {"kafka_default_offsets":"OFFSET_BEGINNING","group.id":"example_tbl_ordertest1_b05da08f-9b9d-4fe1-b1f2-25d7116d617c"}
           Statistic: {"receivedBytes":313,"errorRows":0,"committedTaskNum":1,"loadedRows":6,"loadRowsRate":0,"abortedTaskNum":0,"totalRows":6,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":699}
            Progress: {"0":"5"}
   TimestampProgress: {"0":"1686143856061"}
ReasonOfStateChanged:
        ErrorLogUrls:
         TrackingSQL:
            OtherMsg:
LatestSourcePosition: {"0":"6"}
1 row in set (0.01 sec)

Если задача загрузки находится в состоянии PAUSED или CANCELLED из-за исключений, вы можете устранить неполадки на основе полей ReasonOfStateChanged, ErrorLogUrls, TrackingSQL и OtherMsg в возвращённом результате.

MySQL [example_db]> SHOW ROUTINE LOAD FOR example_tbl_ordertest2\G
*************************** 1. row ***************************
                  Id: 10204
                Name: example_tbl_ordertest2
          CreateTime: 2023-12-22 12:13:18
           PauseTime: 2023-12-22 12:13:38
             EndTime: NULL
              DbName: example_db
           TableName: example_tbl
               State: PAUSED
      DataSourceType: KAFKA
      CurrentTaskNum: 0
       JobProperties: {"partitions":"*","rowDelimiter":"\t","partial_update":"false","columnToColumnExpr":"order_id,pay_dt,customer_name,nationality,temp_gender,price","maxBatchIntervalS":"10","partial_update_mode":"null","whereExpr":"*","timezone":"Asia/Shanghai","format":"csv","columnSeparator":"','","log_rejected_record_num":"0","taskTimeoutSecond":"60","json_root":"","maxFilterRatio":"1.0","strict_mode":"false","jsonpaths":"","taskConsumeSecond":"15","desireTaskConcurrentNum":"5","maxErrorNum":"0","strip_outer_array":"false","currentTaskConcurrentNum":"1","maxBatchRows":"200000"}
DataSourceProperties: {"topic":"mytest","currentKafkaPartitions":"0","brokerList":"xxx.xx.xx.xxx:9092"}
    CustomProperties: {"kafka_default_offsets":"OFFSET_BEGINNING","group.id":"example_tbl_ordertest2_b3fada0f-6721-4ad1-920d-e4bf6d6ea7f7"}
           Statistic: {"receivedBytes":541,"errorRows":10,"committedTaskNum":1,"loadedRows":6,"loadRowsRate":0,"abortedTaskNum":0,"totalRows":16,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":646}
            Progress: {"0":"19"}
   TimestampProgress: {"0":"1702623900871"}
ReasonOfStateChanged: ErrorReason{errCode = 102, msg='current error rows is more than max error num'}
        ErrorLogUrls: http://xxx.xx.xx.xxx:8040/api/_load_error_log?file=error_log_b25dcc7e642344b2_b0b342b9de0567db
         TrackingSQL: select tracking_log from information_schema.load_tracking_logs where job_id=10204
            OtherMsg:
LatestSourcePosition: {"0":"20"}
1 row in set (0.00 sec)