Creating a Flink Jar Job

Prerequisites

• Ensure that a dedicated queue has been created. To create a dedicated queue, select Dedicated Resource Mode when you choose the type of a queue during purchase.
• When you use a Flink Jar job to access other external data sources, such as OpenTSDB, HBase, Kafka, GaussDB(DWS), RDS, CSS, CloudTable, DCS Redis, and DDS, you need to create a datasource connection to connect the job running queue to the external data source.
  • For details about the external data sources that can be accessed by Flink jobs, see Cross-Source Analysis Development Methods.
  • For details about how to create a datasource connection, see Enhanced Datasource Connections.

    On the Resources > Queue Management page, locate the queue you have created, and choose More > Test Address Connectivity in the Operation column to check whether the network connection between the queue and the data source is normal. For details, see Testing Address Connectivity.

• When running a Flink Jar job, you need to build the secondary development application code into a Jar package and upload the JAR package 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 in Maven or SBT)
  • Log configuration files (example, log4j.properties/logback.xml)
  • JAR package for log output implementation (example, log4j).

Precautions

Before creating jobs and submitting tasks, you are advised to enable CTS to record operations associated with DLI for later query, audit, and backtrack operations. To view the DLI operations that can be recorded by CTS, see DLI Operations That Can Be Recorded by CTS.

For details about how to enable CTS and view trace details, see the Cloud Trace Service Getting Started.

Creating a Flink Jar 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.
   Figure 1 Creating a Flink Jar job
   

3. Specify job parameters.

   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.
   Tag	Tags used to identify cloud resources. A tag includes the tag key and tag value. If you want to use the same tag to identify multiple cloud resources, that is, to select the same tag from the drop-down list box for all services, you are advised to create predefined tags on the Tag Management Service (TMS). If your organization has configured tag policies for DLI, add tags to resources based on the policies. If a tag does not comply with the tag policies, resource creation may fail. Contact your organization administrator to learn more about tag policies. For details, see Tag Management Service User Guide. NOTE: A maximum of 20 tags can be added. Only one tag value can be added to a tag key. The key name in each resource must be unique. Tag key: Enter a tag key name in the text box. 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_. Tag value: Enter a tag value in the text box. 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.

4. Click OK to enter the editing page.
5. Select a queue. Flink Jar jobs can run only on dedicated queues.
   

   • A Flink Jar job can run only on a pre-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.

6. Configuring Flink Jar Job parameters
   Figure 2 Configuring Flink Jar Job parameters
   

   Table 2 Parameter description

   Name	Description
   Queue	A shared queue is selected by default. You can select a custom queue as needed.
   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. For details about built-in dependency packages, see Built-in Dependencies.
   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: You must enter the class name and confirm the class arguments (separate arguments with spaces). NOTE: When a class belongs to a package, the main class path must contain the complete package path, for example, packagePath.KafkaMessageStreaming.
   Class Arguments	List of arguments of a specified class. The arguments are separated by spaces. Flink parameters support replacement of non-sensitive global variables. For example, if you add the global variable windowsize in Global Configuration > Global Variables, you can add the -windowsSize {{windowsize}} parameter for the Flink Jar job.
   JAR Package Dependencies	Select a user-defined package dependency. The dependent program packages are stored in the classpath directory of the cluster. 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. For details about built-in dependency packages, see Built-in Dependencies.
   Other Dependencies	User-defined dependency files. Other dependency files need to be referenced in the code. 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 command 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")
   Job Type	This parameter is displayed when the queue type is CCE. Basic Image: Select the image name and image version. Image set on the Software Repository for Container (SWR) console.
   Flink Version	Before selecting a Flink version, you need to select the queue to which the Flink version belongs. Currently, versions 1.10, 1.11, and 1.12 are supported.
   Runtime Configuration	User-defined optimization parameters. The parameter format is key=value. Flink optimization parameters support replacement non-sensitive global variable. For example, if you create global variable phase in Global Configuration > Global Variables, optimization parameter table.optimizer.agg-phase.strategy={{phase}} can be added to the Flink Jar job.

7. Configure job parameters.
   Figure 3 Configuring job parameters
   

   Table 3 Parameter description

   Name	Description
   CUs	One CU has one vCPU and 4-GB memory. The number of CUs ranges from 2 to 10,000.
   Job Manager CUs	Set the number of CUs on a management unit. The value ranges from 1 to 4. The default value is 1.
   Parallelism	Maximum number of parallel operators in a job. The value ranges from 1 to 10,000. 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 the OBS bucket. CAUTION: You are advised to select this parameter. Otherwise, no run log is generated after the job is executed. If the job is abnormal, the run log cannot be obtained for fault locating. If this option is selected, you need to set the following parameters: OBS Bucket: Select an OBS bucket to store 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. If this option is selected, you need to set the following parameters: Max. Retry Attempts: maximum number of retry times upon an exception. The unit is Times/hour. Unlimited: The number of retries is unlimited. Limited: The number of retries is user-defined. Restore Job from Checkpoint: Restore the job from the saved checkpoint. If you select this parameter, you also need to set Checkpoint Path. Checkpoint Path: Select the checkpoint saving path. The checkpoint path must be the same as that you set in the application package. Note that the checkpoint path for each job must be unique. Otherwise, the checkpoint cannot be obtained.

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 the price, 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 available buttons are as follows:

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