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 a project ID, see Obtaining a Project ID.

Request Parameters

**Table 2** Parameter descriptions
Parameter	Mandatory	Type	Description
name	Yes	String	Job name. The value can contain up to 57 characters.
desc	No	String	Job description. Length range: 0 to 512 characters.
queue_name	No	String	Queue name. 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.
checkpoint_enabled	No	Boolean	Whether to enable the automatic job snapshot function. Options: true: Enable this function. false: Do not enable this function. Default value: false.
checkpoint_mode	No	Integer	Snapshot mode. Options: 1: ExactlyOnce, meaning data is consumed only once. 2: AtLeastOnce, meaning data is consumed at least once. Default value: 1.
checkpoint_interval	No	Integer	Snapshot interval. The unit is second. The default value is 10.
log_enabled	No	Boolean	Whether to enable the job log function. Options: true: Enable this function. false: Do not enable this 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 this 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 uploaded to OBS. You can customize the JAR file where the main job class is. For Flink 1.15 or later, you can only select packages from OBS, instead of DLI. Example: obs://bucket_name/test.jar
dependency_jars	No	Array of strings	Name of the package uploaded to OBS. You can customize other dependency packages of the job. For Flink 1.15 or later, you can only select packages from OBS, instead of DLI. Example: obs://bucket_name/test1.jar, obs://bucket_name/test2.jar
dependency_files	No	Array of strings	Name of the resource package uploaded to OBS. You can customize the dependency files of the job. For Flink 1.15 or later, you can only select packages from OBS, instead of DLI. Example: [obs://bucket_name/file1, obs://bucket_name/file2] 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_numbertm_cus)/(cu_number-manager_cu_number)**.
execution_agency_urn	No	String	Name of the agency authorized to DLI. This parameter is configurable in Flink 1.15.
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 parameters
Parameter	Mandatory	Type	Description
key	Yes	String	Tag key. NOTE: A tag key can contain a maximum of 36 characters. Special characters (=*<>\\|) are not allowed, and the key cannot start with a space.
value	Yes	String	Tag value. NOTE: A tag value can contain a maximum of 43 characters. Special characters (=*<>\\|) are not allowed, and the value cannot start with a space.

Response Parameters

**Table 4** Response parameters
Parameter	Mandatory	Type	Description
is_success	No	String	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"
    ],
    "execution_agency_urn": "myAgencyName",
    "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 called, the system does not return the result similar to the preceding example, but returns an error code and error message. For details, see Error Codes.

Parent Topic: Flink Job-related APIs

Previous topic: Updating a SQL Job

Next topic: Updating a Flink Jar Job