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

Java SDK 快速开始 - 批量计算

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

简介：Java快速开始例子 本文档将介绍如何使用 Java 版 SDK 来提交一个作业，目的是统计一个日志文件中“INFO”,”WARN”,”ERROR”,”DEBUG”出现的次数。 步骤 作业准备 上传数据文件到 OSS 使用示例代码 编译打包 上传到 OSS 使用 SDK 创建（提交）作业 查看结……

Java快速开始例子

本文档将介绍如何使用 Java 版 SDK 来提交一个作业，目的是统计一个日志文件中“INFO”,”WARN”,”ERROR”,”DEBUG”出现的次数。

步骤

  • 作业准备
    • 上传数据文件到 OSS
    • 使用示例代码
    • 编译打包
    • 上传到 OSS
  • 使用 SDK 创建（提交）作业
  • 查看结果

1. 作业准备

本作业是统计一个日志文件中“INFO”,”WARN”,”ERROR”,”DEBUG”出现的次数。

该作业包含 3 个任务: split, count 和 merge:

  • split 任务会把日志文件分成 3 份。
  • count 任务会统计每份日志文件中“INFO”,”WARN”,”ERROR”,”DEBUG”出现的次数 (count 任务需要配置 InstanceCount 为 3，表示同时启动 3 台机器运行个 count 程序)。
  • merge 任务会把 count 任务的结果统一合并起来。

DAG图例:

DAG图例

(1) 上传数据文件到OSS

下载本例所需的数据: log-count-data.txt

将 log-count-data.txt 上传到:

oss://your-bucket/log-count/log-count-data.txt

  • your-bucket 表示您自己创建的 bucket，本例假设 region 为: cn-shenzhen.
  • 如何上传到 OSS，请参考 OSS 上传文档 。

(2) 使用示例代码

本示例将采用 Java 来编写作业任务，使用 maven 来编译，推荐使用 IDEA：http://www.jetbrains.com/idea/download/ 选择 Community 版本(免费).

示例程序下载：java-log-count.zip

这是一个 maven 工程。

  • 注意：无需修改代码。

(3) 编译打包

运行命令编译打包:

  1. mvn package

即可在 target 得到下面 3 个 jar 包:

  1. batchcompute-job-log-count-1.0-SNAPSHOT-Split.jar
  2. batchcompute-job-log-count-1.0-SNAPSHOT-Count.jar
  3. batchcompute-job-log-count-1.0-SNAPSHOT-Merge.jar

再将 3 个 jar 包，打成一个 tar.gz 压缩包，命令如下:

  1. > cd target  #进入 target 目录
  2. > tar -czf worker.tar.gz *SNAPSHOT-*.jar #打包

运行以下命令，查看包的内容是否正确:

  1. > tar -tvf worker.tar.gz
  2. batchcompute-job-log-count-1.0-SNAPSHOT-Split.jar
  3. batchcompute-job-log-count-1.0-SNAPSHOT-Count.jar
  4. batchcompute-job-log-count-1.0-SNAPSHOT-Merge.jar
  • 注意：BatchCompute 只支持以 tar.gz 为后缀的压缩包, 请注意务必用以上方式（gzip）打包, 否则将会无法解析。

(4) 上传到OSS

本例将 worke.tar.gz 上传到 OSS 的 your-bucket 中:

oss://your-bucket/log-count/worker.tar.gz

  • 如要运行本例子，您需要创建自己的 bucket，并且把 worker.tar.gz 文件上传至您自己创建的 bucket 路径下。

2. 使用SDK创建(提交)作业

(1) 新建一个maven工程

在 pom.xml 中增加以下 dependencies：

  1.     <dependencies>
  2.         <dependency>
  3.             <groupId>com.aliyun</groupId>
  4.             <artifactId>aliyun-java-sdk-batchcompute</artifactId>
  5.             <version>5.2.0</version>
  6.         </dependency>
  8.         <dependency>
  9.             <groupId>com.aliyun</groupId>
  10.             <artifactId>aliyun-java-sdk-core</artifactId>
  11.             <version>3.2.3</version>
  12.         </dependency>
  13.     </dependencies>

(2) 新建一个java类： Demo.java

提交作业需要指定集群 ID 或者使用匿名集群参数。本例子使用匿名集群方式进行。匿名集群需要配置 2 个参数, 其中:

  • 可用的镜像 ID, 可以使用系统提供的 Image，也可以自行制作镜像, 请看 使用镜像 。
  • 实例规格（InstanceType,实例类型），请看 目前支持类型 。

在 OSS 中创建存储 StdoutRedirectPath（程序输出结果）和 StderrRedirectPath（错误日志）的文件路径，本例中创建的路径为

oss://your-bucket/log-count/logs/

  • 如需运行本例，请按照上文所述的变量获取以及与上文对应的您的 OSS 路径对程序中注释中的变量进行修改。

Java SDK 提交程序模板如下，程序中具体参数含义请参照 SDK 接口说明。

Demo.java:

  1. /*
  2. * IMAGE_ID：ECS 镜像，由上文所述获取
  3. * INSTANCE_TYPE: 实例类型，由上文所述获取
  4. * REGION_ID：提交作业的地域，此项需与上文 OSS 存储 worker 的bucket 地域一致
  5. * ACCESS_KEY_ID: AccessKeyId 可以由上文所述获取
  6. * ACCESS_KEY_SECRET: AccessKeySecret 可以由上文所述获取
  7. * WORKER_PATH：由上文所述打包上传的 worker 的 OSS 存储路径
  8. * LOG_PATH：错误反馈和 task 输出的存储路径，logs 文件需事先自行创建
  9.  */
  10.  import com.aliyuncs.batchcompute.main.v20151111.*;
  11.  import com.aliyuncs.batchcompute.model.v20151111.*;
  12.  import com.aliyuncs.batchcompute.pojo.v20151111.*;
  13.  import com.aliyuncs.exceptions.ClientException;
  15.  import java.util.ArrayList;
  16.  import java.util.List;
  18.  public class Demo {
  20.      static String IMAGE_ID = "img-ubuntu";;  //这里填写您的 ECS 镜像 ID
  21.      static String INSTANCE_TYPE = "ecs.sn1.medium"; //根据 region 填写合适的 InstanceType
  23.      static String REGION_ID = "cn-shenzhen";   //这里填写 region
  24.      static String ACCESS_KEY_ID = "";  //"your-AccessKeyId";   这里填写您的 AccessKeyId
  25.      static String ACCESS_KEY_SECRET = ""; //"your-AccessKeySecret";    这里填写您的 AccessKeySecret
  26.      static String WORKER_PATH = ""; //"oss://your-bucket/log-count/worker.tar.gz"; //   这里填写您上传的 worker.tar.gz 的 OSS 存储路径
  27.      static String LOG_PATH = ""; // "oss://your-bucket/log-count/logs/"; //   这里填写您创建的错误反馈和 task 输出的 OSS 存储路径
  28.      static String MOUNT_PATH = ""; // "oss://your-bucket/log-count/";
  30.      public static void main(String[] args){
  32.          /** 构造 BatchCompute 客户端 */
  33.          BatchCompute client = new BatchComputeClient(REGION_ID, ACCESS_KEY_ID, ACCESS_KEY_SECRET);
  35.          try{
  37.              /** 构造 Job 对象 */
  38.              JobDescription jobDescription = genJobDescription();
  40.              //创建 Job
  41.              CreateJobResponse response = client.createJob(jobDescription);
  43.              //创建成功后，返回 jobId
  44.              String jobId = response.getJobId();
  46.              System.out.println("Job created success, got jobId: "+jobId);
  49.              //查询 job 状态
  50.              GetJobResponse getJobResponse = client.getJob(jobId);
  52.              Job job = getJobResponse.getJob();
  54.              System.out.println("Job state:"+job.getState());
  56.          } catch (ClientException e) {
  57.              e.printStackTrace();
  59.              System.out.println("Job created failed, errorCode:"+ e.getErrCode()+", errorMessage:"+e.getErrMsg());
  60.          }
  61.      }
  63.      private static JobDescription genJobDescription(){
  67.          JobDescription jobDescription = new JobDescription();
  69.          jobDescription.setName("java-log-count");
  70.          jobDescription.setPriority(0);
  71.          jobDescription.setDescription("log-count demo");
  72.          jobDescription.setJobFailOnInstanceFail(true);
  73.          jobDescription.setType("DAG");
  75.          DAG taskDag = new DAG();
  78.          /** 添加 split task */
  80.          TaskDescription splitTask =  genTaskDescription();
  81.          splitTask.setTaskName("split");
  82.          splitTask.setInstanceCount(1);
  83.          splitTask.getParameters().getCommand().setCommandLine("java -jar batchcompute-job-log-count-1.0-SNAPSHOT-Split.jar");
  84.          taskDag.addTask(splitTask);
  86.          /** 添加 count task */
  87.          TaskDescription countTask =  genTaskDescription();
  88.          countTask.setTaskName("count");
  89.          countTask.setInstanceCount(3);
  90.          countTask.getParameters().getCommand().setCommandLine("java -jar batchcompute-job-log-count-1.0-SNAPSHOT-Count.jar");
  91.          taskDag.addTask(countTask);
  93.          /** 添加 merge task */
  94.          TaskDescription mergeTask =  genTaskDescription();
  95.          mergeTask.setTaskName("merge");
  96.          mergeTask.setInstanceCount(1);
  97.          mergeTask.getParameters().getCommand().setCommandLine("java -jar batchcompute-job-log-count-1.0-SNAPSHOT-Merge.jar");
  98.          taskDag.addTask(mergeTask);
  102.          /** 添加 Task 依赖:  split-->count-->merge  */
  104.          List<String> taskNameTargets = new ArrayList();
  105.          taskNameTargets.add("merge");
  106.          taskDag.addDependencies("count", taskNameTargets);
  108.          List<String> taskNameTargets2 = new ArrayList();
  109.          taskNameTargets2.add("count");
  110.          taskDag.addDependencies("split", taskNameTargets2);
  112.          //dag
  113.          jobDescription.setDag(taskDag);
  115.          return jobDescription;
  116.      }
  118.      private static TaskDescription genTaskDescription(){
  120.          AutoCluster autoCluster = new AutoCluster();
  121.          autoCluster.setInstanceType(INSTANCE_TYPE);
  122.          autoCluster.setImageId(IMAGE_ID);
  123.          //autoCluster.setResourceType("OnDemand");
  125.          TaskDescription task = new TaskDescription();
  126.          //task.setTaskName("Find");
  128.         //如果使用 VPC，需要配置 cidrBlock, 请确保 IP 段不冲突
  129.         Configs configs = new Configs();
  130.         Networks networks = new Networks();
  131.         VPC vpc = new VPC();
  132.         vpc.setCidrBlock("192.168.0.0/16");
  133.         networks.setVpc(vpc);
  134.         configs.setNetworks(networks);
  135.         autoCluster.setConfigs(configs);
  137.          //打包上传的作业的 OSS 全路径
  138.          Parameters p = new Parameters();
  139.          Command cmd = new Command();
  140.          //cmd.setCommandLine("");
  141.          //打包上传的作业的 OSS 全路径
  142.          cmd.setPackagePath(WORKER_PATH);
  143.          p.setCommand(cmd);
  144.          //错误反馈存储路径
  145.          p.setStderrRedirectPath(LOG_PATH);
  146.          //最终结果输出存储路
  147.          p.setStdoutRedirectPath(LOG_PATH);
  149.          task.setParameters(p);
  150.          task.addInputMapping(MOUNT_PATH, "/home/input");
  151.          task.addOutputMapping("/home/output",MOUNT_PATH);
  153.          task.setAutoCluster(autoCluster);
  154.          //task.setClusterId(clusterId);
  155.          task.setTimeout(30000); /* 30000 秒*/
  156.          task.setInstanceCount(1); /** 使用 1 个实例来运行 */
  158.          return task;
  159.      }
  160.  }

正常输出样例：

  1. Job created success, got jobId: job-01010100010192397211
  2. Job state:Waiting

3. 查看作业状态

您可以用 SDK 中的 获取作业信息 方法获取作业状态：

  1. //查询 job 状态
  2. GetJobResponse getJobResponse = client.getJob(jobId);
  3. Job job = getJobResponse.getJob();
  4. System.out.println("Job state:"+job.getState());

Job 的 state 可能为：Waiting、Running、Finished、Failed、Stopped.

4. 查看结果

您可以登录 batchcompute 控制台 查看 job 状态。

Job 运行结束，您可以登录 OSS 控制台 查看your-bucket 这个 bucket 下面的这个文件：/log-count/merge_result.json。

内容应该如下：

  1. {"INFO": 2460, "WARN": 2448, "DEBUG": 2509, "ERROR": 2583}

您也可以使用 OSS 的 SDK 来获取结果。


本站部分内容转载于网络，版权归原作者所有，转载之目的在于传播更多优秀技术内容，如有侵权请联系QQ/微信：153890879删除，谢谢！
上一篇：公网域名解析_云解析服务 DNS_产品介绍 下一篇：没有了

推荐图文

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

随机推荐