Running a Flink Job

Video Tutorial

This tutorial submits a Flink job to on an MRS 3.1.0 cluster where Kerberos authentication is disabled, to process data stored in OBS.

The UI may vary depending on the version. This tutorial is for reference only.

Prerequisites

• You have uploaded the program packages and data files required by jobs to OBS or HDFS.
• If the job program needs to read and analyze data in the OBS file system, you need to configure storage-compute decoupling for the MRS cluster. For details, see Configuring Storage-Compute Decoupling for an MRS Cluster.

Submitting a Job on the Console

1. Log in to the MRS console.
2. On the Active Clusters page, select a running cluster and click its name to switch to the cluster details page.
3. In the Basic Information area of the Dashboard page, click Synchronize on the right side of IAM User Sync to synchronize IAM users.

   Perform this step only when Kerberos authentication is enabled for the cluster.

   

   • After the IAM user synchronization is complete, wait for 5 minutes before submitting a job. For details about IAM user synchronization, see Synchronizing IAM Users to MRS..
   • When the policy of the user group an IAM user belongs to changes from MRS ReadOnlyAccess to MRS CommonOperations, MRS FullAccess, or MRS Administrator, or vice versa, it takes time for the cluster node's System Security Services Daemon (SSSD) cache to refresh. To prevent job submission failure, wait for five minutes after user synchronization is complete before submitting the job with the new policy.
   • If the IAM username contains spaces (for example, admin 01), jobs cannot be added.

4. Click Job Management. On the displayed job list page, click Create.
5. Set Type to Flink and configure Flink job information be referring to Table 1.
   Figure 1 Creating a Flink job
   

   Table 1 Job configuration information

   Parameter	Description	Example
   Name	Job name. It contains 1 to 64 characters. Only letters, digits, hyphens (-), and underscores (_) are allowed.	flink_job
   Program Path	Path of the program package to be executed. You can enter the path or click HDFS or OBS to select a file. The value contains a maximum of 1,023 characters. It cannot contain special characters (;|&>,<'$) and cannot be left blank or all spaces. The OBS program path should start with obs://, for example, obs://wordcount/program/XXX.jar. The HDFS program path should start with hdfs://, for example, hdfs://hacluster/user/XXX.jar. The Flink job execution program must end with .jar.	-
   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 lists the common program parameters of Flink jobs. You can configure the parameters based on the execution program and cluster resources.	-
   Parameters	(Optional) Key parameter for program execution. The parameter is specified by the function of the custom program. MRS is only responsible for loading the parameters. Multiple parameters are separated by spaces. The value can contain a maximum of 150,000 characters and can be left blank. The value cannot contain special characters such as ;|&><'$ CAUTION: When entering a parameter containing sensitive information (for example, login password), you can add an at sign (@) before the parameter name to encrypt the parameter value. This prevents the sensitive information from being persisted in plaintext. When you view job information on the MRS console, the sensitive information is displayed as *. Example: username=testuser @password=User password	-
   Service Parameter	(Optional) Service parameters for the job. To modify the current job, change this parameter. For permanent changes to the entire cluster, refer to Modifying the Configuration Parameters of an MRS Cluster Component and modify the cluster component parameters accordingly. Click on the right to add more parameters. If a job needs to access OBS using AK/SK, add the following service configuration parameters: fs.obs.access.key: key ID for accessing OBS. fs.obs.secret.key: key corresponding to the key ID for accessing OBS.	-
   Command Reference	Command submitted to the background for execution when a job is submitted.	-

   Table 2 Flink job running program parameters

   Parameter	Description	Example
   -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
   -ys	Number of TaskManager cores.	2
   -ynm	Custom name of an application on YARN.	test
   -c	Class of the program entry method (for example, the main or getPlan() method). This parameter is required only when the JAR program does not specify the class of its manifest.	com.bigdata.mrs.test

6. Confirm job configuration information and click OK.
7. After the job is submitted, you can view the job running status and execution result in the job list. After the job status changes to Completed, you can view the analysis result of related programs.

Submitting a Job Using the Cluster Client

1. Install the MRS cluster client. For details, see Installing an MRS Cluster Client.

   The MRS cluster comes with a client installed for job submission by default, which can also be used directly. For MRS 3.x and later versions, the default client installation path is /opt/Bigdata/client on the Master node. For versions earlier than MRS 3.x, the default client installation path is /opt/client on the Master node.

2. Log in to the node where the client is located as the MRS cluster client installation user.
3. Run the following command to initialize environment variables:

   cd /opt/Bigdata/client

   source bigdata_env

4. Perform the following steps to create a user for submitting jobs and modify the security configuration of the cluster client only for clusters with Kerberos authentication enabled.
   1. Prepare a user for submitting Flink jobs.
      • MRS 3.x or earlier: For details, see Preparing a Development User.
      • MRS 3.x or later: For details, see Preparing a Development User.
   2. Log in to Manager as the newly created user.
      • For MRS 3.x earlier: 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.
      • For MRS 3.x or later: Log in to Manager of the cluster. Choose System > Permission > Manage User. On the displayed page, locate the row that contains the added user, click More in the Operation column, and select Download authentication credential.
   3. 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 files to the /etc/ directory on the node.
   4. For MRS 3.x or later: 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.
   5. 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

   6. 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. For MRS 3.x or earlier, external SSL is disabled by default in security clusters. To enable external SSL, run this script again after configuration. The configuration parameters do not exist in the default Flink configuration of MRS, if you enable SSL for external connections, you need to add the parameters listed in Table 3.

      Table 3 Parameter description

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

      

      • For MRS 3.x or earlier: The generate_keystore.sh script is automatically generated.
      • The generated flink.keystore, flink.truststore, and security.cookie items are automatically filled in the corresponding configuration items in flink-conf.yaml.
      • For MRS 3.x or later: 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.

   7. 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. For MRS 3.x or later: 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

        4. For MRS 3.x or earlier: Change the values of the following parameters in the flink-conf.yaml file to relative paths:

           security.ssl.internal.keystore: ssl/flink.keystore
           security.ssl.internal.truststore: ssl/flink.truststore

   8. 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

5. Run the Flink job.

   This section uses the WordCount sample program provided by the client as an example.

   • 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