Flink提供了一個命令行接口(CLI)用來運行打成JAR包的程序，并且可以控制程序的運行。命令行接口在Flink安裝完之后即可擁有，本地單節點或是分布式的部署安裝都會有命令行接口。命令行接口啟動腳本是 $FLINK_HOME/bin目錄下的flink腳本， 默認情況下會連接運行中的Flink master(JobManager)，JobManager的啟動腳本與CLI在同一安裝目錄下。

使用命令行接口的先決條件是JobManager已經被啟動或是在Flink YARN環境下。JobManager可以通過如下命令啟動:

$FLINK_HOME/bin/start-local.sh或$FLINK_HOME/bin/start-cluster.sh

1. Example

(1) 運行示例程序，不傳參數：

./bin/flink run ./examples/batch/WordCount.jar

(2) 運行示例程序，帶輸入和輸出文件參數：

./bin/flink run ./examples/batch/WordCount.jar --input file:///home/xiaosi/a.txt --output file:///home/xiaosi/result.txt

(3) 運行示例程序，帶輸入和輸出文件參數,并設置16個并發度：

./bin/flink run -p 16 ./examples/batch/WordCount.jar --input file:///home/xiaosi/a.txt --output file:///home/xiaosi/result.txt

(4) 運行示例程序，并禁止Flink輸出日志

./bin/flink run -q ./examples/batch/WordCount.jar

(5) 以獨立(detached)模式運行示例程序

./bin/flink run -d ./examples/batch/WordCount.jar

(6) 在指定JobManager上運行示例程序

./bin/flink run -m myJMHost:6123 ./examples/batch/WordCount.jar --input file:///home/xiaosi/a.txt --output file:///home/xiaosi/result.txt

(7) 運行示例程序，指定程序入口類(Main方法所在類)：

./bin/flink run -c org.apache.flink.examples.java.wordcount.WordCount ./examples/batch/WordCount.jar --input file:///home/xiaosi/a.txt --output file:///home/xiaosi/result.txt

(8) 運行示例程序，使用per-job YARN 集群啟動 2 個TaskManager

./bin/flink run -m yarn-cluster -yn 2 ./examples/batch/WordCount.jar --input hdfs:///xiaosi/a.txt --output hdfs:///xiaosi/result.txt

(9) 以JSON格式輸出 WordCount示例程序優化執行計劃：

./bin/flink info ./examples/batch/WordCount.jar --input file:///home/xiaosi/a.txt --output file:///home/xiaosi/result.txt

(10) 列出已經調度的和正在運行的Job(包含Job ID信息)

./bin/flink list

(11) 列出已經調度的Job(包含Job ID信息)

./bin/flink list -s

(13) 列出正在運行的Job(包含Job ID信息)

./bin/flink list -r

(14) 列出在Flink YARN中運行Job

./bin/flink list -m yarn-cluster -yid -r

(15) 取消一個Job

./bin/flink cancel

(16) 取消一個帶有保存點(savepoint)的Job

./bin/flink cancel -s [targetDirectory]

(17) 停止一個Job(只適用于流計算Job)

./bin/flink stop

備注:

取消和停止Job區別如下：

調用取消Job時，作業中的operator立即收到一個調用cancel()方法的指令以盡快取消它們。如果operator在調用取消操作后沒有停止，Flink將定期開啟中斷線程來取消作業直到作業停止。

調用停止Job是一種停止正在運行的流作業的更加優雅的方法。停止僅適用于使用實現`StoppableFunction`接口的源的那些作業。當用戶請求停止作業時，所有源將收到調用stop()方法指令。但是Job還是會持續運行，直到所有來源已經正確關閉。這允許作業完成處理所有正在傳輸的數據(inflight data)。

2. 保存點

保存點通過命令行客戶端進行控制：

2.1 觸發保存點

./bin/flink savepoint [savepointDirectory]

返回創建的保存點的路徑。你需要此路徑來還原和處理保存點。

觸發保存點時，可以選擇是否指定savepointDirectory。如果在此處未指定，則需要為Flink安裝配置默認的保存點目錄(請參閱保存點)。

2.2 根據保存點取消Job

你可以自動觸發保存點并取消一個Job:

./bin/flink cancel -s [savepointDirectory]

如果沒有指定保存點目錄，則需要為Flink安裝配置默認的保存點目錄(請參閱保存點)。如果保存點觸發成功，該作業將被取消

2.3 恢復保存點

./bin/flink run -s ...

這個run命令提交Job時帶有一個保存點標記，這使得程序可以從保存點中恢復狀態。保存點路徑是通過保存點觸發命令得到的。

默認情況下，我們嘗試將所有保存點狀態與正在提交的作業相匹配。 如果要允許跳過那些無法使用它恢復新作業的保存點狀態(allow to skip savepoint state that cannot be restored with the new job)，則可以設置allowNonRestoredState標志。如果當保存點觸發時，從你程序中刪除了作為程序一部分的operator，但是仍然要使用保存點，則需要允許這一點(You need to allow this if you removed an operator from your program that was part of the program when the savepoint was triggered and you still want to use the savepoint.)。

./bin/flink run -s -n ...

如果你的程序刪除了作為保存點一部分的operator,這時會非常有用(This is useful if your program dropped an operator that was part of the savepoint.)。

2.4 銷毀保存點

./bin/flink savepoint -d

銷毀一個保存點同樣需要一個路徑。這個保存點路徑是通過保存點觸發命令得到的。

3. 用法

下面是Flink命令行接口的用法:

xiaosi@yoona:~/qunar/company/opt/flink-1.3.2$ ./bin/flink

./flink [OPTIONS] [ARGUMENTS]

The following actions are available:

Action "run" compiles and runs a program.

Syntax: run [OPTIONS]

"run" action options:

-c,--class Class with the program entrypoint ("main" method or

"getPlan()" method. Only

needed if the JAR file does

not specify the class in its

manifest.

-C,--classpath Adds a URL to each user codeclassloader on all nodes in

the cluster. The paths must

specify a protocol (e.g.

file://) and be accessible

on all nodes (e.g. by means

of a NFS share). You can use

this option multiple times

for specifying more than one

URL. The protocol must be

supported by the {@link

java.net.URLClassLoader}.

-d,--detached If present, runs the job indetached mode

-m,--jobmanager Address of the JobManager(master) to which to

connect. Use this flag to

connect to a different

JobManager than the one

specified in the

configuration.

-n,--allowNonRestoredState Allow to skip savepointstate that cannot be

restored. You need to allow

this if you removed an

operator from your program

that was part of the program

when the savepoint was

triggered.

-p,--parallelism The parallelism with whichto run the program. Optional

flag to override the default

value specified in the

configuration.

-q,--sysoutLogging If present, suppress loggingoutput to standard out.

-s,--fromSavepoint Path to a savepoint torestore the job from (for

example

hdfs:///flink/savepoint-1537

).

-z,--zookeeperNamespace Namespace to create theZookeeper sub-paths for high

availability mode

Options for yarn-cluster mode:

-yD Dynamic properties-yd,--yarndetached Start detached-yid,--yarnapplicationId Attach to running YARN session-yj,--yarnjar Path to Flink jar file-yjm,--yarnjobManagerMemory Memory for JobManager Container [inMB]

-yn,--yarncontainer Number of YARN container to allocate(=Number of Task Managers)

-ynm,--yarnname Set a custom name for the applicationon YARN

-yq,--yarnquery Display available YARN resources(memory, cores)

-yqu,--yarnqueue Specify YARN queue.-ys,--yarnslots Number of slots per TaskManager-yst,--yarnstreaming Start Flink in streaming mode-yt,--yarnship Ship files in the specified directory(t for transfer)

-ytm,--yarntaskManagerMemory Memory per TaskManager Container [inMB]

-yz,--yarnzookeeperNamespace Namespace to create the Zookeepersub-paths for high availability mode

Options for yarn mode:

-ya,--yarnattached Start attached-yD Dynamic properties-yj,--yarnjar Path to Flink jar file-yjm,--yarnjobManagerMemory Memory for JobManager Container [inMB]

-yqu,--yarnqueue Specify YARN queue.-yt,--yarnship Ship files in the specified directory(t for transfer)

-yz,--yarnzookeeperNamespace Namespace to create the Zookeepersub-paths for high availability mode

Action "info" shows the optimized execution plan of the program (JSON).

Syntax: info [OPTIONS]

"info" action options:

-c,--class Class with the program entry point ("main"method or "getPlan()" method. Only needed

if the JAR file does not specify the class

in its manifest.

-p,--parallelism The parallelism with which to run theprogram. Optional flag to override the

default value specified in the

configuration.

Options for yarn-cluster mode:

-yid,--yarnapplicationId Attach to running YARN session

Options for yarn mode:

Action "list" lists running and scheduled programs.

Syntax: list [OPTIONS]

"list" action options:

-m,--jobmanager Address of the JobManager (master) to whichto connect. Use this flag to connect to a

different JobManager than the one specified

in the configuration.

-r,--running Show only running programs and their JobIDs-s,--scheduled Show only scheduled programs and their JobIDsOptions for yarn-cluster mode:

-yid,--yarnapplicationId Attach to running YARN session

Options for yarn mode:

Action "stop" stops a running program (streaming jobs only).

Syntax: stop [OPTIONS]

"stop" action options:

-m,--jobmanager Address of the JobManager (master) to whichto connect. Use this flag to connect to a

different JobManager than the one specified

in the configuration.

Options for yarn-cluster mode:

-yid,--yarnapplicationId Attach to running YARN session

Options for yarn mode:

Action "cancel" cancels a running program.

Syntax: cancel [OPTIONS]

"cancel" action options:

-m,--jobmanager Address of the JobManager (master)to which to connect. Use this flag

to connect to a different JobManager

than the one specified in the

configuration.

-s,--withSavepoint Trigger savepoint and cancel job.The target directory is optional. If

no directory is specified, the

configured default directory

(state.savepoints.dir) is used.

Options for yarn-cluster mode:

-yid,--yarnapplicationId Attach to running YARN session

Options for yarn mode:

Action "savepoint" triggers savepoints for a running job or disposes existing ones.

Syntax: savepoint [OPTIONS] []

"savepoint" action options:

-d,--dispose Path of savepoint to dispose.-j,--jarfile Flink program JAR file.-m,--jobmanager Address of the JobManager (master) to whichto connect. Use this flag to connect to a

different JobManager than the one specified

in the configuration.

Options for yarn-cluster mode:

-yid,--yarnapplicationId Attach to running YARN session

Options for yarn mode:

Please specify an action.

備注:

Flink版本:1.3

由于翻譯者水平有限，如果有問題，歡迎指正。在閱讀時可以參考原文閱讀。

總結

以上是生活随笔為你收集整理的flink启动命令参数_[Flink]Flink1.3 指南四 命令行接口-阿里云开发者社区的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。