Help Center/ MapReduce Service/ User Guide (Kuala Lumpur Region)/ Managing Clusters/ Job Management/ Running a Flink Job
Updated on 2022-08-12 GMT+08:00
View PDF
Share

Running a Flink Job

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

  1. Log in to the MRS console.
  2. Choose Clusters > Active Clusters, select a running cluster, and click its name to switch to the cluster details page.
  3. 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. For details, see Synchronizing IAM Users to MRS.

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

  4. Click the Jobs tab.
  5. Click Create. The Create Job page is displayed.
  6. 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

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

    -

  7. Confirm job configuration information and click OK.

    After the job is created, you can manage it.

Submitting a Job in the Background

In MRS 3.x and later versions, the default installation path of the client is /opt/Bigdata/client. In MRS 3.x and earlier versions, the default installation path is /opt/client. For details, see the actual situation.

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

    source /opt/Bigdata/client/bigdata_env

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

    1. Prepare a user for submitting Flink jobs.
    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 user.keytab file to the client node, for example, to the /opt/Bigdata/client/Flink/flink/conf directory on the client node. If the client is installed on a node outside the cluster, copy the krb5.conf file to the /etc/ directory on this 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. Run the following command to perform security hardening in the bin directory of the Flink client. Set password to a new password for submitting jobs.

      sh generate_keystore.sh <password>

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

      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.
      • For MRS 3.x or earlier: The generate_keystore.sh script is automatically generated.
      • Perform Authentication and Encryption. The generated flink.keystore, flink.truststore, and security.cookie files 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.

    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

  4. 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"
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"
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"
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