更新时间:2022-12-07 GMT+08:00

新建Flink Jar作业

功能介绍

该API用于创建用户自定义的作业,目前支持jar格式,运行在独享队列中。

URI

  • URI格式

    POST /v1.0/{project_id}/streaming/flink-jobs

  • 参数说明
    表1 URI参数说明

    参数名称

    是否必选

    参数类型

    说明

    project_id

    String

    项目编号,用于资源隔离。获取方式请参考获取项目ID

请求消息

表2 参数说明

参数名称

是否必选

参数类型

说明

name

String

作业名称。长度限制:1-57个字符。

desc

String

作业描述。长度限制:0-512个字符。

queue_name

String

队列名称。长度限制:0-128个字符。

cu_number

Integer

用户为作业选择的CU数量。

manager_cu_number

Integer

用户为作业选择的管理节点CU数量,对应为flink jobmanager数量。默认值为“1”。

parallel_number

Integer

用户为作业选择的并发量。

log_enabled

Boolean

是否开启作业日志。

  • 开启:true
  • 关闭:false
  • 默认:false

obs_bucket

String

“log_enabled”“true”时, 用户授权保存作业日志的OBS桶名。

smn_topic

String

当作业异常时,向该SMN主题推送告警信息。

main_class

String

作业入口类。

entrypoint_args

String

作业入口类参数,多个参数之间空格分隔。

restart_when_exception

Boolean

是否开启异常重启功能,默认值为“false”。

entrypoint

String

用户已上传到DLI资源管理系统的程序包名,用户自定义作业主类所在的jar包。

dependency_jars

Array of Strings

用户已上传到DLI资源管理系统的程序包名,用户自定义作业的其他依赖包。

示例“myGroup/test.jar,myGroup/test1.jar”。

dependency_files

Array of Strings

用户已上传到DLI资源管理系统的资源包名,用户自定义作业的依赖文件。

示例:"myGroup/test.cvs,myGroup/test1.csv"。

通过在应用程序中添加以下内容可访问对应的依赖文件。其中,“fileName”为需要访问的文件名,“ClassName”为需要访问该文件的类名。

ClassName.class.getClassLoader().getResource("userData/fileName")

tm_cus

Integer

每个taskmanager的CU数,默认值为“1”。

tm_slot_num

Integer

每个taskmanager的slot数,默认值为“(parallel_number*tm_cus)/(cu_number-manager_cu_number)”。

resume_checkpoint

Boolean

异常重启是否从checkpoint恢复。

resume_max_num

Integer

异常重试最大次数,单位:次/小时。取值范围:-1或大于0。默认值为“-1”,表示无限次数。

checkpoint_path

String

用户Jar中checkpoint的储存地址,不同作业路径需要保持不同。

tags

Array of Objects

Flink jar作业的标签。具体请参考表3

runtime_config

String

Flink作业运行时自定义优化参数。

表3 tags参数

参数名称

是否必选

参数类型

说明

key

String

标签的键。

value

String

标签的值。

响应消息

表4 响应参数说明

参数名称

是否必选

参数类型

说明

is_success

Boolean

执行请求是否成功。“true”表示请求执行成功。

message

String

消息内容。

job

Object

作业状态信息。具体请参考表5

表5 job参数说明

参数名称

是否必选

参数类型

说明

job_id

Long

作业ID。

status_name

String

当前状态名称。

status_desc

String

当前状态描述。包含异常状态原因及建议。

请求示例

{
    "name": "test",
    "desc": "job for test",
    "queue_name": "testQueue",
    "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 -rate 3",
    "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所示。

表6 状态码

状态码

描述

200

新建Flink自定义作业成功。

400

输入参数无效。

错误码

调用接口出错后,将不会返回上述结果,而是返回错误码和错误信息,更多介绍请参见错误码