SHOW ROUTINE LOAD

Description

用户展示 Routine Load 作业信息。

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

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

语句返回的结果如下：

1mysql> show routine load\G
2*************************** 1. row ***************************
3                  Id: 10280
4                Name: job1
5          CreateTime: 2020-12-10 19:32:58
6           PauseTime: NULL
7             EndTime: NULL
8              DbName: default_cluster:db1
9           TableName: tbl1
10               State: RUNNING
11      DataSourceType: KAFKA
12      CurrentTaskNum: 1
13       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"}
14DataSourceProperties: {"topic":"test","currentKafkaPartitions":"0","brokerList":"127.0.0.1:9094"}
15    CustomProperties: {}
16           Statistic: {"receivedBytes":0,"errorRows":0,"committedTaskNum":0,"loadedRows":0,"loadRowsRate":0,"abortedTaskNum":0,"totalRows":0,"unselectedRows":0,"receivedBytesRate":0,"taskExecuteTimeMs":1}
17            Progress: {"0":"OFFSET_BEGINNING"}
18ReasonOfStateChanged:
19        ErrorLogUrls:
20            OtherMsg:
211 row in set (0.01 sec)

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

• State：作业运行状态。

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

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

  JSON 复制

  1{
  2    "partitions": "*",                  // 指定导入的分区列表，* 表示未指定。
  3    "columnToColumnExpr": "",           // 列映射和转换关系。空表示未指定。
  4    "maxBatchIntervalS": "10",          // 子任务最大运行时长，单位秒。
  5    "whereExpr": "*",                   // 列过滤条件。* 表示未指定。
  6    "timezone": "Asia/Shanghai",        // 时区。
  7    "mergeType": "APPEND",              // 数据合并类型。
  8    "format": "csv",                    // 导入数据格式。
  9    "columnSeparator": "','",           // 列分隔符。
  10    "json_root": "",                    // Json Root。
  11    "maxBatchSizeBytes": "104857600",   // 子任务最大消费字节数。
  12    "exec_mem_limit": "2147483648",     // 子任务内存限制。
  13    "strict_mode": "false",             // 是否开启严格模式。
  14    "jsonpaths": "",                    // json paths
  15    "deleteCondition": "*",             // 是否指定 Marked Delete 列。* 表示未指定。 
  16    "desireTaskConcurrentNum": "1",     // 用户设置的期望最大并发子任务数。
  17    "maxErrorNum": "0",                 // 做大容忍错误行数。
  18    "strip_outer_array": "false",       // 对于 Json 格式数据，是否展开数组。
  19    "currentTaskConcurrentNum": "1",    // 当前并发子任务数量。
  20    "num_as_string": "false",           // 是否将 Json 数据中所有字段解析为字符串类型。
  21    "maxBatchRows": "200000"            // 子任务最大消费行数。
  22}

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

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

  SQL 复制

  1{
  2    "receivedBytes": 0,         // 已经接受的数据大小。
  3    "errorRows": 0,             // 错误的数据行。
  4    "committedTaskNum": 0,      // 成功导入的子任务数量。
  5    "loadedRows": 0,            // 已经导入的行数。
  6    "loadRowsRate": 0,          // 平均每秒导入的行数。
  7    "abortedTaskNum": 0,        // 失败或未消费到数据的子任务数量。
  8    "totalRows": 0,             // 总消费行数。
  9    "unselectedRows": 0,        // 被 where 条件过滤的行数。
  10    "receivedBytesRate": 0,     // 每秒接收的数据大小。
  11    "taskExecuteTimeMs": 1      // 子任务累计执行时间。
  12}

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

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

  SQL 复制

  1SHOW 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 的所有例行导入作业（包括已停止或取消的作业）。结果为一行或多行。

   SQL 复制

   1SHOW ALL ROUTINE LOAD FOR test1;

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

   SQL 复制

   1SHOW ROUTINE LOAD FOR test1;

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

   SQL 复制

   1use example_db;
   2SHOW ALL ROUTINE LOAD;

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

   SQL 复制

   1use example_db;
   2SHOW ROUTINE LOAD;

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

   SQL 复制

   1SHOW ROUTINE LOAD FOR example_db.test1;

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

   SQL 复制

   1SHOW ALL ROUTINE LOAD FOR example_db.test1;

Keywords

1SHOW, ROUTINE, LOAD

典型实践

1. 观察导入作业进度。

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

   1. Statistic

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

   2. Progress

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