Submitting a Flink Job

Describes about how to submit a Flink job.

IMPORTANT
The DEP-CLI and Flink-CLI works same, and the commands are same too. The only difference is that all commands in DEP-CLI start with the /opt/mapr/flink/flink-<version>/bin/ prefix.
IMPORTANT
You can see the Flink documentation page: Command-Line Interface for the latest information. Following commands for starting/stopping a Flink cluster are shown as an example.

Session mode

To submit a job in session mode, run the following command:

/opt/mapr/flink/flink-<version>/bin/flink run -d <job_jar>

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 might want to 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>