跳到主要内容
版本:0.6.12

Starwhale 任务 SDK

job

通过Job URI参数获取 starwhale.Job 对象,可以获得 Standalone/Server/Cloud 实例上的任务。

@classmethod
def job(
cls,
uri: str,
) -> Job:

参数

  • uri: (str, required)
    • Job URI格式的字符串。

使用示例

from starwhale import job
# get job object of uri=https://server/job/1
j1 = job("https://server/job/1")
# get job from standalone instance
j2 = job("local/project/self/job/xm5wnup")
j3 = job("xm5wnup")

class starwhale.Job

starwhale.Job 实现对 Starwhale 任务的抽象,能够对 Standalone/Server/Cloud 实例上的任务进行一些信息获取类的操作。

list

list 是一个 classmethod 方法,能够列出某个项目下的任务。

@classmethod
def list(
cls,
project: str = "",
page_index: int = DEFAULT_PAGE_IDX,
page_size: int = DEFAULT_PAGE_SIZE,
) -> t.Tuple[t.List[Job], t.Dict]:

参数

  • project: (str, optional)
    • Project URI,Standalone/Server/Cloud 实例上的项目都可以。
    • 若不指定 project 参数,则使用 swcli project selected 命令选定的项目。
  • page_index: (int, optional)
    • 当获取 Server/Cloud 实例的项目列表时,支持翻页操作,该参数可以指定页面序号。
      • 默认值为 1。
      • 页面起始序号为 1。
    • Standalone 实例不支持翻页操作,设置该参数无效。
  • page_size: (int, optional)
    • 当获取 Server/Cloud 实例的项目列表时,支持翻页操作,该参数可以指定每页返回的任务数量。
      • 默认值为 1。
      • 页面起始序号为 1。
    • Standalone 实例不支持翻页操作,设置该参数无效。

使用示例

from starwhale import Job
# list jobs of current selected project
jobs, pagination_info = Job.list()
# list jobs of starwhale/public project in the cloud.starwhale.cn instance
jobs, pagination_info = Job.list("https://cloud.starwhale.cn/project/starwhale:public")
# list jobs of id=1 project in the server instance, page index is 2, page size is 10
jobs, pagination_info = Job.list("https://server/project/1", page_index=2, page_size=10)

get

get 是一个 classmethod 方法,能够获得某个特定任务的信息,返回 Starwhale.Job 对象,与 starwhale.job 函数功能和参数定义上完全一致。

使用示例

from starwhale import Job
# get job object of uri=https://server/job/1
j1 = Job.get("https://server/job/1")
# get job from standalone instance
j2 = Job.get("local/project/self/job/xm5wnup")
j3 = Job.get("xm5wnup")

summary

summary 是一个 property 属性,返回任务运行中写入 summary 表中的数据,字典类型。

@property
def summary(self) -> t.Dict[str, t.Any]:

使用示例

from starwhale import jobs
j1 = job("https://server/job/1")
print(j1.summary)

tables

tables 是一个 property 属性,返回任务运行中创建的表名(不包括 summary 表,以为 summary 表是项目级别自动创建的),列表类型。

@property
def tables(self) -> t.List[str]:

使用示例

from starwhale import jobs
j1 = job("https://server/job/1")
print(j1.tables)

get_table_rows

get_table_rows 是一个 method 方法,可以根据表名等参数返回数据表的记录,迭代器类型。

def get_table_rows(
self,
name: str,
start: t.Any = None,
end: t.Any = None,
keep_none: bool = False,
end_inclusive: bool = False,
) -> t.Iterator[t.Dict[str, t.Any]]:

参数

  • name: (str, required)
    • datastore 表名。通过 tables 属性获得的表名,可以传给 name 参数。
  • start: (Any, optional)
    • 返回记录中,ID的起始值。
    • 默认值为 None,表示从头开始。
  • end: (Any, optional)
    • 返回记录中,ID的结束值。
    • 默认值为 None ,表示一直到表末尾。
    • startend 都为 None,则会以迭代器方式返回整个表的数据。
  • keep_none: (bool, optional)
    • 是否返回值为 None的记录。
    • 默认为 False。
  • end_inclusive: (bool, optional)
    • end 参数设置时,迭代记录的时候,是否包含end记录。
    • 默认为 False。

使用示例

from starwhale import job
j = job("local/project/self/job/xm5wnup")
table_name = j.tables[0]
for row in j.get_table_rows(table_name):
print(row)
rows = list(j.get_table_rows(table_name, start=0, end=100))
# return the first record from the results table
result = list(j.get_table_rows('results', start=0, end=1))[0]

status

status 是一个 property 属性,返回当前Job的实时状态,字符串类型,状态包含 CREATED, READY, PAUSED, RUNNING, CANCELLING, CANCELED, SUCCESS, FAILUNKNOWN

@property
def status(self) -> str:

create

create 是一个 classmethod 方法,能够创建 Standalone 实例或 Server/Cloud 实例上的任务,包括Model Evluation, Fine-tuning, Online Serving 和 Developing 等类型的任务。函数返回 Job 类型的对象。

  • create 通过 project 参数决定生成的任务运行在何种实例上,包括 Standalone 和 Server/Cloud 实例。
  • 在 Standalone 实例下,create 创建一个同步执行的任务。
  • 在 Server/Cloud 实例下, create 创建一个异步执行的任务。
@classmethod
def create(
cls,
project: Project | str,
model: Resource | str,
run_handler: str,
datasets: t.List[str | Resource] | None = None,
runtime: Resource | str | None = None,
resource_pool: str = DEFAULT_RESOURCE_POOL,
ttl: int = 0,
dev_mode: bool = False,
dev_mode_password: str = "",
dataset_head: int = 0,
overwrite_specs: t.Dict[str, t.Any] | None = None,
) -> Job:

参数

对所有实例都生效的参数:

  • project: (Project|str, required)
    • Project 对象或 Project URI 字符串。
  • model: (Resource|str, required)
    • Model URI 字符串或 Model 类型的 Resource 对象,表示要运行的 Starwhale 模型包。
  • run_handler: (str, required)
    • Starwhale 模型包中对应的可运行的 handler 名称,比如 mnist 的 evaluate handler: mnist.evaluator:MNISTInference.evaluate
  • datasets: (List[str | Resource], optional)
    • Starwhale 模型包运行所需要的数据集,非必需。

仅对 Standalone 实例生效的参数:

  • dataset_head: (int, optional)
    • 一般用于调试场景,只使用数据集前 N 条数据来供 Starwhale 模型来消费。

仅对 Server/Cloud 实例生效的参数:

  • runtime: (Resource | str, optional)
    • Runtime URI 字符串或 Runtime 类型的 Resource 对象,表示要运行任务所需要的 Starwhale 运行时。
    • 当不指定该参数时,会尝试使用 Starwhale 模型包的内建运行时。
    • 创建 Standalone 实例下的任务,使用 Python 脚本所用的 Python 解释器环境作为自己的运行时,不支持通过 runtime 参数指定运行时。若有指定运行时的需要,可以使用 swcli model run 命令。
  • resource_pool: (str, optional)
    • 指定任务运行在哪个资源池中,默认为 default 资源池。
  • ttl: (int, optional)
    • 任务最大存活时间,超时后会被杀掉。
    • 参数单位为秒。
    • 默认情况下,ttl为0,表示没有超时限制,任务会按预期运行。
    • 当ttl小于0时,也表示没有超时限制。
  • dev_mode: (bool, optional)
    • 是否设置为调试模式。开启此模式后,可以通过VSCode Web进入到相关环境中。
    • 默认不进入调试模式。
  • dev_mode_password: (str, optional)
    • 调试模式下VSCode Web的登录密码。
    • 默认为空,此时会用任务的UUID作为密码,可以通过 job.info().job.uuid 获得。
  • overwrite_specs: (Dict[str, Any], optional)
    • 支持设置 handler 的 replicasresources 字段。
    • 若为空,则使用模型包中对应 handler 设置的值。
    • overwrite_specs 的 key 为 handler 的名字,比如 mnist 的 evaluate handler: mnist.evaluator:MNISTInference.evaluate
    • overwrite_specs 的 value 为设置的值,字典格式,支持设置 replicasresources , 比如 {"replicas": 1, "resources": {"memory": "1GiB"}}

使用示例

  • 创建一个 Cloud 实例的任务
from starwhale import Job
project = "https://cloud.starwhale.cn/project/starwhale:public"
job = Job.create(
project=project,
model=f"{project}/model/mnist/version/v0",
run_handler="mnist.evaluator:MNISTInference.evaluate",
datasets=[f"{project}/dataset/mnist/version/v0"],
runtime=f"{project}/runtime/pytorch",
overwrite_specs={"mnist.evaluator:MNISTInference.evaluate": {"resources": "4GiB"},
"mnist.evaluator:MNISTInference.predict": {"resources": "8GiB", "replicas": 10}}
)
print(job.status)
  • 创建一个 Standalone 实例的任务
from starwhale import Job
job = Job.create(
project="self",
model="mnist",
run_handler="mnist.evaluator:MNISTInference.evaluate",
datasets=["mnist"],
)
print(job.status)