Creating a Flink OpenSource SQL Job

Prerequisites

• You have prepared the source and sink streams.
• A datasource connection has been created to enable the network between the queue where the job is about to run and external data sources.
  • For details about the external data sources that can be accessed by Flink jobs, see Common Development Methods for DLI Cross-Source Analysis.
  • For details about how to create a datasource connection, see Configuring the Network Connection Between DLI and Data Sources (Enhanced Datasource Connection).

    On the Resources > Queue Management page, locate the queue you have created, click More in the Operation column, and select Test Address Connectivity to check if the network connection between the queue and the data source is normal. For details, see Testing the Network Connectivity Between a Queue and a Data Source.

Precautions

Before creating and submitting jobs, you are advised to enable CTS to record DLI operations for queries, audits, and tracking. Using CTS to Audit DLI lists DLI operations that can be recorded by CTS.

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

Creating a Flink OpenSource SQL Job

1. In the navigation pane of the DLI management console, choose Job Management > Flink Jobs.
2. In the upper right corner of the Flink Jobs page, click Create Job.
   Figure 1 Creating a Flink OpenSource SQL job
   

3. Configure job parameters.

   Table 1 Job parameters

   Parameter	Description
   Type	Set Type to Flink OpenSource SQL. You will need to start jobs by compiling SQL statements.
   Name	Job name. The value can contain up to 57 characters. Only letters, digits, hyphens (-), and underscores (_) are allowed. The job name must be globally unique.
   Description	Job description. It can contain up to 512 characters.
   Template Name	You can select a sample template or a custom job template. For details about templates, see Managing Flink Job Templates.
   Tags	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.

4. Click OK to enter the editing page.
5. Edit an OpenSource SQL job.

   Enter detailed SQL statements in the statement editing area. For details about SQL statements, see Data Lake Insight Flink OpenSource SQL Syntax Reference.

6. Click Check Semantics.
   • You can Start a job only after the semantic verification is successful.
   • If verification is successful, the message "The SQL semantic verification is complete. No error." will be displayed.
   • If verification fails, a red "X" mark will be displayed in front of each SQL statement that produced an error. You can move the cursor to the "X" mark to view error details and change the SQL statement as prompted.
   

   Flink 1.15 does not support syntax verification.

7. Set job running parameters.

   Table 2 Running parameters

   Parameter	Description
   Queue	Resource queue used to execute Flink jobs. A queue determines the compute resources accessible to a job during its operation within an elastic resource pool. Every queue is allocated with specific resources, known as CUs, whose configuration significantly impacts the job's performance and execution efficiency. Before submitting a job, assess its resource needs and select an appropriate queue. Flink OpenSource SQL jobs support selecting For general purpose queues.
   Flink Version	Flink version used for job running. Flink versions have varying feature support. If you choose to use Flink 1.15, make sure to configure the agency information for the cloud service that DLI is allowed to access in the job. For details about the syntax of Flink 1.15, see Flink OpenSource SQL 1.15 Usage and Flink OpenSource SQL 1.15 Syntax. For details about the syntax of Flink 1.12, see Flink OpenSource SQL 1.12 Syntax. NOTE: You are advised not to use Flink of different versions for a long time. Doing so can lead to code incompatibility, which can negatively impact job execution efficiency. Doing so may result in job execution failures due to conflicts in dependencies. Jobs rely on specific versions of libraries or components.
   UDF Jar	UDF JAR file, which contains UDFs that can be called in subsequent jobs. There are the following ways to manage UDF JAR files: Upload packages to OBS: Upload JAR files to an OBS bucket in advance and select the corresponding OBS path. Upload packages to DLI: Upload JAR files to an OBS bucket in advance and create a package on the Data Management > Package Management page of the DLI management console. For details, see Creating a DLI Package. For Flink 1.15 or later, only OBS packages can be selected when creating jobs, and DLI packages are not supported.
   Agency	If you choose Flink 1.15 or later to execute your job, you can create a custom agency to allow DLI to access other services. For details about how to create a custom agency, see Creating a Custom DLI Agency.
   Resource Configuration Version	DLI offers various resource configuration templates based on different Flink engine versions. Compared with the v1 template, the v2 template does not support the setting of the number of CUs. The v2 template supports the setting of Job Manager Memory and Task Manager Memory. V1: applicable to Flink 1.12 and 1.15. V2: applicable to Flink 1.15 and 1.17. You are advised to use the parameter settings of v2. For details about the parameters of v1, see Table 3. For details about the parameters of v2, see Table 4.

   Table 3 Resource specification parameters of v1

   Parameter	Description
   CUs	Sum of the number of compute units and JobManager CUs of DLI. CU is also the billing unit of DLI. One CU equals one vCPU and 4 GB of memory. The value is the number of CUs required for job running and cannot exceed the number of CUs in the bound queue. NOTE: When Task Manager Config is selected, elastic resource pool queue management is optimized by automatically adjusting CUs to match Actual CUs after setting Slot(s) per TM. CUs = Actual number of CUs = max[Job Manager CPU + Task Manager CPU, (Job Manager Memory + Task Manager Memory/4)] Job Manager CPU + Task Manager CPU = Actual TMs x CU(s) per TM + Job Manager CUs. Job Manager Memory + Task Manager Memory = Actual TMs x Memory per TM + Job Manager Memory If Slot(s) per TM is set, then: Actual TMs = Parallelism/Slot(s) per TM. If Slot(s) per TM is not set, then: Actual TMs = (CUs – Job Manager CUs)/CU(s) per TM. If Memory per TM and Job Manager Memory in the optimization parameters are not set, then: Memory per TM = CU(s) per TM x 4. Job Manager Memory = Job Manager CUs x 4. The parallelism degree of Spark resources is jointly determined by the number of Executors and the number of Executor CPU cores.
   Job Manager CUs	Number of JobManager CUs. If the current job is running on a basic edition elastic resource pool (16–64 CUs), it is recommended that the JobManager's CU count does not exceed 2 to avoid resource scheduling failures during job execution.
   Parallelism	Number of tasks concurrently executed by each operator in a job. NOTE: This value cannot be greater than four times the compute units (CUs – Job Manager CUs).
   Task Manager Config	Whether TaskManager resource parameters are set If selected, you need to configure the following parameters: CU(s) per TM: Number of resources occupied by each TaskManager. If the current job is running on a basic edition elastic resource pool (16–64 CUs), it is recommended that a single TaskManager's CU count does not exceed 2 to avoid resource scheduling failures during job execution. Slot(s) per TM: Number of slots contained in each TaskManager. If not selected, the system automatically uses the default values. CU(s) per TM: The default value is 1. Slot(s) per TM: The default value is (Parallelism x CU(s) per TM)/(CUs – Job Manager CUs).
   OBS Bucket	OBS bucket to store job logs and checkpoint information. If the OBS bucket you selected is unauthorized, click Authorize.
   Save Job Log	Whether job running logs are saved to OBS. The logs are saved in the following path: Bucket name/jobs/logs/Directory starting with the job ID. CAUTION: You are advised to select this parameter. Otherwise, no run logs will be generated after the job is executed. If the job runs abnormally later, you will be unable to obtain the run logs for troubleshooting. If selected, you need to configure the following parameters: OBS Bucket: Select an OBS bucket to store job logs. If the OBS bucket you selected is unauthorized, click Authorize. NOTE: If Enable Checkpointing and Save Job Log are both selected, you only need to authorize OBS once.
   Root Log Level	This parameter is supported in Flink 1.15 or later. The log level determines the importance and urgency of log messages. The root log level serves as a global setting, providing a default log level for the entire application. If no specific log level is configured for a particular logger or class, the root log level will be applied. For example, if the root log level is set to INFO, all loggers without individual configurations will output logs at the INFO level or higher. Log levels are prioritized from low to high as follows: TRACE < DEBUG < INFO < WARN < ERROR < FATAL. In practice, log levels can be used to filter and display log information effectively. For example: During development, TRACE and DEBUG levels may be enabled for detailed insights. In production environments, only INFO, WARN, and ERROR levels are typically enabled to reduce log file size and improve performance. For details on supported root log level types and descriptions, see Table 5.
   Class Log Level	This parameter is supported in Flink 1.15 or later. A class log level refers to a log level specifically assigned to a particular class or package. It enables finer-grained control over logging for different classes or packages. By configuring a class log level, you can obtain more detailed logs for specific classes or packages without affecting other parts of the system. For example, during debugging, you might set the log level of a critical class to DEBUG to gather additional debugging information.
   Alarm on Job Exception	Whether to notify users of any job exceptions, such as running exceptions or arrears, via SMS or email. If selected, you need to configure the following parameters: SMN Topic Select a custom SMN topic. For details about how to create a custom SMN topic, see Creating a Topic.
   Enable Checkpointing	Whether to enable job snapshots. If this function is enabled, jobs can be restored based on the checkpoints. If selected, you need to configure the following parameters: Checkpoint Interval: interval for creating checkpoints, in seconds. The value ranges from 1 to 999999, and the default value is 30. Checkpoint Mode can be set to either of the following values: At least once: Events are processed at least once. Exactly once: Events are processed only once. OBS Bucket: Select an OBS bucket to store your checkpoints. If the OBS bucket you selected is unauthorized, click Authorize. The checkpoint path is Bucket name/jobs/checkpoint/Directory starting with the job ID. NOTE: If Enable Checkpointing and Save Job Log are both selected, you only need to authorize OBS once.
   Auto Restart upon Exception	Whether automatic restart is enabled. If enabled, jobs will be automatically restarted and restored when exceptions occur. If selected, you need to configure the following parameters: Max. Retry Attempts: maximum number of retries 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: This parameter is available only when Enable Checkpointing is selected.
   Idle State Retention Time	Clears intermediate states of operators such as GroupBy, RegularJoin, Rank, and Depulicate that have not been updated after the maximum retention time. The default value is 1 hour.
   Dirty Data Policy	Policy for processing dirty data. The following policies are supported: Ignore, Trigger a job exception, and Save. If you set this field to Save, the Dirty Data Dump Address must be set. Click the address box to select the OBS path for storing dirty data. This parameter is available only when a DIS data source is used.

   Table 4 Resource specification parameters of v2

   Parameter	Description
   Parallelism	Number of tasks concurrently executed by each operator in a job. NOTE: The minimum parallelism must not be less than 1. The default value is 1. This value cannot be greater than four times the compute units (CUs – Job Manager CUs).
   Job Manager CPU	Number of vCPUs available for JobManager. The default value is 1. The minimum value cannot be less than 0.5. If the current job is running on a basic edition elastic resource pool (16–64 CUs), it is recommended that the JobManager's CPU value does not exceed 2 to avoid resource scheduling failures during job execution.
   Job Manager Memory	Memory available for JobManager. The default value is 4 GB. The minimum size cannot be less than 2 GB (2,048 MB). The default unit is GB, which can be set to GB or MB.
   Task Manager CPU	Number of vCPUs available for TaskManager. The default value is 1. The minimum value cannot be less than 0.5. If the current job is running on a basic edition elastic resource pool (16–64 CUs), it is recommended that the TaskManager's CPU value does not exceed 2 to avoid resource scheduling failures during job execution.
   Task Manager Memory	Memory available for TaskManager. The default value is 4 GB. The minimum size cannot be less than 2 GB (2,048 MB). The default unit is GB, which can be set to GB or MB.
   Slot(s) per TM	Number of parallel tasks that a single TaskManager can support. Each task slot can execute one task in parallel. Increasing task slots enhances the parallel processing capacity of TaskManagers but also increases resource consumption. The number of task slots is linked to the CPU count of TaskManagers since each CPU can offer one task slot. By default, a single TM slot is set to 1. The minimum parallelism must not be less than 1.
   OBS Bucket	OBS bucket to store job logs and checkpoint information. If the OBS bucket you selected is unauthorized, click Authorize.
   Save Job Log	Whether job running logs are saved to OBS. The logs are saved in the following path: Bucket name/jobs/logs/Directory starting with the job ID. CAUTION: You are advised to select this parameter. Otherwise, no run logs will be generated after the job is executed. If the job runs abnormally later, you will be unable to obtain the run logs for troubleshooting. If selected, you need to configure the following parameters: OBS Bucket: Select an OBS bucket to store job logs. If the OBS bucket you selected is unauthorized, click Authorize. NOTE: If Enable Checkpointing and Save Job Log are both selected, you only need to authorize OBS once.
   Root Log Level	This parameter is supported in Flink 1.15 or later. The log level determines the importance and urgency of log messages. The root log level serves as a global setting, providing a default log level for the entire application. If no specific log level is configured for a particular logger or class, the root log level will be applied. For example, if the root log level is set to INFO, all loggers without individual configurations will output logs at the INFO level or higher. Log levels are prioritized from low to high as follows: TRACE < DEBUG < INFO < WARN < ERROR < FATAL. In practice, log levels can be used to filter and display log information effectively. For example: During development, TRACE and DEBUG levels may be enabled for detailed insights. In production environments, only INFO, WARN, and ERROR levels are typically enabled to reduce log file size and improve performance. For details on supported root log level types and descriptions, see Table 5.
   Class Log Level	This parameter is supported in Flink 1.15 or later. A class log level refers to a log level specifically assigned to a particular class or package. It enables finer-grained control over logging for different classes or packages. By configuring a class log level, you can obtain more detailed logs for specific classes or packages without affecting other parts of the system. For example, during debugging, you might set the log level of a critical class to DEBUG to gather additional debugging information.
   Alarm on Job Exception	Whether to notify users of any job exceptions, such as running exceptions or arrears, via SMS or email. If selected, you need to configure the following parameters: SMN Topic Select a custom SMN topic. For details about how to create a custom SMN topic, see Creating a Topic.
   Enable Checkpointing	Whether to enable job snapshots. If this function is enabled, jobs can be restored based on the checkpoints. If selected, you need to configure the following parameters: Checkpoint Interval: interval for creating checkpoints, in seconds. The value ranges from 1 to 999999, and the default value is 30. Checkpoint Mode can be set to either of the following values: At least once: Events are processed at least once. Exactly once: Events are processed only once. OBS Bucket: Select an OBS bucket to store your checkpoints. If the OBS bucket you selected is unauthorized, click Authorize. The checkpoint path is Bucket name/jobs/checkpoint/Directory starting with the job ID. NOTE: If Enable Checkpointing and Save Job Log are both selected, you only need to authorize OBS once.
   Auto Restart upon Exception	Whether automatic restart is enabled. If enabled, jobs will be automatically restarted and restored when exceptions occur. If selected, you need to configure the following parameters: Max. Retry Attempts: maximum number of retries 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: This parameter is available only when Enable Checkpointing is selected.
   Idle State Retention Time	Clears intermediate states of operators such as GroupBy, RegularJoin, Rank, and Depulicate that have not been updated after the maximum retention time. The default value is 1 hour.
   Dirty Data Policy	Policy for processing dirty data. The following policies are supported: Ignore, Trigger a job exception, and Save. If you set this field to Save, the Dirty Data Dump Address must be set. Click the address box to select the OBS path for storing dirty data. This parameter is available only when a DIS data source is used.

   Table 5 Root log levels

   Level	Description	Use Case
   TRACE	The most granular level of logging, typically used during development and debugging. It captures all operational details, including variable values and function calls.	Primarily for the development phase to help developers understand code execution flow and state.
   DEBUG	Less detailed than TRACE, intended for debugging purposes. It logs program runtime states and key variable values but omits exhaustive detail.	Mainly used during development and testing phases to assist developers in identifying issues.
   INFO	Records important information during normal operations, useful for system operators and maintainers without impacting system functionality.	Tracks major system operations and statuses, such as startup, shutdown, and configuration changes.
   WARN	Potential issues or anomalies that do not disrupt system operation but warrant attention from developers or operators.	Logs situations that may affect system performance or functionality, serving as a warning for possible problems.
   ERROR	Highlights critical issues or exceptions that impair system functionality and require immediate resolution.	Captures errors and exceptions during system operation, enabling developers to quickly identify and address them.

8. (Optional) Set the runtime configuration as required. For details about related parameters, see How Do I Optimize the Performance of a Flink Job?
   Figure 2 Runtime configuration
   

   You can set compute resource specification parameters on the Runtime Configuration tab of Flink jobs, and the parameter values have a higher priority than the specified values.

   Table 6 describes the parameter mapping.

   

   In Flink 1.12, you are advised to set compute resource specification parameters based on the configuration method on the console. Using custom parameter settings may result in discrepancies in actual CU statistics.

   Table 6 Mapping between compute resource specification parameters on the console and those on the Runtime Configuration tab

   Runtime Configuration	Compute Resource Specification Parameter of v1	Compute Resource Specification Parameter of v2	Description
   kubernetes.jobmanager.cpu	Job Manager CUs	Job Manager CPU	Number of vCPUs available for JobManager. The default value is 1. The minimum value cannot be less than 0.5.
   kubernetes.taskmanager.cpu	CU(s) per TM	Task Manager CPU	Number of vCPUs available for TaskManager. The default value is 1. The minimum value cannot be less than 0.5.
   jobmanager.memory.process.size	-	Job Manager Memory	Memory available for JobManager. The default value is 4 GB. The minimum size cannot be less than 2 GB (2,048 MB). The default unit is GB, which can be set to GB or MB.
   taskmanager.memory.process.size	-	Task Manager Memory	Memory available for TaskManager. The default value is 4 GB. The minimum size cannot be less than 2 GB (2,048 MB). The default unit is GB, which can be set to GB or MB.

9. Click Save.
10. Click Start. 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. Once a job is successfully submitted, its status changes from Submitting to Running. After the execution is complete, the status changes to Completed.

    If the job status is Submission failed or Running exception, the job fails to submit or run. In this case, you can hover over the status icon in the Status column of the job list to view the error details. You can click to copy these details. Rectify the fault based on the error information and resubmit the job.

    

    Other buttons are as follows:

    • Save As: Save the created job as a new job.
    • Static Stream Graph: Provide the static concurrency estimation function and stream graph display function. See Figure 4.
    • Simplified Stream Graph: Display the data processing flow from the source to the sink. See Figure 3.
    • Format: Format the SQL statements in the editing box.
    • Set as Template: Set the created SQL statements as a job template.
    • Theme Settings: Set the theme related parameters, including Font Size, Wrap, and Page Style.
    • Help: Redirect to the help center to provide you with the SQL syntax for stream jobs.

Simplified Stream Graph

On the OpenSource SQL job editing page, click Simplified Stream Graph.

Simplified stream graph viewing is only supported in Flink 1.12 and Flink 1.10.

Figure 3 Simplified stream graph

Static Stream Graph

On the OpenSource SQL job editing page, click Static Stream Graph.

• Simplified stream graph viewing is only supported in Flink 1.12 and Flink 1.10.
• If you use a UDF in a Flink OpenSource SQL job, it is not possible to generate a static stream graph.

The Static Stream Graph page also allows you to:

• Estimate concurrencies. Click Estimate Concurrencies on the Static Stream Graph page to estimate concurrencies. Click Restore Initial Value to restore the initial value after concurrency estimation.
• Zoom in or out the page.
• Expand or merge operator chains.
• You can edit Parallelism, Output rate, and Rate factor.
  • Parallelism: indicates the number of concurrent tasks.
  • Output rate: indicates the data traffic of an operator. The unit is piece/s.
  • Rate factor: indicates the retention rate after data is processed by operators. Rate factor = Data output volume of an operator/Data input volume of the operator (Unit: %)

Figure 4 Static stream graph