FlinkJob作业接口应用示例

概述

本文指导用户通过调用API的方式，执行Flink作业的完整流程，包括启动Flink作业、查询作业状态和查询Flink作业详情等。

本流程假设终端租户已经在Console界面完成AI DataLake的服务授权，且已创建工作空间和Flink端点。API的调用方法请参见如何调用API。

前提条件

• workspace_id：工作空间ID，可在控制台的工作空间管理页面查看。

• endpoint_name：Flink端点名称，可在控制台的端点管理页面查看，或参考查询所有端点列表 - ListV2AllEndpoints获取。

启动Flink作业

调用启动Flink作业接口，提交Flink作业，同时记录接口返回的“job_id”。

• 请求示例

  POST /v2/workspaces/{workspace_id}/flink-jobs

  {workspace_id}：工作空间ID，用于资源隔离。

• Body：

  {
      "name": "flink-job-test",
      "endpoint_name": "flink-test",
      "execution_mode": "STREAMING",
      "resource_config": {
          "parallelism": 1,
          "jobmanager_resource_spec": {
              "cpu": 500,
              "memory": 1024,
              "disk": 30
          },
          "taskmanager_resource_spec": {
              "cpu": 500,
              "memory": 1024,
              "disk": 30
          }
      },
      "job_config": {
          "type": "FLINK_JAR_JOB",
          "main_jar": "obs://infotest/flink/jars/WindowJoin.jar",
          "main_class": "org.apache.flink.streaming.examples.join.WindowJoin",
          "main_args": [
              "--output",
              "/tmp/result.txt"
          ]
      },
      "logging_config": {
          "root_logger_level": "INFO"
      },
      "flink_config": {
          "taskmanager.memory.framework.heap.size": "64mb",
          "taskmanager.memory.framework.off-heap.size": "64mb",
          "taskmanager.memory.managed.fraction": "0.2",
          "taskmanager.memory.network.min": "32mb",
          "taskmanager.memory.network.max": "64mb"
      }
  }

• 响应示例：

  {
         "job_id": "019fa293-6a76-xxxx-xxxx-10e8fc29bba9"
  }

查询Flink作业的状态

调用查询Flink作业状态接口，查询作业状态，等待作业状态变为SUCCEED，表示作业执行完成。

• 请求示例：

  GET /v2/workspaces/{workspace_id}/flink-jobs/{job_id}/state

  • {workspace_id}：工作空间ID，用于资源隔离。
  • {job_id}：启动Flink作业响应信息中的“job_id”参数值。

• 响应示例：

  {  
         "job_id": "019fa293-6a76-xxxx-xxxx-10e8fc29bba9",
         "state": "SUCCEED"
  }

查询Flink作业详情

调用查看Flink作业详情接口，查看作业的完整配置信息和执行结果。

• 请求示例：

  GET /v2/workspaces/{workspace_id}/flink-jobs/{job_id}

  • {workspace_id}：工作空间ID，用于资源隔离。
  • {job_id}：启动Flink作业响应信息中的“job_id”参数值。
• 响应示例：

  {
      "job_id": "019fa293-6a76-xxxx-xxxx-10e8fc29bba9",
      "client_token": "019fa293-6a76-xxxx-xxxx-1fdfa6b993a4",
      "name": "flink-job-test",
      "endpoint_name": "flink-test",
      "status": {
          "state": "SUCCEED"
      },
      "job_agency": "ai_datalake_alpha_admin_trust",
      "obs_bucket_name": "ai4db",
      "job_config": {
          "type": "FLINK_JAR_JOB",
          "main_jar": "obs://infotest/flink/jars/WindowJoin.jar",
          "main_class": "org.apache.flink.streaming.examples.join.WindowJoin",
          "main_args": [
              "--output",
              "/tmp/result.txt"
          ]
      },
      "resource_config": {
          "parallelism": 1,
          "jobmanager_resource_spec": {
              "cpu": 500,
              "memory": 1024,
              "disk": 30
          },
          "taskmanager_resource_spec": {
              "cpu": 500,
              "memory": 1024,
              "disk": 30
          }
      },
      "flink_config": {
          "high-availability.storageDir": "obs://ai4db/AIDataLake/Workspace_57fd5733-20b4-xxxx-xxxx-c6e7f39b1891/Flink/Endpoint_bd0f32d1-25aa-461e-xxxx-78daa554dd5c/jobs/ha/019fa293-6a76-xxxx-xxxx-10e8fc29bba9",
          "taskmanager.memory.framework.off-heap.size": "64mb",
          "taskmanager.memory.network.max": "64mb",
          "execution.checkpointing.savepoint-dir": "obs://ai4db/AIDataLake/Workspace_57fd5733-20b4-xxxx-xxxx-c6e7f39b1891/Flink/Endpoint_bd0f32d1-25aa-461e-xxxx-78daa554dd5c/jobs/savepoint/019fa293-6a76-xxxx-xxxx-10e8fc29bba9",
          "taskmanager.memory.network.min": "32mb",
          "taskmanager.memory.framework.heap.size": "64mb",
          "taskmanager.memory.managed.fraction": "0.2",
          "execution.checkpointing.num-retained": "6",
          "execution.checkpointing.dir": "obs://ai4db/AIDataLake/Workspace_57fd5733-20b4-xxxx-xxxx-c6e7f39b1891/Flink/Endpoint_bd0f32d1-25aa-461e-xxxx-78daa554dd5c/jobs/checkpoint/019fa293-6a76-xxxx-xxxx-10e8fc29bba9"
      },
      "execution_mode": "STREAMING",
      "restore_strategy": {
          "restore_type": "NONE"
      },
      "logging_config": {
          "root_logger_level": "INFO"
      },
      "metric": {
          "total_memory": 1948,
          "total_cpu": 900,
          "total_disk": -1
      },
      "create_time": 1785138997878,
      "create_user": {
          "user_id": "019e19fb78887b57bxxxxxxxxcd5a002",
          "user_name": "aid_developer03"
      },
      "update_time": 1785138997878,
      "update_user": {
          "user_id": "019e19fb78887b57bxxxxxxxxcd5a002",
          "user_name": "aid_developer03"
      },
      "log_url": "obs://ai4db/AIDataLake/Workspace_57fd5733-20b4-xxxx-xxxx-c6e7f39b1891/Flink/Endpoint_bd0f32d1-25aa-461e-xxxx-78daa554dd5c/jobs/log/019fa293-6a76-7a9f-xxxx-10e8fc29bba9",
      "description": "",
      "start_time": 1785139107050,
      "end_time": 0
  }

取消Flink作业执行

若作业执行异常需要终止，调用取消Flink作业接口取消正在执行的作业。仅支持取消处于QUEUED或RUNNING状态的作业。

• 请求示例：

  POST /v2/workspaces/{workspace_id}/flink-jobs/{job_id}/cancel

  • {workspace_id}：工作空间ID，用于资源隔离。
  • {job_id}：启动Flink作业响应信息中的“job_id”参数值。
• 响应示例：无响应体，HTTP状态码204表示取消成功。