Starting and Stopping a Flink Cluster

Describes how to start and stop a Flink cluster.

Deployment Modes

Only YARN is supported as a resource manager for Flink, in DEP 10.0.0. Flink on YARN supports both session and application modes of deployment. For more information, see the following Apache Flink documentation pages:
IMPORTANT
The DEP-Flink CLI and Apache-Flink CLI are similar, and the commands are same for both CLIs. The only difference is that all commands start with the /opt/mapr/flink/flink-<version>/bin/ prefix in DEP-Flink CLI.
IMPORTANT
You can see the Flink documentation page: Command-Line Interface for the latest information. The following commands for starting/stopping a Flink cluster are shown as an example.

Starting a Flink Cluster

Session mode
Use the yarn-session.sh script, and start the Flink cluster with the following command:
/opt/mapr/flink/flink-<version>/bin/yarn-session.sh --detached
Application mode
In application mode, use the following command for both starting a cluster, and submitting a job:
/opt/mapr/flink/flink-<version>/bin/flink run -t yarn-application <job_jar>
NOTE
  • With Application mode, you can configure yarn.classpath.include-user-jar to DISABLED to avoid the error similar to:
    org.apache.flink.api.common.functions.InvalidTypesException: The types of the interface
      <class> could not be inferred. Support for synthetic interfaces, lambdas, and generic or raw
      types is limited at this point
    for example, you can add the following to config.yaml file to avoid the preceding error:
    yarn:
  classpath:
    include-user-jar: DISABLED
  • Also, if you use any Schema Registry client libraries, ensure that you exclude all org.apache.zookeeper:* dependencies. for example:
    <dependency>
            <groupId>io.confluent</groupId>
            <artifactId>kafka-avro-serializer</artifactId>
            <version>${schema.registry.version}</version>
            <exclusions>
                    <exclusion>
                            <groupId>org.apache.zookeeper</groupId>
                            <artifactId>zookeeper</artifactId>
                    </exclusion>
            </exclusions>
        </dependency>

Stopping a Flink cluster

Session Mode

To gracefully stop a cluster, you must have the YARN application ID of the cluster. See the YARN application list or search in Flink logs to find the ID.

Once you have the YARN application ID, use the following command to stop the cluster:
echo "stop" | /opt/mapr/flink/flink-<version>/bin/yarn-session.sh -id <application_id>
NOTE
  • The user who started the Flink cluster can only stop the cluster gracefully.
  • Cluster administrator can end the YARN application directly. However, if you end the application, some uncleaned resources may remain.
Application Mode
  1. To stop a Flink cluster in application mode, you need to end all jobs running on the cluster. To end a particular job, you must have both YARN application id and the Job ID. The Job ID is printed in console upon submitting a job. Alternatively, you can also find the Job ID by listing all the running jobs on the cluster as follows:
    /opt/mapr/flink/flink-<version>/bin/flink list -t yarn-application -Dyarn.application.id=<application_id>
  2. Using Flink CLI, you can cancel (ending without saving point) a job, as shown in the following example:
    /opt/mapr/flink/flink-<version>/bin/flink cancel -t yarn-application -Dyarn.application.id=<application_id> <job_id>

Configuring mapR ticket expiration time

Any long-running YARN job may fail due to ticket expiration. Before submitting a YARN job that runs for a longer period, consider configuring the expiration time.

You can configure the timing in either of the following ways:
  • Globally (for all YARN jobs in a cluster), specify the yarn.mapr.ticket.expiration value in yarn-site.xml (default value, one week). Or
  • Alternatively, you can increase the expiration time of a specific YARN job (default value, two weeks), as follows:
    1. Generate a ticket and specify the duration:
      maprlogin password -duration <duration>
    2. Submit a YARN job (start the Flink cluster).
    3. (Optional) Remove the generated ticket: maprlogin logout
      NOTE
      YARN chooses the maximum expiration time between the two values from the generated tickets.