Updated on 2024-03-25 GMT+08:00

Creating a Flink Jar job

Function

This API is used to create custom jobs, which currently support the JAR format and run in dedicated queues.

URI

  • URI format

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

  • Parameter description
    Table 1 URI parameter

    Parameter

    Mandatory

    Type

    Description

    project_id

    Yes

    String

    Project ID, which is used for resource isolation. For details about how to obtain its value, see Obtaining a Project ID.

Request

Table 2 Parameter description

Parameter

Mandatory

Type

Description

name

Yes

String

Name of the job. The value can contain 1 to 57 characters.

desc

No

String

Job description. Length range: 0 to 512 characters.

queue_name

No

String

Name of a queue. The value can contain 0 to 128 characters.

cu_number

No

Integer

Number of CUs selected for a job.

manager_cu_number

No

Integer

Number of CUs on the management node selected by the user for a job, which corresponds to the number of Flink job managers. The default value is 1.

parallel_number

No

Integer

Number of parallel operations selected for a job.

log_enabled

No

Boolean

Whether to enable the job log function.

  • true: indicates to enable the job log function.
  • false: indicates to disable the job log function.
  • Default value: false

obs_bucket

No

String

OBS bucket where users are authorized to save logs when log_enabled is set to true.

smn_topic

No

String

SMN topic. If a job fails, the system will send a message to users subscribed to the SMN topic.

main_class

No

String

Job entry class.

entrypoint_args

No

String

Job entry parameter. Multiple parameters are separated by spaces.

restart_when_exception

No

Boolean

Whether to enable the function of restart upon exceptions. The default value is false.

entrypoint

No

String

Name of the package that has been uploaded to the DLI resource management system. This parameter is used to customize the JAR file where the job main class is located.

dependency_jars

No

Array of Strings

Name of the package that has been uploaded to the DLI resource management system. This parameter is used to customize other dependency packages.

Example: myGroup/test.jar,myGroup/test1.jar.

dependency_files

No

Array of Strings

Name of the resource package that has been uploaded to the DLI resource management system. This parameter is used to customize dependency files.

Example: myGroup/test.cvs,myGroup/test1.csv.

You can add the following content to the application to access the corresponding dependency file: In the command, fileName indicates the name of the file to be accessed, and ClassName indicates the name of the class that needs to access the file.

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

tm_cus

No

Integer

Number of CUs for each TaskManager. The default value is 1.

tm_slot_num

No

Integer

Number of slots in each TaskManager. The default value is (parallel_number*tm_cus)/(cu_number-manager_cu_number).

feature

No

String

Job feature. Type of the Flink image used by a job.

  • basic: indicates that the basic Flink image provided by DLI is used.
  • custom: indicates that the user-defined Flink image is used.

flink_version

No

String

Flink version. This parameter is valid only when feature is set to basic. You can use this parameter with the feature parameter to specify the version of the DLI basic Flink image used for job running.

image

No

String

Custom image. The format is Organization name/Image name:Image version.

This parameter is valid only when feature is set to custom. You can use this parameter with the feature parameter to specify a user-defined Flink image for job running. For details about how to use custom images, see Data Lake Insight User Guide.

resume_checkpoint

No

Boolean

Whether the abnormal restart is recovered from the checkpoint.

resume_max_num

No

Integer

Maximum number of retry times upon exceptions. The unit is times/hour. Value range: -1 or greater than 0. The default value is -1, indicating that the number of times is unlimited.

checkpoint_path

No

String

Storage address of the checkpoint in the JAR file of the user. The path must be unique.

tags

No

Array of Objects

Label of a Flink JAR job. For details, see Table 3.

runtime_config

No

String

Customizes optimization parameters when a Flink job is running.

Table 3 tags parameter

Parameter

Mandatory

Type

Description

key

Yes

String

Tag key.

NOTE:

A tag key can contain a maximum of 128 characters. Only letters, digits, spaces, and special characters (_.:=+-@) are allowed, but the value cannot start or end with a space or start with _sys_.

value

Yes

String

Tag key.

NOTE:

A tag value can contain a maximum of 255 characters. Only letters, digits, spaces, and special characters (_.:=+-@) are allowed. The value cannot start or end with a space.

Response

Table 4 Response parameters

Parameter

Mandatory

Type

Description

is_success

No

Boolean

Indicates whether the request is successfully executed. Value true indicates that the request is successfully executed.

message

No

String

Message content.

job

No

Object

Information about the job status. For details, see Table 5.

Table 5 job parameters

Parameter

Mandatory

Type

Description

job_id

Yes

Long

Job ID.

status_name

No

String

Name of job status.

status_desc

No

String

Status description. Causes and suggestions for the abnormal status.

Example Request

Create a Flink Jar job named test, set the job to be executed on testQueue, set the number of CUs used by the job, and enable the job log function.

{
    "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"
    ]
}

Example Response

{
  "is_success": true,
  "message": "A Flink job is created successfully.",
  "job": {
    "job_id": 138,
    "status_name": "job_init",
    "status_desc": ""
  }
}

Status Codes

Table 6 describes status codes.

Table 6 Status codes

Status Code

Description

200

The custom Flink job is created successfully.

400

The input parameter is invalid.

Error Codes

If an error occurs when this API is invoked, the system does not return the result similar to the preceding example, but returns the error code and error information. For details, see Error Codes.