AI 助理
备案控制台
官方文档
输入文档关键字查找
首页 批量计算（文档停止维护） 实践教程 AttachCluster最佳实践

AttachCluster最佳实践

更新时间：
我的收藏

0 背景

AttachCluster作业是批量计算最新推出的作业类型。它结合了固定集群作业和AutoCluster作业的优势，既能自动管理集群生命周期，弹性伸缩资源，又能使用分布式缓存节省资源。本文的目的在于介绍在阿里云批量计算服务上运行AttachCluster作业。

1 使用限制

  • 支持创建集群时自定义系统盘和数据盘大小

  • 不支持作业中自定义系统盘

    创建默认集群中定义实例系统盘大小为SystemDiskSize之后，提交到该集群中的所有AttachCluster作业都默认设置为SystemDiskSize。提交job的时候，该字段填0或者不填写。

  • 不支持作业中自定义数据

    创建默认集群中定义实例系统盘大小为SystemDiskSize之后，提交到该集群中的所有AttachCluster作业都默认设置为DataDiskSize。提交job的时候，该字段填0或者不填写。

  • 不支持APP作业模式， 支持DAG作业模式

  • 作业中填写的镜像必须是m-xxx开头的镜像, 不是img开头的镜像（提交job的时候务必仔细检查这里！）

2 准备工作

2.1 开通阿里云批量计算服务

要使用批量计算服务，请根据官方文档里面的指导开通批量计算和其依赖的相关服务，如OSS等。

2.2 升级Python SDK

若您未安装批量计算Python SDK，请您参照安装方法安装该SDK。如果您检查已经安装之后，请您参照Python SDK升级方法， 升级批量计算Python SDK至最新版。

3 创建集群

AttachCluster作业首次使用时，需要创建一个集群，创建方法可参考官方文档 。该集群对配置没有特殊需求，实例数可设置为0。以下是创建集群的Python源代码。

  1. import time
  2. import random
  3. import string
  4. import batchcompute
  5. from batchcompute import CN_SHENZHEN as REGION
  6. from batchcompute import Client, ClientError
  7. from batchcompute.resources import (
  8.        JobDescription, TaskDescription, DAG,
  9.        GroupDescription, ClusterDescription,
  10.        Configs, Networks, VPC, Classic, Mounts, Notification, Topic
  11. )
  12. ACCESS_KEY_ID = 'Your Access Key Id'
  13. ACCESS_KEY_SECRET = 'Your Access Key Secret'
  14. IMAGE_ID = 'img-ubuntu'
  15. INSTANCE_TYPE  = 'ecs.sn2ne.large'
  16. client = Client(REGION, ACCESS_KEY_ID, ACCESS_KEY_SECRET)
  17. def create_cluster(idempotent_token=''):
  18.       try:
  19.              # Cluster description.
  20.              cluster_desc = ClusterDescription()
  21.              cluster_desc.Name = "test-cluster"
  22.              cluster_desc.Description = "demo"
  23.              cluster_desc.ImageId = IMAGE_ID
  24.              cluster_desc.InstanceType = INSTANCE_TYPE
  25.              #Group description
  26.              group_desc1 = GroupDescription()
  27.              group_desc1.DesiredVMCount = 4
  28.              group_desc1.InstanceType = 'ecs.sn1ne.large'  #user group special instance type
  29.              group_desc1.ResourceType = 'OnDemand'
  30.              cluster_desc.add_group('group1', group_desc1)
  31.              #cluster_desc.add_group('group2', group_desc2)
  32.              #Configs
  33.              configs = Configs()
  34.              #Configs.Disks
  35.              configs.add_system_disk(50, 'cloud_efficiency')
  36.              configs.add_data_disk(500, 'cloud_efficiency', '/home/my-data-disk')
  37.              #Configs.Networks
  38.              networks  = Networks()
  39.              vpc = VPC()
  40.              vpc.CidrBlock = '192.168.0.0/16'
  41.              #vpc.VpcId = 'vpc-xxxxx'
  42.              networks.VPC = vpc     
  43.              configs.Networks = networks
  44.              cluster_desc.Configs = configs
  45.              print cluster_desc
  46.              rsp = client.create_cluster(cluster_desc, idempotent_token)
  48.              # get cluster id for attach cluster job
  49.              return rsp.Id
  51.       except ClientError, e:
  52.              print (e.get_status_code(), e.get_code(), e.get_requestid(), e.get_msg())
  53.              return ""
  54. if __name__ == '__main__':
  55.      #Not Use idempotent token
  56.      cluster_id = create_cluster()
  57.      print cluster_id

3 创建作业

在创建作业的时候需要步骤2中的集群Id，填入task的AutoCluster的ClusterId字段中。以下是创建作业的Python源代码。

  1. from batchcompute import Client, ClientError
  2. from batchcompute import CN_SHENZHEN as REGION
  3. from batchcompute.resources import (
  4.        ClusterDescription, GroupDescription, Configs, Networks, VPC,
  5.        JobDescription, TaskDescription, DAG,Mounts,
  6.        AutoCluster,Disks,Notification,
  7. )
  8. access_key_id = "" # your access key id
  9. access_key_secret = "" # your access key secret
  10. image_id = "m-8vbd8lo9xxxx" # the id of a image created before,镜像需要确保已经注册给批量计算，且必须是m-xx开头的镜像，不是img开头的镜像
  11. instance_type = "ecs.sn1.medium" # instance type
  12. inputOssPath = "oss://xxx/input/" # your input oss path
  13. outputOssPath = "oss://xxx/output/" #your output oss path
  14. stdoutOssPath = "oss://xxx/log/stdout/" #your stdout oss path
  15. stderrOssPath = "oss://xxx/log/stderr/" #your stderr oss path
  16. def getAutoClusterDesc():
  17.       auto_desc = AutoCluster()
  18.       # attach cluster这里里填入上一步创建的集群Id
  19.       auto_desc.ClusterId = cls-xxxxx
  21.       auto_desc.ImageId = image_id
  22.       auto_desc.ReserveOnFail = False
  23.       # 实例规格
  24.       auto_desc.InstanceType = instance_type
  25.       #case1 设置上限价格的竞价实例; 
  26.       # auto_desc.ResourceType = "Spot"
  27.       # auto_desc.SpotStrategy = "SpotWithPriceLimit"
  28.       # auto_desc.SpotPriceLimit = 0.5
  29.       #case2 系统自动出价，最高按量付费价格
  30.       # auto_desc.ResourceType = "Spot"
  31.       # auto_desc.SpotStrategy = "SpotAsPriceGo"
  32.       #case3 按量
  33.       auto_desc.ResourceType = "OnDemand"
  34.       #Configs
  35.       configs = Configs()
  36.       #Configs.Networks
  37.       networks  = Networks()
  38.       vpc = VPC()
  39.       #case1 只给CidrBlock
  40.       vpc.CidrBlock = '192.168.0.0/16'
  41.       #case2 CidrBlock和VpcId 都传入，必须保证VpcId的CidrBlock 和传入的CidrBlock保持一致
  42.       # vpc.CidrBlock = '172.26.0.0/16'
  43.       # vpc.VpcId = "vpc-8vbfxdyhxxxx"
  44.       networks.VPC = vpc
  45.       configs.Networks = networks
  46.       # 不支持设置系统盘
  47.       #configs.add_system_disk(size=0, type_='cloud_efficiency')
  48.       #不支持设置数据盘
  49.       # case1 linux环境
  50.       # configs.add_data_disk(size=0, type_='cloud_efficiency', mount_point='/path/to/mount/')
  51.       # case2 windows环境
  52.       # configs.add_data_disk(size=0, type_='cloud_efficiency', mount_point='E:')
  53.       # 设置节点个数
  54.       configs.InstanceCount = 1
  55.       auto_desc.Configs = configs
  56.       return auto_desc
  58. def getDagJobDesc(clusterId = None):
  59.       job_desc = JobDescription()
  60.       dag_desc = DAG()
  61.       mounts_desc = Mounts()
  62.       job_desc.Name = "testBatchSdkJob"
  63.       job_desc.Description = "test job"
  64.       job_desc.Priority = 1
  65.       # 订阅job完成或者失败事件
  66.       noti_desc = Notification()
  67.       noti_desc.Topic['Name'] = "test-topic"
  68.       noti_desc.Topic['Endpoint'] = "http://[UserId].mns.[Region].aliyuncs.com/"
  69.       noti_desc.Topic['Events'] = ["OnJobFinished", "OnJobFailed"]
  70.       # job_desc.Notification = noti_desc
  71.       job_desc.JobFailOnInstanceFail = False
  72.       # 作业运行成功后户自动会被立即释放掉
  73.       job_desc.AutoRelease = False
  74.       job_desc.Type = "DAG"
  75.       echo_task = TaskDescription()
  76.       # echo_task.InputMapping = {"oss://xxx/input/": "/home/test/input/",
  77.       #                          "oss://xxx/test/file": "/home/test/test/file"}
  78.       echo_task.InputMapping = {inputOssPath: "/home/test/input/"}
  79.       echo_task.OutputMapping = {"/home/test/output/":outputOssPath}
  80.       #触发程序运行的命令行
  81.       #case1 执行linux命令行
  82.       echo_task.Parameters.Command.CommandLine = "/bin/bash -c 'echo BatchcomputeService'"
  83.       #case2 执行Windows CMD.exe
  84.       # echo_task.Parameters.Command.CommandLine = "cmd /c 'echo BatchcomputeService'"
  85.       #case3 输入可执行文件
  86.       # PackagePath存放commandLine中的可执行文件或者二进制包
  87.       # echo_task.Parameters.Command.PackagePath = "oss://xxx/package/test.sh"
  88.       # echo_task.Parameters.Command.CommandLine = "sh test.sh"
  89.       # 设置程序运行过程中相关环境变量信息
  90.       echo_task.Parameters.Command.EnvVars["key1"] = "value1"
  91.       echo_task.Parameters.Command.EnvVars["key2"] = "value2"
  92.       # 设置程序的标准输出地址，程序中的print打印会实时上传到指定的oss地址
  93.       echo_task.Parameters.StdoutRedirectPath = stdoutOssPath
  94.       # 设置程序的标准错误输出地址，程序抛出的异常错误会实时上传到指定的oss地址
  95.       echo_task.Parameters.StderrRedirectPath = stderrOssPath
  96.       # 设置任务的超时时间
  97.       echo_task.Timeout = 600
  98.       # 设置任务所需实例个数
  99.       # 环境变量BATCH_COMPUTE_INSTANCE_ID为0到InstanceCount-1
  100.       # 在执行程序中访问BATCH_COMPUTE_INSTANCE_ID，实现数据访问的切片实现单任务并发执行
  101.       echo_task.InstanceCount = 1
  102.       # 设置任务失败后重试次数
  103.       echo_task.MaxRetryCount = 0
  104.       # NAS数据挂载
  105.       #采用NAS时必须保证网络和NAS在同一个VPC内
  106.       nasMountEntry = {
  107.           "Source": "nas://xxxx.nas.aliyuncs.com:/",
  108.           "Destination": "/home/mnt/",
  109.           "WriteSupport":True,
  110.       }
  111.       mounts_desc.add_entry(nasMountEntry)
  112.       mounts_desc.Locale = "utf-8"
  113.       mounts_desc.Lock = False
  114.       # echo_task.Mounts = mounts_desc
  116.       # attach cluster作业该集群字段设置为空
  117.       echo_task.ClusterId = ""
  119.       echo_task.AutoCluster = getAutoClusterDesc()
  120.       # 添加任务
  121.       dag_desc.add_task('echoTask', echo_task)
  122.       # 可以设置多个task,每个task可以根据需求进行设置各项参数
  123.       # dag_desc.add_task('echoTask2', echo_task)
  124.       # Dependencies设置多个task之间的依赖关系,echoTask2依赖echoTask;echoTask3依赖echoTask2
  125.       # dag_desc.Dependencies = {"echoTask":["echoTask2"], "echoTask2":["echoTask3"]}
  126.       job_desc.DAG = dag_desc
  127.       return job_desc
  129. if __name__ == "__main__":
  130.      client = Client(REGION, access_key_id, access_key_secret)
  131.      try:
  132.            job_desc = getDagJobDesc()
  133.            job_id = client.create_job(job_desc).Id
  134.            print('job created: %s' % job_id)
  135.      except ClientError,e:
  136.            print (e.get_status_code(), e.get_code(), e.get_requestid(), e.get_msg())

AttachCluster作业创建已经完成。

上一篇：GTX_FPGA 最佳实践下一篇：开发参考