全部产品
存储与CDN 数据库 安全 应用服务 数加·人工智能 数加·大数据基础服务 互联网中间件 视频服务 阿里云办公 培训与认证 物联网
批量计算

并发

更新时间:2017-06-07 13:26:11

一个作业(Job)中可以有多个任务(Task),一个任务可以指定在多个实例(Instance)上运行程序。

如何并发?

请看下面 job description json 例子:

  1. {
  2. "DAG": {
  3. ...
  4. "Tasks": {
  5. ...
  6. "count": {
  7. "InstanceCount": 3, //指定需要实例数:3台VM
  8. "LogMapping": {},
  9. "AutoCluster": {
  10. "ResourceType": "OnDemand",
  11. "ImageId": "img-ubuntu",
  12. "InstanceType": "bcs.a2.large"
  13. },
  14. "Parameters": {
  15. "Command": {
  16. "EnvVars": {},
  17. "CommandLine": "python count.py",
  18. "PackagePath": "oss://your-bucket/log-count/worker.tar.gz"
  19. },
  20. "InputMappingConfig": {
  21. "Lock": true
  22. },
  23. "StdoutRedirectPath": "oss://your-bucket/log-count/logs/",
  24. "StderrRedirectPath": "oss://your-bucket/log-count/logs/"
  25. },
  26. "OutputMapping": {},
  27. "MaxRetryCount": 0,
  28. "Timeout": 21600,
  29. "InputMapping": {}
  30. }
  31. }
  32. },
  33. "Description": "batchcompute job",
  34. "Priority": 0,
  35. "JobFailOnInstanceFail": true,
  36. "Type": "DAG",
  37. "Name": "log-count"
  38. }

任务count中配置了InstanceCount为3, 表示需要实例数3台, 即在3台VM上运行这个任务的程序。

并发处理不同片段的数据

3台VM上运行的任务程序都是一样的,如何让它处理不同的数据呢?

在任务程序中使用环境变量: BATCH_COMPUTE_DAG_INSTANCE_ID(实例ID) 来区分,可以处理不同片段的数据。

以下是 count.py 代码片段:

  1. ...
  2. # instance_id: should start from 0
  3. instance_id = os.environ['BATCH_COMPUTE_DAG_INSTANCE_ID']
  4. ...
  5. filename = 'part_%s.txt' % instance_id
  6. ...
  7. # 1. download a part
  8. oss_tool.download_file('%s/%s/%s.txt' % (pre, split_results, instance_id ), filename)
  9. ...
  10. # 3. upload result to oss
  11. upload_to = '%s/count_results/%s.json' % (pre, instance_id )
  12. print('upload to %s' % upload_to)
  13. oss_tool.put_data(json.dumps(m), upload_to)
  14. ...

完整例子请看快速开始例子

本文导读目录