On this page

Show all

Help Center/ MapReduce Service/ User Guide (Ankara Region)/ Managing Clusters/ Job Management/ Running a Flink Job

Running a Flink Job

Updated on 2024-11-29 GMT+08:00

View PDF

You can submit programs developed by yourself to MRS to execute them, and obtain the results. This section describes how to submit a Flink job on the MRS management console. Flink jobs are used to submit JAR programs to process streaming data.

Prerequisites

You have uploaded the program packages and data files required for running jobs to OBS or HDFS.

Submitting a Job on the GUI

Log in to the MRS console.
Choose Clusters > Active Clusters, select a running cluster, and click its name to switch to the cluster details page.
If Kerberos authentication is enabled for the cluster, perform the following steps. If Kerberos authentication is not enabled for the cluster, skip this step.

In the Basic Information area on the Dashboard page, click Synchronize on the right side of IAM User Sync to synchronize IAM users.
NOTE:
- When the policy of the user group to which the IAM user belongs changes from MRS ReadOnlyAccess to MRS CommonOperations, MRS FullAccess, or MRS Administrator, wait for 5 minutes until the new policy takes effect after the synchronization is complete because the SSSD (System Security Services Daemon) cache of cluster nodes needs time to be updated. Then, submit a job. Otherwise, the job may fail to be submitted.
- When the policy of the user group to which the IAM user belongs changes from MRS CommonOperations, MRS FullAccess, or MRS Administrator to MRS ReadOnlyAccess, wait for 5 minutes until the new policy takes effect after the synchronization is complete because the SSSD cache of cluster nodes needs time to be updated.
Click the Jobs tab.
Click Create. The Create Job page is displayed.

Set Type to Flink. Configure Flink job information by referring to Table 1.

**Table 1** Job configuration information
Parameter	Description
Name	Job name. It contains 1 to 64 characters. Only letters, digits, hyphens (-), and underscores (_) are allowed. NOTE: You are advised to set different names for different jobs.
Program Path	Path of the program package to be executed. The following requirements must be met: Contains a maximum of 1,023 characters, excluding special characters such as ;\|&><'$. The parameter value cannot be empty or full of spaces. The path of the program to be executed can be stored in HDFS or OBS. The path varies depending on the file system. OBS: The path must start with obs://. Example: *obs://wordcount/program/xxx.jar* HDFS: The path must start with /user. For details about how to import data to HDFS, see Importing Data.
Program Parameter	(Optional) Used to configure optimization parameters such as threads, memory, and vCPUs for the job to optimize resource usage and improve job execution performance. Table 2 describes the common parameters of a running program.
Parameters	(Optional) Key parameter for program execution. The parameter is specified by the function of the user's program. MRS is only responsible for loading the parameter. Multiple parameters are separated by space. The parameter contains a maximum of 150,000 characters. It cannot contain special characters ;\|&><'$, but can be left blank. CAUTION: If you enter a parameter with sensitive information (such as the login password), the parameter may be exposed in the job details display and log printing. Exercise caution when performing this operation.
Service Parameter	(Optional) It is used to modify service parameters for the job. The parameter modification applies only to the current job. To make the modification take effect permanently for the cluster, follow instructions in Configuring Service Parameters. To add multiple parameters, click on the right. To delete a parameter, click Delete on the right. Table 3 describes the common parameters of a service.
Command Reference	Command submitted to the background for execution when a job is submitted.

**Table 2** Program parameters
Parameter	Description	Example Value
-ytm	Memory size of each TaskManager container. (Optional unit. The unit is MB by default.)	1024
-yjm	Memory size of JobManager container. (Optional unit. The unit is MB by default.)	1024
-yn	Number of Yarn containers allocated to applications. The value is the same as the number of TaskManagers.	2
-ys	Number of TaskManager cores.	2
-ynm	Custom name of an application on Yarn.	test
-c	Class of the program entry point (for example, the main or getPlan() method). This parameter is required only when the JAR file does not specify the class of its manifest.	com.bigdata.mrs.test

NOTE:

The -yn parameter is not supported.

**Table 3** Service parameters
Parameter	Description	Example Value
fs.obs.access.key	Key ID for accessing OBS.	-
fs.obs.secret.key	Key corresponding to the key ID for accessing OBS.	-

Confirm job configuration information and click OK.

After the job is created, you can manage it.

Submitting a Job in the Background

The default client installation path is /opt/Bigdata/client. Configure the path based on site requirements.

Log in to the MRS client.
Run the following command to initialize environment variables:

source /opt/Bigdata/client/bigdata_env

If Kerberos authentication is enabled for the cluster, perform the following steps. If Kerberos authentication is not enabled for the cluster, skip this step.

Prepare a user for submitting Flink jobs.
Log in to Manager as the newly created user.
Log in to Manager of the cluster. Choose System > Manage User. In the Operation column of the row that contains the added user, choose More > Download authentication credential to locate the row that contains the user.
Decompress the downloaded authentication credential package and copy the obtained file to a directory on the client node, for example, /opt/Bigdata/client/Flink/flink/conf. If the client is installed on a node outside the cluster, copy the obtained file to the /etc/ directory on this node.
In security mode, add the service IP address of the node where the client is installed and floating IP address of Manager to the jobmanager.web.allow-access-address configuration item in the /opt/Bigdata/client/Flink/flink/conf/flink-conf.yaml file.

Run the following commands to configure security authentication by adding the keytab path and username to the /opt/Bigdata/client/Flink/flink/conf/flink-conf.yaml configuration file.

security.kerberos.login.keytab: <user.keytab file path>
security.kerberos.login.principal: <Username>

Example:

security.kerberos.login.keytab: /opt/Bigdata/client/Flink/flink/conf/user.keytab
security.kerberos.login.principal: test

In the bin directory of the Flink client, run the following command to perform security hardening. Then, set a password for submitting jobs.

sh generate_keystore.sh

This script automatically replaces the SSL value in the /opt/Bigdata/client/Flink/flink/conf/flink-conf.yaml file.

**Table 4** Parameter description
Parameter	Example Value	Description
security.ssl.rest.enabled	true	Switch to enable external SSL.
security.ssl.rest.keystore	${path}/flink.keystore	Path for storing keystore.
security.ssl.rest.keystore-password	123456	Password of the keystore. 123456 indicates a user-defined password is required.
security.ssl.rest.key-password	123456	Password of the SSL key. 123456 indicates a user-defined password is required.
security.ssl.rest.truststore	${path}/flink.truststore	Path for storing the truststore.
security.ssl.rest.truststore-password	123456	Password of the truststore. 123456 indicates a user-defined password is required.

NOTE:

The generated flink.keystore, flink.truststore, and security.cookie items are automatically filled in the corresponding configuration items in flink-conf.yaml.
You can obtain the values of security.ssl.key-password, security.ssl.keystore-password, and security.ssl.truststore-password using the Manager plaintext encryption API by running the following command:
curl -k -i -u <user name>:<password> -X POST -HContent-type:application/json -d '{"plainText":"<password>"}' 'https://x.x.x.x:28443/web/api/v2/tools/encrypt'; In the preceding command, <password> must be the same as the password used for issuing the certificate, and x.x.x.x indicates the floating IP address of Manager in the cluster.

Commands carrying authentication passwords pose security risks. Disable historical command recording before running such commands to prevent information leakage.

Configure paths for the client to access the flink.keystore and flink.truststore files.
- Absolute path: After the script is executed, the file path of flink.keystore and flink.truststore is automatically set to the absolute path opt/Bigdata/client/Flink/flink/conf/ in the flink-conf.yaml file. In this case, you need to move the flink.keystore and flink.truststore files from the conf directory to this absolute path on the Flink client and Yarn nodes.
- Relative path: Perform the following steps to set the file path of flink.keystore and flink.truststore to the relative path and ensure that the directory where the Flink client command is executed can directly access the relative paths.
  1. In the /opt/Bigdata/client/Flink/flink/conf/directory, create a new directory, for example, ssl.
  2. Move the flink.keystore and flink.truststore file to the /opt/Bigdata/client/Flink/flink/conf/ssl/ directory.
  3. Change the values of the following parameters in the flink-conf.yaml file to relative paths:
```
security.ssl.keystore: ssl/flink.keystore
security.ssl.truststore: ssl/flink.truststore
```
If the client is installed on a node outside the cluster, add the following configuration to the configuration file (for example, /opt/Bigdata/client/Flink/fink/conf/flink-conf.yaml). Replace xx.xx.xxx.xxx with the IP address of the node where the client resides.
```
web.access-control-allow-origin: xx.xx.xxx.xxx
jobmanager.web.allow-access-address: xx.xx.xxx.xxx
```

Run a wordcount job.
- Normal cluster (Kerberos authentication disabled)
  - Run the following commands to start a session and submit a job in the session:
```
yarn-session.sh -nm "session-name" -d
flink run /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
```
  - Run the following command to submit a single job on Yarn:
```
flink run -m yarn-cluster /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
```
- Security cluster (Kerberos authentication enabled)
  - If the flink.keystore and flink.truststore file are stored in the absolute path:
    - Run the following commands to start a session and submit a job in the session:
      yarn-session.sh -nm "session-name" -d flink run /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
    - Run the following command to submit a single job on Yarn:
      flink run -m yarn-cluster /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
  - If the flink.keystore and flink.truststore file are stored in the relative path:
    - In the same directory of SSL, run the following command to start a session and submit jobs in the session. The SSL directory is a relative path. For example, if the SSL directory is opt/Bigdata/client/Flink/flink/conf/, then run the following command in this directory:
      yarn-session.sh -t ssl/ -nm "session-name" -d flink run /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar
    - Run the following command to submit a single job on Yarn:
      flink run -m yarn-cluster -yt ssl/ /opt/Bigdata/client/Flink/flink/examples/streaming/WordCount.jar

Parent topic: Job Management

Previous topic: Running a SparkSql Job

Next topic: Viewing Job Configuration and Logs

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

Which of the following issues have you encountered?

Content is inconsistent with the product UI

Unclear descriptions

Lack of examples or code

Incorrect steps

Can't find what I need

Lack of best practices

Feedback (optional)

0/500

Select at least one type of issue, and enter your comments or suggestions.

Enter a maximum of 500 characters.

Submit Cancel

For any further questions, feel free to contact us through the chatbot.

Chatbot