本节介绍如何通过OpenAPI的方式创建作业、运行作业以及查看结果。

使用场景

用户已经在华东2(杭州)创建了项目,现需要在项目中编辑作业并运行。

现有集群的基本配置为:
  • 项目名称为emr_openapi_demo_project,项目ID为FP-D18E9976D5A****。如未创建项目,请创建新项目,详情请参见项目管理
  • 项目管理的集群名为emr_openapi_demo, 集群ID为C-69CB0546800F****。
说明
  • 通过CreateFlowProject创建作业时,可以在返回结果中找到项目的项目ID。
  • 通过ListFlowProject可以查询该Region下所有数据开发项目的列表,也可以在返回结果中找到项目ID。

示例

  • Java
    1. 创建Hive_SQL类型的作业。调用CreateFlowProject创建作业时,下面几个参数是必须的。
      参数 描述
      RegionId 地域ID。例如,cn-hangzhou。
      ProjectId 项目ID。
      Name 自定义作业名字。例如,emr_openapi_hivejob。
      Type 作业类型,目前支持:MR、Spark、Hive_SQL、Hive、Pig、Sqoop、Spark_SQL、Spark_Streaming、Shell。
      Description 作业的描述。
      说明 ClusterId对于创建作业来说是可选项。
      import com.aliyuncs.DefaultAcsClient;
      import com.aliyuncs.IAcsClient;
      import com.aliyuncs.exceptions.ClientException;
      import com.aliyuncs.exceptions.ServerException;
      import com.aliyuncs.profile.DefaultProfile;
      import com.google.gson.Gson;
      import java.util.*;
      import com.aliyuncs.emr.model.v20160408.*;
      public class CreateFlowJob {
          public static void main(String[] args) {
              DefaultProfile profile = DefaultProfile.getProfile("cn-hangzhou", "<accessKeyId>", "<accessSecret>");
              IAcsClient client = new DefaultAcsClient(profile);
              CreateFlowJobRequest request = new CreateFlowJobRequest();
              request.setRegionId("cn-hangzhou");
              request.setProjectId("FP-D18E9976D5A****");
              request.setName("emr_openapi_hivejob");
              request.setDescription("Hive_SQL job created by OpenAPI");
              request.setType("HIVE_SQL");
              request.setAdhoc(false);
              try {
                  CreateFlowJobResponse response = client.getAcsResponse(request);
                  System.out.println(new Gson().toJson(response));
              } catch (ServerException e) {
                  e.printStackTrace();
              } catch (ClientException e) {
                  System.out.println("ErrCode:" + e.getErrCode());
                  System.out.println("ErrMsg:" + e.getErrMsg());
                  System.out.println("RequestId:" + e.getRequestId());
              }
          }
      }
      执行后返回该作业的ID,即JobId。
      {
          "RequestId": "01B0A835-C6AB-4166-8E43-511514D8FAE0",
          "Id": "FJ-E4157D27791D****"
      }
    2. 执行作业。调用SubmitFlowJob提交运行作业,每次只允许存在一个正在运行的实例,以下的几个参数是必须要填写的。
      参数 描述
      ClusterId 执行作业的集群ID。
      JobId 作业ID,即步骤1中返回的结果FJ-E4157D27791D****。
      ProjectId 工作流项目ID。
      RegionId 地域ID。例如,cn-hangzhou。
      import com.aliyuncs.DefaultAcsClient;
      import com.aliyuncs.IAcsClient;
      import com.aliyuncs.exceptions.ClientException;
      import com.aliyuncs.exceptions.ServerException;
      import com.aliyuncs.profile.DefaultProfile;
      import com.google.gson.Gson;
      import java.util.*;
      import com.aliyuncs.emr.model.v20160408.*;
      public class SubmitFlowJob {
          public static void main(String[] args) {
              DefaultProfile profile = DefaultProfile.getProfile("cn-hangzhou", "<accessKeyId>", "<accessSecret>");
              IAcsClient client = new DefaultAcsClient(profile);
              SubmitFlowJobRequest request = new SubmitFlowJobRequest();
              request.setRegionId("cn-hangzhou");
              request.setProjectId("FP-D18E9976D5A****");
              request.setJobId("FJ-E4157D27791D****");
              request.setClusterId("C-69CB0546800F****");
              try {
                  SubmitFlowJobResponse response = client.getAcsResponse(request);
                  System.out.println(new Gson().toJson(response));
              } catch (ServerException e) {
                  e.printStackTrace();
              } catch (ClientException e) {
                  System.out.println("ErrCode:" + e.getErrCode());
                  System.out.println("ErrMsg:" + e.getErrMsg());
                  System.out.println("RequestId:" + e.getRequestId());
              }
          }
      }
      返回FlowJobInstanceID。
      {
          "RequestId": "15BBB0DC-EEC5-4CE4-B4FA-A1D9827F8808",
          "Id": "FJI-54FEBB063136****"
      }
    3. 当作业类型为Hive_SQL或Spark_SQL时,可以调用ListFlowNodeSqlResult得到查询的结果。

      这里需要注意的是,ListFlowNodeSqlResult返回的结果最多只有200行。

      需提供以下参数。

      参数 描述
      RegionId 地域ID。例如,cn-hangzhou。
      ProjectId 工作流项目ID。
      NodeInstanceId 节点实例ID。例如,在步骤2中返回的实例ID FJI-54FEBB063136****。
      import com.aliyuncs.DefaultAcsClient;
      import com.aliyuncs.IAcsClient;
      import com.aliyuncs.exceptions.ClientException;
      import com.aliyuncs.exceptions.ServerException;
      import com.aliyuncs.profile.DefaultProfile;
      import com.google.gson.Gson;
      import java.util.*;
      import com.aliyuncs.emr.model.v20160408.*;
      public class ListFlowNodeSqlResult {
          public static void main(String[] args) {
              DefaultProfile profile = DefaultProfile.getProfile("cn-hangzhou", "<accessKeyId>", "<accessSecret>");
              IAcsClient client = new DefaultAcsClient(profile);
              ListFlowNodeSqlResultRequest request = new ListFlowNodeSqlResultRequest();
              request.setRegionId("cn-hangzhou");
              request.setProjectId("FP-D18E9976D5A****");
              request.setNodeInstanceId("FJI-54FEBB063136****");
              try {
                  ListFlowNodeSqlResultResponse response = client.getAcsResponse(request);
                  System.out.println(new Gson().toJson(response));
              } catch (ServerException e) {
                  e.printStackTrace();
              } catch (ClientException e) {
                  System.out.println("ErrCode:" + e.getErrCode());
                  System.out.println("ErrMsg:" + e.getErrMsg());
                  System.out.println("RequestId:" + e.getRequestId());
              }
          }
      }
    4. 查看作业执行的详细信息,例如,执行状态,时间和时长等。需提供以下参数。
      参数 描述
      RegionId 地域ID。例如,cn-hangzhou。
      ProjectId 工作流项目ID。
      NodeInstanceId 节点实例ID。例如,在步骤2中返回的实例ID FJI-54FEBB063136****。
      import com.aliyuncs.DefaultAcsClient;
      import com.aliyuncs.IAcsClient;
      import com.aliyuncs.exceptions.ClientException;
      import com.aliyuncs.exceptions.ServerException;
      import com.aliyuncs.profile.DefaultProfile;
      import com.google.gson.Gson;
      import java.util.*;
      import com.aliyuncs.emr.model.v20160408.*;
      public class DescribeFlowNodeInstance {
          public static void main(String[] args) {
              DefaultProfile profile = DefaultProfile.getProfile("cn-hangzhou", "<accessKeyId>", "<accessSecret>");
              IAcsClient client = new DefaultAcsClient(profile);
              DescribeFlowNodeInstanceRequest request = new DescribeFlowNodeInstanceRequest();
              request.setRegionId("cn-hangzhou");
              request.setId("FJI-54FEBB063136****");
              request.setProjectId("FP-D18E9976D5A****");
              try {
                  DescribeFlowNodeInstanceResponse response = client.getAcsResponse(request);
                  System.out.println(new Gson().toJson(response));
              } catch (ServerException e) {
                  e.printStackTrace();
              } catch (ClientException e) {
                  System.out.println("ErrCode:" + e.getErrCode());
                  System.out.println("ErrMsg:" + e.getErrMsg());
                  System.out.println("RequestId:" + e.getRequestId());
              }
          }
      }
  • Python
    1. 创建Hive_SQL类型的作业,在创建作业时,下面几个参数是必须的。
      参数 描述
      RegionId 地域ID。例如,cn-hangzhou。
      ProjectId 项目ID。
      Name 自定义作业名字。例如,emr_openapi_hivejob。
      Type 作业类型,目前支持:MR、Spark、Hive_SQL、Hive、Pig、Sqoop、Spark_SQL、Spark_Streaming、Shell。
      Description 作业的描述。
      说明 ClusterId对于创建作业来说是可选项。
      #!/usr/bin/env python
      #coding=utf-8
      
      from aliyunsdkcore.client import AcsClient
      from aliyunsdkcore.acs_exception.exceptions import ClientException
      from aliyunsdkcore.acs_exception.exceptions import ServerException
      from aliyunsdkemr.request.v20160408.CreateFlowJobRequest import CreateFlowJobRequest
      
      client = AcsClient('<accessKeyId>', '<accessSecret>', 'cn-hangzhou')
      
      request = CreateFlowJobRequest()
      request.set_accept_format('json')
      
      request.set_ProjectId("FP-D18E9976D5A****")
      request.set_Name("emr_openapi_hivejob")
      request.set_Description("Hive_SQL job created by OpenAPI")
      request.set_Type("HIVE_SQL")
      request.set_Adhoc(Flase)
      
      response = client.do_action_with_exception(request)
      # python2:  print(response) 
      print(str(response, encoding='utf-8'))
      执行后返回该作业的ID,即JobId。
      {
          "RequestId": "01B0A835-C6AB-4166-8E43-511514D8FAE0",
          "Id": "FJ-E4157D27791D****"
      }
    2. 执行作业。调用SubmitFlowJob提交运行作业,每次只允许存在一个正在运行的实例,以下的几个参数是必须要填写的。
      参数 描述
      ClusterId 执行作业的集群。
      JobId 作业ID,即步骤1中返回的结果FJ-E4157D27791D****。
      ProjectId 工作流项目ID。
      RegionId 地域ID。例如,cn-hangzhou。
      #!/usr/bin/env python
      #coding=utf-8
      
      from aliyunsdkcore.client import AcsClient
      from aliyunsdkcore.acs_exception.exceptions import ClientException
      from aliyunsdkcore.acs_exception.exceptions import ServerException
      from aliyunsdkemr.request.v20160408.SubmitFlowJobRequest import SubmitFlowJobRequest
      
      client = AcsClient('<accessKeyId>', '<accessSecret>', 'cn-hangzhou')
      
      request = SubmitFlowJobRequest()
      request.set_accept_format('json')
      
      request.set_ProjectId("FP-D18E9976D5A****")
      request.set_JobId("FJ-E4157D27791D****")
      request.set_ClusterId("C-69CB0546800F****")
      
      response = client.do_action_with_exception(request)
      # python2:  print(response) 
      print(str(response, encoding='utf-8'))
      返回FlowJobInstanceID。
      {
          "RequestId": "15BBB0DC-EEC5-4CE4-B4FA-A1D9827F8808",
          "Id": "FJI-54FEBB063136****"
      }
    3. 当FlowJob的类型为Hive_SQL或者Spark_SQL时,可以调用ListFlowNodeSqlResult得到查询的结果值。

      这里需要注意的是,ListFlowNodeSqlResult返回的结果最多只有200行。

      需提供以下参数。

      参数 描述
      RegionId 地域ID。例如,cn-hangzhou。
      ProjectId 工作流项目ID。
      NodeInstanceId 节点实例ID。例如,在步骤2中返回的实例ID FJI-54FEBB063136****。
      #!/usr/bin/env python
      #coding=utf-8
      
      from aliyunsdkcore.client import AcsClient
      from aliyunsdkcore.acs_exception.exceptions import ClientException
      from aliyunsdkcore.acs_exception.exceptions import ServerException
      from aliyunsdkemr.request.v20160408.ListFlowNodeSqlResultRequest import ListFlowNodeSqlResultRequest
      
      client = AcsClient('<accessKeyId>', '<accessSecret>', 'cn-hangzhou')
      
      request = ListFlowNodeSqlResultRequest()
      request.set_accept_format('json')
      
      request.set_ProjectId("FP-D18E9976D5A****")
      request.set_NodeInstanceId("FJI-54FEBB063136****")
      
      response = client.do_action_with_exception(request)
      # python2:  print(response) 
      print(str(response, encoding='utf-8'))
    4. 查看作业执行的详细信息,例如执行状态,时间,时长等。需提供以下参数。
      参数 描述
      RegionId 地域ID。例如,cn-hangzhou。
      ProjectId 工作流项目ID。
      NodeInstanceId 节点实例ID。例如,在步骤2中返回的实例ID FJI-54FEBB063136****。
      #!/usr/bin/env python
      #coding=utf-8
      
      from aliyunsdkcore.client import AcsClient
      from aliyunsdkcore.acs_exception.exceptions import ClientException
      from aliyunsdkcore.acs_exception.exceptions import ServerException
      from aliyunsdkemr.request.v20160408.DescribeFlowNodeInstanceRequest import DescribeFlowNodeInstanceRequest
      
      client = AcsClient('<accessKeyId>', '<accessSecret>', 'cn-hangzhou')
      
      request = DescribeFlowNodeInstanceRequest()
      request.set_accept_format('json')
      
      request.set_Id("FJI-54FEBB063136****")
      request.set_ProjectId("FP-D18E9976D5A****")
      
      response = client.do_action_with_exception(request)
      # python2:  print(response) 
      print(str(response, encoding='utf-8'))

参考OpenAPI