全部产品
云市场

AttachCluster最佳实践

更新时间:2020-03-04 10:20:56

0 背景

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

1 准备工作

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

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

1.2 升级Python SDK

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

2 创建集群

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)
  47. # get cluster id for attach cluster job
  48. return rsp.Id
  49. except ClientError, e:
  50. print (e.get_status_code(), e.get_code(), e.get_requestid(), e.get_msg())
  51. return ""
  52. if __name__ == '__main__':
  53. #Not Use idempotent token
  54. cluster_id = create_cluster()
  55. print cluster_id

3 创建作业

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

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

AttachCluster作业创建已经完成。