Typical Commands of the Flink Client

Updated on 2024-10-09 GMT+08:00

View PDF

Before you use the Flink shell script, perform the following operations. For details, see Using the Flink Client to run a wordcount job.

The Flink client has been installed, for example, in the /opt/client directory.
Run the following command to initialize environment variables:

source /opt/client/bigdata_env
If Kerberos authentication has been enabled for the cluster, configure client authentication by referring to 5. If Kerberos authentication is disabled, skip this step.

Run the related commands according to Table 1.

**Table 1** Flink Shell commands
Command	Description	Description
yarn-session.sh	-at,--applicationType <arg>: Defines the Yarn application type. -D <property=value>: Configures dynamic parameter. -d,--detached: Disables the interactive mode and starts a separate Flink Yarn session. -h,--help: Displays the help information about the Yarn session CLI. -id,--applicationId <arg>: Binds to a running Yarn session. -j,--jar <arg>: Sets the path of the user's JAR file. -jm,--jobManagerMemory <arg>: Sets the JobManager memory. -m,--jobmanager <arg>: Address of the JobManager (master) to which to connect. Use this parameter to connect to a specified JobManager. -nl,--nodeLabel <arg>: Specifies the nodeLabel of the Yarn application. -nm,--name <arg>: Customizes a name for the application on Yarn. -q,--query: Queries available Yarn resources. -qu,--queue <arg>: Specifies a Yarn queue. -s,--slots <arg>: Sets the number of slots for each TaskManager. -t,--ship <arg>: specifies the directory of the file to be sent. -tm,--taskManagerMemory <arg>: sets the TaskManager memory. -yd,--yarndetached: starts Yarn in the detached mode. -z,--zookeeperNamespace <args>: specifies the namespace of ZooKeeper. -h: Gets help information.	Start a resident Flink cluster to receive tasks from the Flink client.
flink run	-c,--class <classname>: Specifies a class as the entry for running programs. -C,--classpath <url>: Specifies classpath. -d,--detached: Runs a job in the detached mode. -files,--dependencyFiles <arg>: File on which the Flink program depends. This parameter is available only in MRS 3.2.0 and later versions. -n,--allowNonRestoredState: A state that cannot be restored can be skipped during restoration from a snapshot point in time. For example, if an operator in the program is deleted, you need to add this parameter when restoring the snapshot point. -m,--jobmanager <host:port>: Specifies the JobManager. -p,--parallelism <parallelism>: Specifies the job DOP, which will overwrite the DOP parameter in the configuration file. -q,--sysoutLogging: Disables the function of outputting Flink logs to the console. -s,--fromSavepoint <savepointPath>: Specifies a savepoint path for recovering jobs. -z,--zookeeperNamespace <zookeeperNamespace>: specifies the namespace of ZooKeeper. -yat,--yarnapplicationType <arg>: Defines the Yarn application type. -yD <arg>: Dynamic parameter configuration. -yd,--yarndetached: Starts Yarn in the detached mode. -yh,--yarnhelp: Obtains the Yarn help. -yid,--yarnapplicationId <arg>: Binds a job to a Yarn session. -yj,--yarnjar <arg>: Sets the path to Flink jar file. -yjm,--yarnjobManagerMemory <arg>: Sets the JobManager memory (MB). -ynm,--yarnname <arg>: Customizes a name for the application on Yarn. -yq,--yarnquery: Queries available Yarn resources (memory and CPUs). -yqu,--yarnqueue <arg>: Specifies a Yarn queue. -ys,--yarnslots: Sets the number of slots for each TaskManager. -yt,--yarnship <arg>: Specifies the path of the file to be sent. -ytm,--yarntaskManagerMemory <arg>: Sets the TaskManager memory (MB). -yz,--yarnzookeeperNamespace <arg>: Specifies the namespace of ZooKeeper. The value must be the same as the value of yarn-session.sh -z. -h: Gets help information.	Submit a Flink job. 1. The -y* parameter is used in the yarn-cluster mode. 2. If the parameter is not -y, you need to run the yarn-session* command to start the Flink cluster before running this command to submit a task.
flink info	-c,--class <classname>: Specifies a class as the entry for running programs. -p,--parallelism <parallelism>: Specifies the DOP for running programs. -h: Gets help information.	Display the execution plan (JSON) of the running program.
flink list	-a,--all: displays all jobs. -m,--jobmanager <host:port>: specifies the JobManager. -r,--running: displays only jobs in the running state. -s,--scheduled: displays only jobs in the scheduled state. -z,--zookeeperNamespace <zookeeperNamespace>: specifies the namespace of ZooKeeper. -yid,--yarnapplicationId <arg>: binds a job to a Yarn session. -h: gets help information.	Query running programs in the cluster.
flink stop	-d,--drain: sends MAX_WATERMARK before the savepoint is triggered and the job is stopped. -p,--savepointPath <savepointPath>: path for storing savepoints. The default value is state.savepoints.dir. -m,--jobmanager <host:port>: specifies the JobManager. -z,--zookeeperNamespace <zookeeperNamespace>: specifies the namespace of ZooKeeper. -yid,--yarnapplicationId <arg>: binds a job to a Yarn session. -h: gets help information.	Forcibly stop a running job (only streaming jobs are supported. StoppableFunction needs to be implemented on the source side in service code).
flink cancel	-m,--jobmanager <host:port>: specifies the JobManager. -s,--withSavepoint <targetDirectory>: triggers a savepoint when a job is canceled. The default directory is state.savepoints.dir. -z,--zookeeperNamespace <zookeeperNamespace>: specifies the namespace of ZooKeeper. -yid,--yarnapplicationId <arg>: binds a job to a Yarn session. -h: gets help information.	Cancel a running job.
flink savepoint	-d,--dispose <arg>: specifies a directory for storing the savepoint. -m,--jobmanager <host:port>: specifies the JobManager. -z,--zookeeperNamespace <zookeeperNamespace>: specifies the namespace of ZooKeeper. -yid,--yarnapplicationId <arg>: binds a job to a Yarn session. -h: gets help information.	Trigger a savepoint.
source Client installation directory/bigdata_env	None	Import client environment variables. Restriction: If the user uses a custom script (for example, A.sh) and runs this command in the script, variables cannot be imported to the A.sh script. If variables need to be imported to the custom script A.sh, the user needs to use the secondary calling method. For example, first call the B.sh script in the A.sh script, and then run this command in the B.sh script. Parameters can be imported to the A.sh script but cannot be imported to the B.sh script.
start-scala-shell.sh	local \| remote <host> <port> \| yarn: running mode	Start the scala shell.
sh generate_keystore.sh	-	Run the generate_keystore.sh script to generate security cookie, flink.keystore, and flink.truststore. You need to enter a user-defined password that does not contain number signs (#).

Parent topic: Using Flink

Previous topic: Separate Storage of Cold and Hot Data for Flink Job State Backend

Next topic: Common Flink SQL Syntax

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