列出并查看结构化导入任务

在工作区中列出现有结构化导入任务，选择任务后读取详情，检查状态、行数和错误信息。

准备

  1. 配置 SDK 和个人访问令牌。

  2. 选择工作区并获取工作区 ID。

  3. 任务 ID：无需预先准备；由列出任务并读取详情的示例输出，运行时从列表中选择。

列出任务并读取详情

将下面的 Python 示例保存为脚本执行，或将 Go 示例保存为主程序运行。程序按每页 100 条读取导入任务，只输出结构化任务，再提示输入其中一个任务 ID；只执行读取操作。

import os

import moi_product_sdk as sdk


def main():
    client = sdk.new_with_personal_access_token(
        os.environ["PRODUCT_API_BASE_URL"], os.environ["PRODUCT_API_KEY"]
    )
    service = client.workspace(os.environ["WORKSPACE_ID"]).connectors()
    page, seen = 1, set()
    while True:
        result = service.tasks(
            sdk.with_page(page), sdk.with_page_size(100),
        )
        for task in result.tasks:
            if task.config_type != 3:
                continue
            seen.add(task.id)
            print(f"id={task.id} name={task.name} connector={task.connector_name} status={task.status}")
        if not result.tasks or page * 100 >= result.total:
            break
        page += 1
    if not seen:
        print("当前身份在此工作区未读到结构化导入任务")
        return
    task_id = input("输入上方要查看的任务 ID: ").strip()
    if task_id not in seen:
        raise ValueError("请选择本次列表返回的任务 ID")
    info = service.get_task(task_id)
    if not info.HasField("task") or info.task.id != task_id:
        raise RuntimeError("详情缺失或任务 ID 不匹配")
    task = info.task
    print(f"详情读取成功: id={task.id} name={task.name} status={task.status}")
    print(f"total_rows={task.total_rows} imported_rows={task.imported_rows}")
    print(f"error_code={task.error_code} error_summary={task.error_summary}")
    for name in ("latest_rows", "cumulative_rows"):
        if not task.HasField(name):
            print(f"{name}=未提供")
            continue
        stats = getattr(task, name)
        print(f"{name}.complete={stats.complete}")
        for field in ("read_rows", "succeeded_rows", "failed_rows", "skipped_rows"):
            value = getattr(stats, field) if stats.HasField(field) else "未提供"
            print(f"{name}.{field}={value}")


if __name__ == "__main__":
    main()

package main

import (
	"context"
	"fmt"
	"log"
	"os"

	sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
)

func main() {
	ctx := context.Background()
	client, err := sdk.NewWithPersonalAccessToken(os.Getenv("PRODUCT_API_BASE_URL"), os.Getenv("PRODUCT_API_KEY"))
	if err != nil {
		log.Fatal(err)
	}
	workspace, err := client.Workspace(os.Getenv("WORKSPACE_ID"))
	if err != nil {
		log.Fatal(err)
	}
	service := workspace.Connectors()
	seen := map[string]bool{}
	for page := 1; ; page++ {
		result, err := service.Tasks(ctx, sdk.WithPage(page), sdk.WithPageSize(100))
		if err != nil {
			log.Fatal(err)
		}
		for _, task := range result.GetTasks() {
			if task.GetConfigType() != 3 {
				continue
			}
			seen[task.GetId()] = true
			fmt.Printf("id=%s name=%s connector=%s status=%d\n", task.GetId(), task.GetName(), task.GetConnectorName(), task.GetStatus())
		}
		if len(result.GetTasks()) == 0 || int64(page*100) >= result.GetTotal() {
			break
		}
	}
	if len(seen) == 0 {
		fmt.Println("当前身份在此工作区未读到结构化导入任务")
		return
	}
	fmt.Print("输入上方要查看的任务 ID: ")
	var taskID string
	if _, err := fmt.Scanln(&taskID); err != nil {
		log.Fatal(err)
	}
	if !seen[taskID] {
		log.Fatal("请选择本次列表返回的任务 ID")
	}
	info, err := service.GetTask(ctx, taskID)
	if err != nil {
		log.Fatal(err)
	}
	task := info.GetTask()
	if task == nil || task.GetId() != taskID {
		log.Fatal("详情缺失或任务 ID 不匹配")
	}
	fmt.Printf("详情读取成功: id=%s name=%s status=%d\n", task.GetId(), task.GetName(), task.GetStatus())
	fmt.Printf("total_rows=%d imported_rows=%d\n", task.GetTotalRows(), task.GetImportedRows())
	fmt.Printf("error_code=%s error_summary=%s\n", task.GetErrorCode(), task.GetErrorSummary())
	for _, name := range []string{"latest_rows", "cumulative_rows"} {
		stats := task.GetLatestRows()
		if name == "cumulative_rows" {
			stats = task.GetCumulativeRows()
		}
		if stats == nil {
			fmt.Printf("%s=未提供\n", name)
			continue
		}
		fmt.Printf("%s.complete=%t\n", name, stats.GetComplete())
		fields := []struct {
			name  string
			value *int64
		}{
			{"read_rows", stats.ReadRows}, {"succeeded_rows", stats.SucceededRows},
			{"failed_rows", stats.FailedRows}, {"skipped_rows", stats.SkippedRows},
		}
		for _, field := range fields {
			if field.value == nil {
				fmt.Printf("%s.%s=未提供\n", name, field.name)
			} else {
				fmt.Printf("%s.%s=%d\n", name, field.name, *field.value)
			}
		}
	}
}

查看任务状态和统计

输出字段

如何判断

id、name、connector

列表中的任务 ID、名称和来源连接器名称；输入其中一个 ID 后，详情必须返回相同 ID，程序才输出“详情读取成功”。

status

当前任务状态。结合错误信息检查处理情况。

total_rows、imported_rows

任务总行数与已导入行数，用于查看处理进度。

latest_rows、cumulative_rows

本次与累计行数统计。complete 表示该组统计是否完整；read_rows、succeeded_rows、failed_rows、skipped_rows 分别为读取、成功、失败和跳过行数。“未提供”与明确返回零不同。

error_code、error_summary

任务错误代码和摘要，用于排查处理失败。

请求失败、所选任务不在列表中或详情 ID 不匹配时，程序终止。分页期间任务可能变化，详情显示本次查询的状态。

最后更新于