命令行界面

译者：flink.sojb.cn

Flink提供命令行界面（CLI）来运行打包为JAR文件的程序，并控制它们的执行。CLI是任何Flink设置的一部分，可在本地单节点设置和分布式设置中使用。它位于<flink-home>/bin/flink 默认情况下，并连接到从同一安装目录启动的正在运行的Flink主服务器（JobManager）。

使用命令行界面的先决条件是Flink主机（JobManager）已启动（通过 <flink-home>/bin/start-cluster.sh）或YARN环境可用。

命令行可用于

例子

  • 运行没有参数的示例程序：

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

  • 使用输入和结果文件的参数运行示例程序：

    1. ./bin/flink run ./examples/batch/WordCount.jar \
    2.                      --input file:///home/user/hamlet.txt --output file:///home/user/wordcount_out

  • 运行带有并行性的示例程序16以及输入和结果文件的参数：

    1. ./bin/flink run -p 16 ./examples/batch/WordCount.jar \
    2.                      --input file:///home/user/hamlet.txt --output file:///home/user/wordcount_out

  • 运行禁用flink log输出的示例程序：

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

  • 以分离模式运行示例程序：

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

  • 在特定的JobManager上运行示例程序：

    1. ./bin/flink run -m myJMHost:8081 \
    2.                        ./examples/batch/WordCount.jar \
    3.                        --input file:///home/user/hamlet.txt --output file:///home/user/wordcount_out

  • 以特定类作为入口点运行示例程序：

    1. ./bin/flink run -c org.apache.flink.examples.java.wordcount.WordCount \
    2.                        ./examples/batch/WordCount.jar \
    3.                        --input file:///home/user/hamlet.txt --output file:///home/user/wordcount_out

  • 使用具有2个TaskManagers 的每作业YARN群集运行示例程序：

    1. ./bin/flink run -m yarn-cluster -yn 2 \
    2.                        ./examples/batch/WordCount.jar \
    3.                        --input hdfs:///user/hamlet.txt --output hdfs:///user/wordcount_out

  • 以Word为单位显示WordCount示例程序的优化执行计划：

    1. ./bin/flink info ./examples/batch/WordCount.jar \
    2.                         --input file:///home/user/hamlet.txt --output file:///home/user/wordcount_out

  • 列出计划和正在运行的作业（包括其JobID）：

    1. ./bin/flink list

  • 列出预定作业（包括其作业ID）：

    1. ./bin/flink list -s

  • 列出正在运行的作业（包括其作业ID）：

    1. ./bin/flink list -r

  • 列出所有现有工作（包括其作业ID）：

    1. ./bin/flink list -a

  • 列出在Flink YARN会话中运行Flink作业：

    1. ./bin/flink list -m yarn-cluster -yid <yarnApplicationID> -r

  • 取消工作：

    1. ./bin/flink cancel <jobID>

  • 使用保存点取消作业：

    1. ./bin/flink cancel -s [targetDirectory] <jobID>

  • 停止工作（仅限流处理工作）：

    1. ./bin/flink stop <jobID>

  • 修改正在运行的作业（仅限流式处理作业）:. / bin/flink modify <jobid>-p<newparallelism></newparallelism></jobid>

注意：取消和停止（流处理）作业的区别如下：

在取消呼叫中，作业中的算子立即接收cancel()方法调用以尽快取消它们。如果算子在取消呼叫后没有停止，Flink将开始定期中断线程，直到它停止。

“停止”呼叫是一种更优雅的方式来停止正在运行的流处理作业。Stop仅适用于使用实现StoppableFunction接口的源的作业。当用户请求停止作业时，所有源都将接收stop()方法调用。该工作将继续运行，直到所有资源正常关闭。这允许作业完成处理所有飞行数据。

保存点

保存点通过命令行客户端控制：

触发保存点

  1. ./bin/flink savepoint <jobId> [savepointDirectory]

这将触发具有ID的作业的保存点jobId，并返回创建的保存点的路径。您需要此路径来还原和部署保存点。

此外，您可以选择指定目标文件系统目录以存储保存点。该目录需要可由JobManager访问。

如果未指定目标目录，则需要配置默认目录。否则，触发保存点将失败。

使用YARN触发保存点

  1. ./bin/flink savepoint <jobId> [savepointDirectory] -yid <yarnAppId>

这将触发具有ID jobId和YARN应用程序ID 的作业的保存点yarnAppId，并返回创建的保存点的路径。

其他所有内容与上面触发保存点部分中描述的相同。

使用保存点取消

您可以自动触发保存点并取消作业。

  1. ./bin/flink cancel -s [savepointDirectory] <jobID>

如果未配置保存点目录，则需要为Flink安装配置默认保存点目录（请参阅保存点）。

只有保存点成功，才会取消该作业。

恢复保存点

  1. ./bin/flink run -s <savepointPath> ...

run命令有一个保存点标志来提交作业，该作业从保存点恢复其状态。savepoint trigger命令返回保存点路径。

默认情况下，我们尝试将所有保存点状态与正在提交的作业进行匹配。如果要允许跳过无法使用新作业恢复的保存点状态，可以设置allowNonRestoredState标志。如果在触发保存点并且仍想使用保存点时从程序中删除了作为程序一部分的 算子，则需要允许此 算子操作。

  1. ./bin/flink run -s <savepointPath> -n ...

如果您的程序删除了属于保存点的 算子，这将非常有用。

配置保存点

  1. ./bin/flink savepoint -d <savepointPath>

在给定路径处理保存点。savepoint trigger命令返回保存点路径。

如果使用自定义状态实例（例如自定义还原状态或RocksDB状态），则必须指定触发保存点的程序JAR的路径，以便使用用户代码类加载器处理保存点：

  1. ./bin/flink savepoint -d <savepointPath> -j <jarFile>

否则，你会遇到一个ClassNotFoundException。

用法

命令行语法如下：

  1. ./flink <ACTION> [OPTIONS] [ARGUMENTS]
  3. The following actions are available:
  5. Action "run" compiles and runs a program.
  7.   Syntax: run [OPTIONS] <jar-file> <arguments>
  8.   "run" action options:
  9.      -c,--class <classname>               Class with the program entry point
  10.                                           ("main" method or "getPlan()" method.
  11.                                           Only needed if the JAR file does not
  12.                                           specify the class in its manifest.
  13.      -C,--classpath <url>                 Adds a URL to each user code
  14.                                           classloader  on all nodes in the
  15.                                           cluster. The paths must specify a
  16.                                           protocol (e.g. file://) and be
  17.                                           accessible on all nodes (e.g. by means
  18.                                           of a NFS share). You can use this
  19.                                           option multiple times for specifying
  20.                                           more than one URL. The protocol must
  21.                                           be supported by the {@link
  22.                                           java.net.URLClassLoader}.
  23.      -d,--detached                        If present, runs the job in detached
  24.                                           mode
  25.      -n,--allowNonRestoredState           Allow to skip savepoint state that
  26.                                           cannot be restored. You need to allow
  27.                                           this if you removed an operator from
  28.                                           your program that was part of the
  29.                                           program when the savepoint was
  30.                                           triggered.
  31.      -p,--parallelism <parallelism>       The parallelism with which to run the
  32.                                           program. Optional flag to override the
  33.                                           default value specified in the
  34.                                           configuration.
  35.      -q,--sysoutLogging                   If present, suppress logging output to
  36.                                           standard out.
  37.      -s,--fromSavepoint <savepointPath>   Path to a savepoint to restore the job
  38.                                           from (for example
  39.                                           hdfs:///flink/savepoint-1537).
  40.   Options for yarn-cluster mode:
  41.      -d,--detached                        If present, runs the job in detached
  42.                                           mode
  43.      -m,--jobmanager <arg>                Address of the JobManager (master) to
  44.                                           which to connect. Use this flag to
  45.                                           connect to a different JobManager than
  46.                                           the one specified in the
  47.                                           configuration.
  48.      -yD <property=value>                 use value for given property
  49.      -yd,--yarndetached                   If present, runs the job in detached
  50.                                           mode (deprecated; use non-YARN
  51.                                           specific option instead)
  52.      -yh,--yarnhelp                       Help for the Yarn session CLI.
  53.      -yid,--yarnapplicationId <arg>       Attach to running YARN session
  54.      -yj,--yarnjar <arg>                  Path to Flink jar file
  55.      -yjm,--yarnjobManagerMemory <arg>    Memory for JobManager Container
  56.                                           with optional unit (default: MB)
  57.      -yn,--yarncontainer <arg>            Number of YARN container to allocate
  58.                                           (=Number of Task Managers)
  59.      -ynm,--yarnname <arg>                Set a custom name for the application
  60.                                           on YARN
  61.      -yq,--yarnquery                      Display available YARN resources
  62.                                           (memory, cores)
  63.      -yqu,--yarnqueue <arg>               Specify YARN queue.
  64.      -ys,--yarnslots <arg>                Number of slots per TaskManager
  65.      -yst,--yarnstreaming                 Start Flink in streaming mode
  66.      -yt,--yarnship <arg>                 Ship files in the specified directory
  67.                                           (t for transfer)
  68.      -ytm,--yarntaskManagerMemory <arg>   Memory per TaskManager Container
  69.                                           with optional unit (default: MB)
  70.      -yz,--yarnzookeeperNamespace <arg>   Namespace to create the Zookeeper
  71.                                           sub-paths for high availability mode
  72.      -ynl,--yarnnodeLabel <arg>           Specify YARN node label for the YARN application
  73.      -z,--zookeeperNamespace <arg>        Namespace to create the Zookeeper
  74.                                           sub-paths for high availability mode
  76.   Options for default mode:
  77.      -m,--jobmanager <arg>           Address of the JobManager (master) to which
  78.                                      to connect. Use this flag to connect to a
  79.                                      different JobManager than the one specified
  80.                                      in the configuration.
  81.      -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
  82.                                      for high availability mode
  84. Action "info" shows the optimized execution plan of the program (JSON).
  86.   Syntax: info [OPTIONS] <jar-file> <arguments>
  87.   "info" action options:
  88.      -c,--class <classname>           Class with the program entry point ("main"
  89.                                       method or "getPlan()" method. Only needed
  90.                                       if the JAR file does not specify the class
  91.                                       in its manifest.
  92.      -p,--parallelism <parallelism>   The parallelism with which to run the
  93.                                       program. Optional flag to override the
  94.                                       default value specified in the
  95.                                       configuration.
  97. Action "list" lists running and scheduled programs.
  99.   Syntax: list [OPTIONS]
  100.   "list" action options:
  101.      -r,--running     Show only running programs and their JobIDs
  102.      -s,--scheduled   Show only scheduled programs and their JobIDs
  103.   Options for yarn-cluster mode:
  104.      -m,--jobmanager <arg>            Address of the JobManager (master) to
  105.                                       which to connect. Use this flag to connect
  106.                                       to a different JobManager than the one
  107.                                       specified in the configuration.
  108.      -yid,--yarnapplicationId <arg>   Attach to running YARN session
  109.      -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
  110.                                       sub-paths for high availability mode
  112.   Options for default mode:
  113.      -m,--jobmanager <arg>           Address of the JobManager (master) to which
  114.                                      to connect. Use this flag to connect to a
  115.                                      different JobManager than the one specified
  116.                                      in the configuration.
  117.      -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
  118.                                      for high availability mode
  120. Action "stop" stops a running program (streaming jobs only).
  122.   Syntax: stop [OPTIONS] <Job ID>
  123.   "stop" action options:
  125.   Options for yarn-cluster mode:
  126.      -m,--jobmanager <arg>            Address of the JobManager (master) to
  127.                                       which to connect. Use this flag to connect
  128.                                       to a different JobManager than the one
  129.                                       specified in the configuration.
  130.      -yid,--yarnapplicationId <arg>   Attach to running YARN session
  131.      -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
  132.                                       sub-paths for high availability mode
  134.   Options for default mode:
  135.      -m,--jobmanager <arg>           Address of the JobManager (master) to which
  136.                                      to connect. Use this flag to connect to a
  137.                                      different JobManager than the one specified
  138.                                      in the configuration.
  139.      -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
  140.                                      for high availability mode
  142. Action "cancel" cancels a running program.
  144.   Syntax: cancel [OPTIONS] <Job ID>
  145.   "cancel" action options:
  146.      -s,--withSavepoint <targetDirectory>   Trigger savepoint and cancel job.
  147.                                             The target directory is optional. If
  148.                                             no directory is specified, the
  149.                                             configured default directory
  150.                                             (state.savepoints.dir) is used.
  151.   Options for yarn-cluster mode:
  152.      -m,--jobmanager <arg>            Address of the JobManager (master) to
  153.                                       which to connect. Use this flag to connect
  154.                                       to a different JobManager than the one
  155.                                       specified in the configuration.
  156.      -yid,--yarnapplicationId <arg>   Attach to running YARN session
  157.      -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
  158.                                       sub-paths for high availability mode
  160.   Options for default mode:
  161.      -m,--jobmanager <arg>           Address of the JobManager (master) to which
  162.                                      to connect. Use this flag to connect to a
  163.                                      different JobManager than the one specified
  164.                                      in the configuration.
  165.      -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
  166.                                      for high availability mode
  168. Action "savepoint" triggers savepoints for a running job or disposes existing ones.
  170.   Syntax: savepoint [OPTIONS] <Job ID> [<target directory>]
  171.   "savepoint" action options:
  172.      -d,--dispose <arg>       Path of savepoint to dispose.
  173.      -j,--jarfile <jarfile>   Flink program JAR file.
  174.   Options for yarn-cluster mode:
  175.      -m,--jobmanager <arg>            Address of the JobManager (master) to
  176.                                       which to connect. Use this flag to connect
  177.                                       to a different JobManager than the one
  178.                                       specified in the configuration.
  179.      -yid,--yarnapplicationId <arg>   Attach to running YARN session
  180.      -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
  181.                                       sub-paths for high availability mode
  183.   Options for default mode:
  184.      -m,--jobmanager <arg>           Address of the JobManager (master) to which
  185.                                      to connect. Use this flag to connect to a
  186.                                      different JobManager than the one specified
  187.                                      in the configuration.
  188.      -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
  189.                                      for high availability mode
  191. Action "modify" modifies a running job (e.g. change of parallelism).
  193.   Syntax: modify <Job ID> [OPTIONS]
  194.   "modify" action options:
  195.      -h,--help                           Show the help message for the CLI
  196.                                          Frontend or the action.
  197.      -p,--parallelism <newParallelism>   New parallelism for the specified job.
  198.      -v,--verbose                        This option is deprecated.
  199.   Options for yarn-cluster mode:
  200.      -m,--jobmanager <arg>            Address of the JobManager (master) to
  201.                                       which to connect. Use this flag to connect
  202.                                       to a different JobManager than the one
  203.                                       specified in the configuration.
  204.      -yid,--yarnapplicationId <arg>   Attach to running YARN session
  205.      -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
  206.                                       sub-paths for high availability mode
  208.   Options for default mode:
  209.      -m,--jobmanager <arg>           Address of the JobManager (master) to which
  210.                                      to connect. Use this flag to connect to a
  211.                                      different JobManager than the one specified
  212.                                      in the configuration.
  213.      -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
  214.                                      for high availability mode