create_job

參數(shù)說(shuō)明：

返回值說(shuō)明：

e.g.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
from batchcompute import Client, ClientError
from batchcompute import CN_ZHANGJIAKOU as REGION
from batchcompute.resources import (
    ClusterDescription, GroupDescription, Configs, Networks, VPC,
    JobDescription, TaskDescription, DAG,Mounts,
    AutoCluster,Disks,Notification,
)

access_key_id = "" # your access key id
access_key_secret = "" # your access key secret
image_id = "m-8vbd8lo9xxxx" # the id of a image created before,鏡像需要確保已經(jīng)注冊(cè)給批量計(jì)算
instance_type = "ecs.sn1.medium" # instance type
inputOssPath = "oss://xxx/input/" # your input oss path
outputOssPath = "oss://xxx/output/" #your output oss path
stdoutOssPath = "oss://xxx/log/stdout/" #your stdout oss path
stderrOssPath = "oss://xxx/log/stderr/" #your stderr oss path

def getAutoClusterDesc():
    auto_desc = AutoCluster()

    auto_desc.ECSImageId = image_id

    #任務(wù)失敗保留環(huán)境，程序調(diào)試階段設(shè)置。環(huán)境保留費(fèi)用會(huì)繼續(xù)產(chǎn)生請(qǐng)注意及時(shí)手動(dòng)清除環(huán)境任務(wù)失敗保留環(huán)境，
    # 程序調(diào)試階段設(shè)置。環(huán)境保留費(fèi)用會(huì)繼續(xù)產(chǎn)生請(qǐng)注意及時(shí)手動(dòng)清除環(huán)境
    auto_desc.ReserveOnFail = False

    # 實(shí)例規(guī)格
    auto_desc.InstanceType = instance_type

    #case1 設(shè)置上限價(jià)格的競(jìng)價(jià)實(shí)例; 
    # auto_desc.ResourceType = "Spot"
    # auto_desc.SpotStrategy = "SpotWithPriceLimit"
    # auto_desc.SpotPriceLimit = 0.5

    #case2 系統(tǒng)自動(dòng)出價(jià)，最高按量付費(fèi)價(jià)格
    # auto_desc.ResourceType = "Spot"
    # auto_desc.SpotStrategy = "SpotAsPriceGo"

    #case3 按量
    auto_desc.ResourceType = "OnDemand"

    #Configs
    configs = Configs()
    #Configs.Networks
    networks  = Networks()
    vpc = VPC()

    #case1 只給CidrBlock
    vpc.CidrBlock = '192.168.0.0/16'

    #case2 CidrBlock和VpcId 都傳入，必須保證VpcId的CidrBlock 和傳入的CidrBlock保持一致
    # vpc.CidrBlock = '172.26.0.0/16'
    # vpc.VpcId = "vpc-8vbfxdyhxxxx"

    networks.VPC = vpc
    configs.Networks = networks

    # 設(shè)置系統(tǒng)盤(pán)type（cloud_efficiency/cloud_ssd）以及size（單位GB）
    configs.add_system_disk(size=40, type_='cloud_efficiency')

    #設(shè)置數(shù)據(jù)盤(pán)type（必須和系統(tǒng)盤(pán)type保持一致）size（單位GB）掛載點(diǎn)
    # case1 linux環(huán)境
    # configs.add_data_disk(size=40, type_='cloud_efficiency', mount_point='/path/to/mount/')

    # case2 windows環(huán)境
    # configs.add_data_disk(size=40, type_='cloud_efficiency', mount_point='E:')

    # 設(shè)置節(jié)點(diǎn)個(gè)數(shù)
    configs.InstanceCount = 1
    auto_desc.Configs = configs
    return auto_desc

def getDagJobDesc(clusterId = None):
    job_desc = JobDescription()
    dag_desc = DAG()
    mounts_desc = Mounts()

    job_desc.Name = "testBatchSdkJob"
    job_desc.Description = "test job"
    job_desc.Priority = 1

    # 訂閱job完成或者失敗事件
    noti_desc = Notification()
    noti_desc.Topic['Name'] = "test-topic"
    noti_desc.Topic['Endpoint'] = "http://[UserId].mns.[Region].aliyuncs.com/"
    noti_desc.Topic['Events'] = ["OnJobFinished", "OnJobFailed"]
    # job_desc.Notification = noti_desc

    job_desc.JobFailOnInstanceFail = False

    # 作業(yè)運(yùn)行成功后戶(hù)自動(dòng)會(huì)被立即釋放掉
    job_desc.AutoRelease = False
    job_desc.Type = "DAG"

    echo_task = TaskDescription()

    # 程序的輸入路徑映射,程序直接訪問(wèn)/home/test/input/來(lái)訪問(wèn)oss://xxx/input/中的文件
    # 支持文件掛載,在程序中直接訪問(wèn)文件
    # echo_task.InputMapping = {"oss://xxx/input/": "/home/test/input/",
    #                          "oss://xxx/test/file": "/home/test/test/file"}
    echo_task.InputMapping = {inputOssPath: "/home/test/input/"}

    # 程序的輸出路徑映射,可執(zhí)行程序?qū)⒔Y(jié)果輸出到/home/test/output/，
    # 程序執(zhí)行完畢后批量計(jì)算將/home/test/output/中的結(jié)果上傳到oss://xxx/output/中
    # 輸入和輸出oss路徑不要有交叉，如輸入為oss://xxx/input/,輸出為oss://xxx/input/output/；
    # 這樣會(huì)導(dǎo)致未定義行為程序執(zhí)行性能不能保證
    echo_task.OutputMapping = {"/home/test/output/":outputOssPath}

    #觸發(fā)程序運(yùn)行的命令行
    #case1 執(zhí)行l(wèi)inux命令行
    echo_task.Parameters.Command.CommandLine = "/bin/bash -c 'echo BatchcomputeService'"

    #case2 執(zhí)行Windows CMD.exe
    # echo_task.Parameters.Command.CommandLine = "cmd /c 'echo BatchcomputeService'"

    #case3 輸入可執(zhí)行文件
    # PackagePath存放commandLine中的可執(zhí)行文件或者二進(jìn)制包
    # echo_task.Parameters.Command.PackagePath = "oss://xxx/package/test.sh"
    # echo_task.Parameters.Command.CommandLine = "sh test.sh"

    # 設(shè)置程序運(yùn)行過(guò)程中相關(guān)環(huán)境變量信息
    echo_task.Parameters.Command.EnvVars["key1"] = "value1"
    echo_task.Parameters.Command.EnvVars["key2"] = "value2"

    # 設(shè)置docker參數(shù)
    #case1 docker鏡像在oss registry上
    # echo_task.Parameters.Command.EnvVars["BATCH_COMPUTE_DOCKER_IMAGE"] = "localhost:5000/yuorBucket/dockers:0.1"
    # echo_task.Parameters.Command.EnvVars["BATCH_COMPUTE_DOCKER_REGISTRY_OSS_PATH"] = "oss://your-bucket/dockers"

    #case2 docker鏡像在容器倉(cāng)庫(kù)
    # echo_task.Parameters.Command.Docker.Image = "registry.cn-beijing.aliyuncs.com/demotest/test:0.1"

    # 設(shè)置程序的標(biāo)準(zhǔn)輸出地址，程序中的print打印會(huì)實(shí)時(shí)上傳到指定的oss地址
    echo_task.Parameters.StdoutRedirectPath = stdoutOssPath

    # 設(shè)置程序的標(biāo)準(zhǔn)錯(cuò)誤輸出地址，程序拋出的異常錯(cuò)誤會(huì)實(shí)時(shí)上傳到指定的oss地址
    echo_task.Parameters.StderrRedirectPath = stderrOssPath

    # 設(shè)置任務(wù)的超時(shí)時(shí)間
    echo_task.Timeout = 600

    # 設(shè)置任務(wù)所需實(shí)例個(gè)數(shù)
    # 環(huán)境變量BATCH_COMPUTE_INSTANCE_ID為0到InstanceCount-1
    # 在執(zhí)行程序中訪問(wèn)BATCH_COMPUTE_INSTANCE_ID，實(shí)現(xiàn)數(shù)據(jù)訪問(wèn)的切片實(shí)現(xiàn)單任務(wù)并發(fā)執(zhí)行
    echo_task.InstanceCount = 1

    # 設(shè)置任務(wù)失敗后重試次數(shù)
    echo_task.MaxRetryCount = 0

    # NAS數(shù)據(jù)掛載
    #采用NAS時(shí)必須保證網(wǎng)絡(luò)和NAS在同一個(gè)VPC內(nèi)
    nasMountEntry = {
        "Source": "nas://xxxx.nas.aliyuncs.com:/",
        "Destination": "/home/mnt/",
        "WriteSupport":True,
    }
    mounts_desc.add_entry(nasMountEntry)
    mounts_desc.Locale = "utf-8"
    mounts_desc.Lock = False
    # echo_task.Mounts = mounts_desc

    # 采用固定集群提交作業(yè)
    # echo_task.ClusterId = clusterId

    #采用auto集群提交作業(yè)
    echo_task.AutoCluster = getAutoClusterDesc()

    # 添加任務(wù)
    dag_desc.add_task('echoTask', echo_task)

    # 可以設(shè)置多個(gè)task,每個(gè)task可以根據(jù)需求進(jìn)行設(shè)置各項(xiàng)參數(shù)
    # dag_desc.add_task('echoTask2', echo_task)

    # Dependencies設(shè)置多個(gè)task之間的依賴(lài)關(guān)系,echoTask2依賴(lài)echoTask;echoTask3依賴(lài)echoTask2
    # dag_desc.Dependencies = {"echoTask":["echoTask2"], "echoTask2":["echoTask3"]}
    # 
    job_desc.DAG = dag_desc
    return job_desc

if __name__ == "__main__":
    client = Client(REGION, access_key_id, access_key_secret)
    try:
        job_desc = getDagJobDesc()
        job_id = client.create_job(job_desc).Id
        print('job created: %s' % job_id)
    except ClientError,e:
        print (e.get_status_code(), e.get_code(), e.get_requestid(), e.get_msg())

（1）JobDescription 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（2）DAG 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

方法說(shuō)明 ：

（3） TaskDescription 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（4） Parameters 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（5） AutoCluster 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（6） Command 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（7） InputMappingConfig 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（8） Notification 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（9） Topic 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（10） Mounts 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（11） MountPoint 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（12） NAS 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：

（13） OSS 類(lèi)

參數(shù)說(shuō)明：

屬性說(shuō)明：