创建并提交Flink作业

场景描述

本章节指导用户通过API创建并运行Flink自定义作业。

约束限制

• 新队列第一次运行作业时，需要一定的时间，通常为6~10分钟。

涉及接口

• 创建弹性资源池：创建弹性资源池。
• 创建队列：在弹性资源池中添加队列。
• 上传分组资源：上传Flink自定义作业所需的资源包。
• 查询组内资源包：确认上传的资源包是否正确。
• 创建Flink自定义作业：创建Flink自定义作业。
• 批量运行作业：运行Flink自定义作业。

操作步骤

1. 创建弹性资源池elastic_pool_dli
   • 接口相关信息

     URI格式：POST /v3/{project_id}/elastic-resource-pools

     • {project_id}信息请从获取项目ID获取。
     • 请求参数说明详情，请参见创建弹性资源池。
   • 请求示例
     • 描述：在项目ID为48cc2c48765f481480c7db940d6409d1的项目下创建一个名称为elastic_pool_dli的弹性资源池。
     • 示例URL：POST https://{endpoint}/v3/48cc2c48765f481480c7db940d6409d1/elastic-resource-pools

       {endpoint}信息请从地区和终端节点获取。

     • Body：

       {
         "elastic_resource_pool_name" : "elastic_pool_dli",
         "description" : "test",
         "cidr_in_vpc" : "172.16.0.0/14",
         "charging_mode" : "1",
         "max_cu" : 64,
         "min_cu" : 64
       }

   • 响应示例

     {
       "is_success": true,
       "message": ""
     }

2. 在弹性资源池中添加队列queue1
   • 接口相关信息

     URI格式：POST /v1.0/{project_id}/queues

     • {project_id}信息请从获取项目ID获取。
     • 请求参数说明详情，请参见创建队列。
   • 请求示例
     • 描述：在项目ID为48cc2c48765f481480c7db940d6409d1的项目下创建一个名称为queue1的弹性资源池。
     • 示例URL：POST https://{endpoint}/v1.0/48cc2c48765f481480c7db940d6409d1/queues

       {endpoint}信息请从地区和终端节点获取。

     • Body：

       {
           "queue_name": "queue1",
           "queue_type": "sql",
           "description": "test",
           "cu_count": 16,
           "enterprise_project_id": "elastic_pool_dli"
       }

   • 响应示例

     {
       "is_success": true,
       "message": ""
     }

3. 上传Flink自定义作业资源包。具体请参考3
4. 查询组内资源包。具体请参考4
5. 创建Flink自定义作业。
   • 接口相关信息

     URI格式：POST /v1.0/{project_id}/streaming/flink-jobs

     • {project_id}信息请从获取项目ID获取。
     • 请求参数说明详情，请参见创建数据库（废弃）。
   • 请求示例
     • 描述：在项目ID为48cc2c48765f481480c7db940d6409d1项目下，创建Flink自定义作业。
     • 示例URL：POST https://{endpoint}/v1.0/48cc2c48765f481480c7db940d6409d1/streaming/flink-jobs

       {endpoint}信息请从地区和终端节点获取。

     • Body：

       {
           "name": "test",
           "desc": "job for test",
           "queue_name": "queue1",
           "manager_cu_number": 1,
           "cu_number": 2,
           "parallel_number": 1,
           "tm_cus": 1,
           "tm_slot_num": 1,
           "log_enabled": true,
           "obs_bucket": "bucketName",
           "smn_topic": "topic",
           "main_class": "org.apache.flink.examples.streaming.JavaQueueStream",
           "restart_when_exception": false,
           "entrypoint": "javaQueueStream.jar",
           "entrypoint_args":"-windowSize 2000 -rate3",
           "dependency_jars": [
               "myGroup/test.jar",
               "myGroup/test1.jar"
           ],
           "dependency_files": [
               "myGroup/test.csv",
               "myGroup/test1.csv"
           ]
       }

   • 响应示例

     {
       "is_success": true,
       "message": "新建flink作业成功",
       "job": {
         "job_id": 138,
         "status_name": "job_init",
         "status_desc": ""
       }
     }

6. 批量运行作业。
   • 接口相关信息

     URI格式：POST /v1.0/{project_id}/streaming/jobs/run

     • {project_id}信息请从获取项目ID获取。
     • 请求参数说明详情，请参见批量运行作业。
   • 请求示例
     • 描述：在项目ID为48cc2c48765f481480c7db940d6409d1项目下，运行job_id为298765和298766的作业。
     • 示例URL：POST https://{endpoint}/v1.0/48cc2c48765f481480c7db940d6409d1/streaming/jobs/run

       {endpoint}信息请从地区和终端节点获取。

     • Body：

       {
           "job_ids": [131,130,138,137],
           "resume_savepoint": true
       }

   • 响应示例

     [
         {
             "is_success": "true",
             "message": "作业提交请求下发成功"
         },
         {
             "is_success": "true",
             "message": "作业提交请求下发成功"
         },
         {
             "is_success": "true",
             "message": "作业提交请求下发成功"
         },
         {
             "is_success": "true",
             "message": "作业提交请求下发成功"
         }
     ]