当前位置：主页 > 查看内容

创建(提交)作业 - 批量计算

发布时间：2021-09-20 00:00| 有 位朋友查看

简介：接口 createJob 参数 参数 类型 是否必需 说明 jobDescription JobDescription 是 Job对象中有各个任务的描述信息，和任务的DAG依赖。 jobDescription 的具体属性信息参考 DAG作业 和 APP作业 。 返回值 创建成功后返回一个 CreateJobResponse 实例，可以通……

接口

createJob

参数

参数 类型 是否必需 说明
jobDescription JobDescription 是 Job对象中有各个任务的描述信息，和任务的DAG依赖。

jobDescription 的具体属性信息参考DAG作业 和APP作业。

返回值

创建成功后返回一个 CreateJobResponse 实例，可以通过 response.getJobId() 获取创建的作业ID。创建失败后，抛出异常： ClientException 。

例子

Java 源码

  1. import com.aliyuncs.batchcompute.main.v20151111.*;
  2. import com.aliyuncs.batchcompute.model.v20151111.*;
  3. import com.aliyuncs.batchcompute.pojo.v20151111.*;
  4. import com.aliyuncs.exceptions.ClientException;
  6. public class CreateAppJob {
  7.     static String ACCESS_KEY_ID = "xxx";  //这里填写您的 AccessKeyId
  8.     static String ACCESS_KEY_SECRET = "xxx"; //这里填写您的 AccessKeySecret
  9.     static String REGION_ID = "cn-xxx";   //这里填写 region
  10.     static String ClusterId = "cls-xxx"; //提交DAG固定集群作业需要修改其他场景不需要修改
  12.     static boolean IS_DAG_JOB = true; //APP作业和DAG作业开关，默认提交DAG作业
  13.     static boolean IS_AUTO_CLUSTER = true; //固定集群和非固定集群作业开关，默认提交固定集群作业
  15.     public static void main(String[] args) {
  16.         BatchCompute client = new BatchComputeClient(REGION_ID, ACCESS_KEY_ID, ACCESS_KEY_SECRET);
  17.         try {
  18.             JobDescription jobDescription = getJobDesc();
  20.             CreateJobResponse response = client.createJob(jobDescription);
  21.             String jobId = response.getJobId();
  22.             //创建成功
  24.             System.out.println("jobId:" + jobId);
  25.             System.out.println("RequestId: " + response.getRequestId());
  26.             System.out.println("StatusCode: " + response.getStatusCode());
  27.         } catch (ClientException e) {
  28.             e.printStackTrace();
  29.             //创建失败
  30.         }
  31.     }
  33.     private static JobDescription getJobDesc() {
  34.         JobDescription desc = new JobDescription();
  36.         desc.setName("javaSdkJob");
  37.         desc.setDescription("javaSdkJob");
  39.         //设置作业优先级
  40.         desc.setPriority(1);
  42.         desc.setJobFailOnInstanceFail(true);
  43.         desc.setAutoRelease(false);
  45.         if (IS_DAG_JOB) {
  46.             //设置 DAG task
  47.             desc.setType("DAG");
  48.             desc.setDag(getDagDesc());
  49.         }else{
  50.             //设置 APP task
  51.             desc.setType("App");
  52.             desc.setApp(getAppJobDescription());
  53.         }
  55.         //根据业务需要设置订阅作业实践
  56. //        Notification noti = new Notification();
  57. //        Topic topic = new Topic();
  58. //        topic.addEvent(Topic.ON_JOB_FAILED);
  59. //        topic.addEvent(Topic.ON_JOB_FINISHED);
  60. //        noti.setTopic(topic);
  61. //        topic.setName("tp_n1");
  62. //        topic.setEndpoint("xxxxx");
  63. //        desc.setNotification(noti);
  65.         return desc;
  66.     }
  67.     private static AppJobDescription getAppJobDescription() {
  68.         AppJobDescription appJobDescription = new AppJobDescription();
  70.         appJobDescription.setAppName("JavaSdkApp");
  71.         appJobDescription.addInputs("inputFile", "oss://test/input/cromwell_app.txt");
  72.         appJobDescription.addOutputs("outputFile", "oss://test/output/ret/");
  74.         AppJobDescription.Logging logging = new AppJobDescription.Logging();
  75.         logging.setStderrPath("oss://test/output/error/");
  76.         logging.setStdoutPath("oss://test/output/log/");
  77.         appJobDescription.setLogging(logging);
  79.         appJobDescription.addConfig("ResourceType", "OnDemand");
  80.         appJobDescription.addConfig("InstanceType", "ecs.sn2ne.large");
  81.         appJobDescription.addConfig("InstanceCount", 1);
  83.         appJobDescription.addConfig("MinDiskSize", 40);
  84.         appJobDescription.addConfig("DiskType", "cloud_efficiency");
  85.         appJobDescription.addConfig("MaxRetryCount", 1);
  86.         appJobDescription.addConfig("Timeout", 1000);
  87.         appJobDescription.addConfig("ReserveOnFail", true);
  88.         appJobDescription.addConfig("ClassicNetwork", false);//设置集群网络方式，false为VPC组网
  90.         appJobDescription.addConfig("MinDataDiskSize", 40);
  91.         //注意磁盘类型和 DiskType 保持一致
  92.         appJobDescription.addConfig("DataDiskType", "cloud_efficiency");
  93.         //挂载点根据需要做修改，windows 为 “E:, F:, G: 等”
  94.         appJobDescription.addConfig("DataDiskMountPoint", "/home/mount/");
  95.         return appJobDescription;
  96.     }
  97.     private static DAG getDagDesc() {
  98.         DAG dag = new DAG();
  99.         TaskDescription task = new TaskDescription();
  101.         task.setTaskName("javaSdkTask");
  103.         //设置实例信息
  104.         task.setInstanceCount(1);
  106.         if (IS_AUTO_CLUSTER){
  107.             //设置Auto cluster
  108.             task.setAutoCluster(getAutoCluster());
  109.         }else{
  110.             //设置固定集群信息
  111.             task.setClusterId(ClusterId);
  112.         }
  114.         task.setMaxRetryCount(2);
  115.         task.setTimeout(10000);
  117.         Parameters parameters = new Parameters();
  118.         Command cmd = new Command();
  119.         //设置程序启动命令
  120.         cmd.setCommandLine("python runtask.py 顿雳意当更冁");
  121.         //设置程序启动脚本或者执行文件地址
  122.         cmd.setPackagePath("oss://yuanhyyshenzhen/test/installpackage/runtask.tar.gz");
  124.         //docker 镜像设置方式：推荐使用容器镜像模式
  125.         //1、镜像在oss registry上，设置docker的方式
  126.         //oss registry模式 参数设置好后自行打开注释
  127.         //cmd.addEnvVars("BATCH_COMPUTE_DOCKER_IMAGE", "localhost:5000/yuorBucket/dockers:0.1");//镜像名称;
  128.         //cmd.addEnvVars("BATCH_COMPUTE_DOCKER_REGISTRY_OSS_PATH", "oss://your-bucket/dockers");//设置OSS地址
  130.         //2、镜像在容器镜像仓库，设置docker方式
  131.         //Command.Docker docker = new Command.Docker();
  132.         //docker.setImage("registry.cn-beijing.aliyuncs.com/demotest/test:0.1");
  133.         //cmd.setDocker(docker);
  135.         parameters.setCommand(cmd);
  137.         //设置标准输出 上传的OSS路径
  138.         parameters.setStderrRedirectPath("oss://test/output/error/");
  139.         parameters.setStdoutRedirectPath("oss://test/output/log/");
  141.         InputMappingConfig input = new InputMappingConfig();
  142.         input.setLocale("GBK");
  143.         input.setLock(true);
  144.         parameters.setInputMappingConfig(input);
  146.         task.setParameters(parameters);
  148.         //设置输入OSS路径和本地路径关系
  149.         task.addInputMapping("oss://test/input/", "/home/admin/disk1/");
  151.         //设置输出本地路径和OSS地址
  152.         task.addOutputMapping("/home/admin/disk2/", "oss://test/output/ret/");
  154.         //设置挂载信息
  155.         Mounts mounts = new Mounts();
  157.         MountEntry mountEntry = new MountEntry();
  158.         mountEntry.setDestination("/home/mount");
  159.         mountEntry.setSource("oss://test/mount/");
  160.         mountEntry.setWriteSupport(false);
  162.         mounts.setCacheSupport(false);
  164.         //windows set GBK; Liux set utf-8
  165.         //mounts.setLocale("GBK");
  166.         mounts.setLock(false);
  167.         mounts.addEntries(mountEntry);
  169.         //task.setMounts(mounts);
  171.         dag.addTask(task);
  172.         return dag;
  173.     }
  174.     private static AutoCluster getAutoCluster() {
  175.         AutoCluster autoCluster = new AutoCluster();
  177.         //设置集群镜像信息ECSImageId 在不同region可能会发生变化
  178.         //autoCluster.setECSImageId("m-wz9dk5nao5z3fw6bo9k6");
  179.         //建议使用setImageId接口设置
  180.         autoCluster.setImageId("img-ubuntu");
  182.         autoCluster.setInstanceType("ecs.s3.large");
  183.         autoCluster.setReserveOnFail(true);
  185.         //设置资源类型只有ResourceType为Spot的情况下后面两项有效
  186.         autoCluster.setResourceType("OnDemand");
  187.         //autoCluster.setSpotPriceLimit(5.6f);
  188.         //autoCluster.setSpotStrategy("Spot");
  190.         //设置config信息
  191.         autoCluster.setConfigs(getConfigDesc());
  193.         return autoCluster;
  194.     }
  195.     private static Configs getConfigDesc() {
  196.         Configs configs = new Configs();
  198.         //设置系统磁盘类型以及大型
  199.         Disks disks = new Disks();
  200.         SystemDisk systemDisk = new SystemDisk();
  201.         systemDisk.setSize(40);//GB
  202.         systemDisk.setType("cloud_efficiency");
  203.         disks.setSystemDisk(systemDisk);
  205.         DataDisk dataDisk = new DataDisk();
  206.         dataDisk.setMountPoint("/home/dataDisk/");
  207.         dataDisk.setSize(40);
  208.         dataDisk.setType("cloud_efficiency");
  209.         disks.setDataDisk(dataDisk);
  211.         configs.setDisks(disks);
  213.         //设置网络类型
  214.         Networks networks = new Networks();
  215.         VPC vpc = new VPC();
  216.         vpc.setCidrBlock("10.0.0.0/12");
  217.         networks.setVpc(vpc);
  218.         configs.setNetworks(networks);
  220.         return configs;
  221.     }
  222. }
  1. 执行结果:
  2. ```JSON
  3. {
  4.     jobId: job-000000005BE3E897000007FA00114EE9
  5.     RequestId: null
  6.     StatusCode: 201
  7. }

注意

  1. 本实例代码支持提交 APP 和 DAG 类型作业，支持 AutoCluster 和固定集群类型的作业，提交作业之前根据 业务需要修改开关(IS_DAG_JOB 和 IS_AUTO_CLUSTER)即可。
  2. 若是提交 APP 类型作业，需要在提交作业之前 创建 APP，然后根据 APP 的创建参数做对应修改作业参数，最后进行作业提交。
  3. 提交固定集群作业之前需要先创建集群，修改 ClusterId 为新创建的集群，然后提交作业。
  4. 提交作业前 请确保 OSS 地址填写正确并且已经上传输入或者执行文件到对应的 OSS 路径

本站部分内容转载于网络，版权归原作者所有，转载之目的在于传播更多优秀技术内容，如有侵权请联系QQ/微信：153890879删除，谢谢！
上一篇：同区域跨帐号VPC互通_云连接 CC_快速入门_同区域VPC互通 下一篇：没有了

推荐图文

  • 周排行
  • 月排行
  • 总排行

随机推荐