文档中心

SHOW ROUTINE LOAD

Description

用户展示 Routine Load 作业信息。

SHOW [ALL] ROUTINE LOAD [FOR [db.]job_name]

• ALL：如果指定 ALL 关键词，则会显示所有正在运行的和已经停止的作业。否则，仅显示 非STOPPED 状态的作业。
• db：如不指定数据库，则默认展示当前数据库下的作业。

语句返回的结果如下：

mysql> show routine load\G *************************** 1. row *************************** Id: 10280 Name: job1
          CreateTime: 2020-12-10 19:32:58 PauseTime: NULL EndTime: NULL DbName: default_cluster:db1
           TableName: tbl1
               State: RUNNING
      DataSourceType: KAFKA
      CurrentTaskNum: 1 JobProperties: {"partitions":"*","columnToColumnExpr":"","

maxBatchIntervalS":"10","whereExpr":"*","timezone":"Asia/Shanghai","mergeType":"APPEND","

format":"csv","columnSeparator":"','","json_root":"","maxBatchSizeBytes":"104857600","exec_mem_limit":"

2147483648","strict_mode":"false","jsonpaths":"","deleteCondition":"*","desireTaskConcurrentNum":"1","

maxErrorNum":"0","strip_outer_array":"false","currentTaskConcurrentNum":"1","execMemLimit":"2147483648","

num_as_string":"false","maxBatchRows":"200000"}
DataSourceProperties: {"topic":"test","currentKafkaPartitions":"0","brokerList":"127.0.0.1:9094"}
    CustomProperties: {}
           Statistic: {"receivedBytes":0,"errorRows":0,"committedTaskNum":0,"loadedRows":0,"loadRowsRate":0,"

abortedTaskNum":0,"totalRows":0,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":1}
            Progress: {"0":"OFFSET_BEGINNING"}
ReasonOfStateChanged:
        ErrorLogUrls:
            OtherMsg: 1 row in set (0.01 sec)

• Id：作业唯一ID。
• Name：作业名称。
• CreateTime：作业创建时间。
• PauseTime：最近一次作业被暂停的时间。
• EndTime：作业结束时间。
• DbName：数据库名称。
• TableName：导入的表的名称。

• State：作业运行状态。

  • NEED_SCHEDULE：作业等待被调度
  • RUNNING：作业运行中。
  • PAUSED：作业被暂停。
  • STOPPED：作业已停止。
• DataSourceType：数据源类型。
• CurrentTaskNum：当前运行的子任务数量。

• JobProperties：作业配置详情，是一个 Json 格式字符串。

  { "partitions": "*", // 指定导入的分区列表，* 表示未指定。 "columnToColumnExpr": "", // 列映射和转换关系。

• 空表示未指定。 "maxBatchIntervalS": "10", // 子任务最大运行时长，单位秒。 "whereExpr": "*", // 列过滤条件。

• * 表示未指定。 "timezone": "Asia/Shanghai", // 时区。 "mergeType": "APPEND", // 数据合并类型。 "format": "

• csv", // 导入数据格式。 "columnSeparator": "','", // 列分隔符。 "json_root": "", // Json Root。 

• "maxBatchSizeBytes": "104857600", // 子任务最大消费字节数。 "exec_mem_limit": "2147483648", 

• // 子任务内存限制。 "strict_mode": "false", // 是否开启严格模式。 "jsonpaths": "", // json paths "

• deleteCondition": "*", // 是否指定 Marked Delete 列。* 表示未指定。  "desireTaskConcurrentNum": "1",

•  // 用户设置的期望最大并发子任务数。 "maxErrorNum": "0", // 做大容忍错误行数。 "strip_outer_array": "

• false", // 对于 Json 格式数据，是否展开数组。 "currentTaskConcurrentNum": "1", // 当前并发子任务数量。

•  "num_as_string": "false", // 是否将 Json 数据中所有字段解析为字符串类型。 "maxBatchRows": "

• 200000" // 子任务最大消费行数。 }

• CustomProperties：用户配置的额外属性。是一个 Json 格式字符串。

• Statistic：作业运行统计信息。是一个 Json 格式字符串。

  { "receivedBytes": 0, // 已经接受的数据大小。 "errorRows": 0, // 错误的数据行。 "committedTaskNum": 0, 

• // 成功导入的子任务数量。 "loadedRows": 0, // 已经导入的行数。 "loadRowsRate": 0, // 平均每秒导入的行数。

•  "abortedTaskNum": 0, // 失败或未消费到数据的子任务数量。 "totalRows": 0, // 总消费行数。 "unselectedRows":

•  0, // 被 where 条件过滤的行数。 "receivedBytesRate": 0, // 每秒接收的数据大小。 "taskExecuteTimeMs":

•  1 // 子任务累计执行时间。 }

• ReasonOfStateChanged：作业状态改变的原因。

• ErrorLogUrls：当有错误数据时，这里会显示最近的3个有错误数据的子任务的错误数据url连接。可以通过以下语句查看错误数据。

  SHOW LOAD WARNINGS ON "your_url";

  错误信息的说明，可以参阅 SHOW LOAD

• Progress：消费进度。是一个 Json 格式字符串。Key 为 Kafka 的 Partition ID。Value 含义如下：

  • OFFSET_BEGINNING：初始状态，表示从头开始消费，并且尚未开始消费。
  • OFFSET_END：初始状态，表示从末尾开始消费，并且尚未开始消费。
  • OFFSET_ZERO：初始状态，表示 0 开始消费，并且尚未开始消费。
  • 整型数值：对应 Kafka 分区 已经消费 的 offset。
• OtherMsg：其他信息。

example

1. 展示名称为 test1 的所有例行导入作业（包括已停止或取消的作业）。结果为一行或多行。

   SHOW ALL ROUTINE LOAD FOR test1;

2. 展示名称为 test1 的当前正在运行的例行导入作业

   SHOW ROUTINE LOAD FOR test1;

3. 显示 example_db 下，所有的例行导入作业（包括已停止或取消的作业）。结果为一行或多行。

   use example_db; SHOW ALL ROUTINE LOAD;

4. 显示 example_db 下，所有正在运行的例行导入作业

   use example_db; SHOW ROUTINE LOAD;

5. 显示 example_db 下，名称为 test1 的当前正在运行的例行导入作业

   SHOW ROUTINE LOAD FOR example_db.test1;

6. 显示 example_db 下，名称为 test1 的所有例行导入作业（包括已停止或取消的作业）。结果为一行或多行。

   SHOW ALL ROUTINE LOAD FOR example_db.test1;

Keywords

SHOW, ROUTINE, LOAD

最佳实践

1. 观察导入作业进度。

   可以通过返回结果中的如下字段来观察导入作业的消费进度。

   1. Statistic

      观察 Statistic 中 committedTaskNum 和 abortedTaskNum 两个数值的变化。如果 committedTaskNum 在持续增长，说明作业在正常消费。如果 abortedTaskNum 在持续增长，则说明可能 Kafka 中已经没有需要消费的数据；或者无法消费到 Kafka 的数据；或者遇到其他错误。

   2. Progress

      观察每个 Kafka Partition 的 offset 消费进度，也可以知晓整体的消费进度。