(Recommended) Creating a Flink OpenSource SQL Job¶
This section describes how to create a Flink OpenSource SQL job.
DLI Flink OpenSource SQL jobs are fully compatible with the syntax of Flink 1.10 and 1.12 provided by the community. In addition, Redis, GaussDB(DWS), and DIS data source types are added based on the community connector. For details about the syntax and restrictions of Flink SQL DDL, DML, and functions, see Table API & SQL.
Prerequisites¶
You have prepared the data input and data output channels. For details, see Preparing Flink Job Data.
Before creating a Flink OpenSource SQL 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.
Creating a Flink OpenSource SQL Job¶
In the left navigation pane of the DLI management console, choose Job Management > Flink Jobs. The Flink Jobs page is displayed.
In the upper right corner of the Flink Jobs page, click Create Job.
Specify job parameters.
Table 1 Job configuration information¶ Parameter
Description
Type
Set Type to Flink OpenSource SQL. You will need to start jobs by compiling SQL statements.
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.
Template Name
You can select a sample template or a custom job template. For details about templates, see Managing Flink Templates.
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).
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.
Click OK to enter the editing page.
Edit an OpenSource SQL job.
Enter details SQL statements in the SQL statement edit area. For details about SQL statements, see the Data Lake Insight Flink OpenSource SQL Syntax Reference.
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.
Set job running parameters.
Table 2 Running parameters¶ Parameter
Description
Queue
A shared queue is selected by default. Select a CCE queue with dedicated resources and configure the following parameters:
UDF Jar: You can customize a UDF Jar file. 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.
In SQL, you can call a user-defined function that is inserted into a JAR file.
Note
During job creation, a sub-user can only select a queue that has been allocated to the user.
If the remaining capacity of the selected queue cannot meet the job requirements, the system automatically scales up the capacity and you will be billed based on the increased capacity. When a queue is idle, the system automatically scales in the queue.
CUs
Sum of the number of compute units and job manager CUs of DLI. One CU equals 1 vCPU and 4 GB.
The value is the number of CUs required for job running and cannot exceed the number of CUs in the bound queue.
Job Manager CUs
Number of CUs of the management unit.
Parallelism
Number of Flink OpenSource SQL jobs that run at the same time
Note
This value cannot be greater than four times the compute units (number of CUs minus the number of job manager CUs).
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.
OBS Bucket
OBS bucket to store job logs and checkpoint information. If the selected OBS bucket is not authorized, click Authorize.
Save Job Log
Whether to save the job running logs to OBS. The logs are saved in the following path: Bucket name/jobs/logs/Directory starting with the job ID.
Caution
CAUTION: You are advised to configure this parameter. Otherwise, no run log is generated after the job is executed. If the job fails, 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 user job logs. If the selected OBS bucket is not authorized, click Authorize.
Note
If both Enable Checkpointing and Save Job Log are selected, you only need to authorize OBS once.
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.
Enable Checkpointing
Whether to enable job snapshots. If this function is enabled, jobs can be restored based on the checkpoints.
If this option is selected, you need to set the following parameters:
Checkpoint Interval (s) indicates the interval for creating checkpoints. 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 selected OBS bucket is not authorized, click Authorize.
The checkpoint path is Bucket name/jobs/checkpoint/Directory starting with the job ID.
Note
If both Enable Checkpointing and Save Job Log are selected, you only need to authorize OBS once.
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: This parameter is available only when Enable Checkpointing is selected.
Idle State Retention Time
How long the state of a key is retained without being updated before it is removed in GroupBy or Window. 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.
(Optional) Set the runtime configuration as required.
Click Save.
Click Start. On the displayed Start Flink Jobs 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.Note
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.
Simplified Stream Graph: Display the data processing flow from the source to the sink.
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.
Simplified Stream Graph¶
On the OpenSource SQL job editing page, click Simplified Stream Graph.
Static Stream Graph¶
On the OpenSource SQL job editing page, click Static Stream Graph.
Note
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: %)