Updated on 2022-07-04 GMT+08:00

Creating a Custom Flink Job

This section describes how to create a custom Flink job. You can perform secondary development based on Flink APIs, build your own JAR file, and submit the JAR file to DLI queues. DLI is fully compatible with open-source community APIs. To create a custom Flink job, you need to compile and build application JAR files. You must have a certain understanding of Flink secondary development and have high requirements related to stream computing complexity.

Prerequisites

  • Ensure that a dedicated queue has been created.
  • When running a custom Flink job, you need to build the secondary development application code into a JAR file and upload the JAR file to the created OBS bucket. Choose Data Management > Package Management to create a package. For details, see Creating a Package.

    DLI does not support the download function. If you need to modify the uploaded data file, please edit the local file and upload it again.

  • Flink dependencies have been built in the DLI server and security hardening has been performed based on the open-source community version. To avoid dependency package compatibility issues or log output and dump issues, be careful to exclude the following files when packaging:
    • Built-in dependencies (or set the package dependency scope to "provided")
    • Log configuration files (example, log4j.properties/logback.xml)
    • JAR package for log output implementation (example, log4j).

Creating a Custom Flink Job

  1. In the left navigation pane of the DLI management console, choose Job Management > Flink Jobs. The Flink Jobs page is displayed.
  2. In the upper right corner of the Flink Jobs page, click Create Job.
  3. Configure job information.

    Table 1 Job configuration information

    Parameter

    Description

    Type

    Select Flink Jar.

    Name

    Name of a job. Enter 1 to 57 characters. Only letters, digits, hyphens (-), and underscores (_) are allowed.

    NOTE:

    The job name must be globally unique.

    Description

    Description of a job. It can be up to 512 characters long.

  4. Click OK to enter the Edit page.
  5. Upload the JAR file.

    Table 2 Description

    Name

    Description

    Application

    User-defined package. Before selecting a JAR file to be inserted, upload the corresponding JAR file to the OBS bucket and choose Data Management > Package Management to create a package. For details, see Creating a Package.

    Main Class

    The name of the JAR package to be loaded, for example, KafkaMessageStreaming.

    • Default: Specified based on the Manifest file in the JAR package.
    • Manually assign: Enter the class name and confirm the class arguments (separate arguments with spaces).
    NOTE:

    When a class belongs to a package, the package path must be carried, for example, packagePath.KafkaMessageStreaming.

    Class Arguments

    List of arguments of a specified class. The arguments are separated by spaces.

    JAR Package Dependencies

    Select a user-defined package dependency. Before selecting a JAR file to be inserted, upload the corresponding JAR file to the OBS bucket and choose Data Management > Package Management to create a package. Select JAR as the package type. For details, see Creating a Package.

    Other Dependencies

    User-defined dependency files. Before selecting a dependency file, upload the file to the OBS bucket and choose Data Management > Package Management to create a package. The package type is not limited. For details, see Creating a Package.

    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")

    Image

    Image set on the Software Repository for Container (SWR) console.

  6. Configure job parameters.

    Table 3 Description

    Name

    Description

    CUs

    One CU has one vCPU and 4-GB memory. The number of CUs ranges from 2 to 400.

    Job Manager CUs

    Set the number of CUs on a management unit. The value ranges from 1 to 4. The default value is 1.

    Max Concurrent Jobs

    Maximum number of parallel operators in a job.

    NOTE:
    • The value must be less than or equal to four times the number of compute units (CUs minus the number of job manager CUs).
    • You are advised to set this parameter to a value greater than that configured in the code. Otherwise, job submission may fail.

    Task Manager Configuration

    Whether to set Task Manager resource parameters

    If this option is selected, you need to set the following parameters:

    • CU(s) per TM: Number of resources occupied by each Task Manager.
    • Slot(s) per TM: Number of slots contained in each Task Manager.

    Save Job Log

    Whether to save the job running logs to OBS

    If this option is selected, you need to set the following parameters:

    OBS Bucket: Select an OBS bucket to store user job logs. If the selected OBS bucket is not authorized, click Authorize.

    Alarm Generation upon Job Exception

    Whether to report job exceptions, for example, abnormal job running or exceptions due to an insufficient balance, to users via SMS or email

    If this option is selected, you need to set the following parameters:

    SMN Topic

    Select a user-defined SMN topic. For details about how to customize SMN topics, see Creating a Topic in the Simple Message Notification User Guide.

    Auto Restart upon Exception

    Whether to enable automatic restart. If this function is enabled, any job that has become abnormal will be automatically restarted.

  7. Select a queue to which the job belongs.

    • A custom Flink job can run only on a created dedicated queue.
    • If no dedicated queue is available in the Queue drop-down list, create a dedicated queue and bind it to the current user.

  8. Click Save on the upper right of the page.
  9. Click Start on the upper right side of the page. On the displayed Start Flink Job page, confirm the job specifications, and click Start Now to start the job.

    After the job is started, the system automatically switches to the Flink Jobs page, and the created job is displayed in the job list. You can view the job status in the Status column. After a job is successfully submitted, the job status will change from Submitting to Running. After the execution is complete, the message Completed is displayed.

    If the job status is Submission failed or Running exception, the job submission failed or the job did not execute successfully. In this case, you can move the cursor over the status icon in the Status column of the job list to view the error details. You can click to copy these details. After handling the fault based on the provided information, resubmit the job.

    Other buttons are as follows:

    Save As: Save the created job as a new job.