Flink Yarn 作业提交

心已赠人 2022-03-18 08:28 419阅读 0赞

1、下载测试文件

wget -O LICENSE-2.0.txt http://www.apache.org/licenses/LICENSE-2.0.txt

2、将文件上传至hdfs

hdfs dfs -put LICENSE-2.0.txt  /user/flink

3、提交作业

使用以下命令将Flink程序提交到YARN群集：

./bin/flink

该命令将显示如下的帮助菜单：

[flink@big-data-1 flink-1.7.2]$ ./bin/flink
    2019-02-20 15:35:14,263 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-flink.
    2019-02-20 15:35:14,263 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-flink.
    ./flink <ACTION> [OPTIONS] [ARGUMENTS]
    
    The following actions are available:
    
    Action "run" compiles and runs a program.
    
      Syntax: run [OPTIONS] <jar-file> <arguments>
      "run" action options:
         -c,--class <classname>               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.
         -C,--classpath <url>                 Adds a URL to each user code
                                              classloader  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 in detached
                                              mode
         -n,--allowNonRestoredState           Allow to skip savepoint state 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 <parallelism>       The parallelism with which to run the
                                              program. Optional flag to override the
                                              default value specified in the
                                              configuration.
         -q,--sysoutLogging                   If present, suppress logging output to
                                              standard out.
         -s,--fromSavepoint <savepointPath>   Path to a savepoint to restore the job
                                              from (for example
                                              hdfs:///flink/savepoint-1537).
         -sae,--shutdownOnAttachedExit        If the job is submitted in attached
                                              mode, perform a best-effort cluster
                                              shutdown when the CLI is terminated
                                              abruptly, e.g., in response to a user
                                              interrupt, such as typing Ctrl + C.
      Options for yarn-cluster mode:
         -d,--detached                        If present, runs the job in detached
                                              mode
         -m,--jobmanager <arg>                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.
         -sae,--shutdownOnAttachedExit        If the job is submitted in attached
                                              mode, perform a best-effort cluster
                                              shutdown when the CLI is terminated
                                              abruptly, e.g., in response to a user
                                              interrupt, such as typing Ctrl + C.
         -yD <property=value>                 use value for given property
         -yd,--yarndetached                   If present, runs the job in detached
                                              mode (deprecated; use non-YARN
                                              specific option instead)
         -yh,--yarnhelp                       Help for the Yarn session CLI.
         -yid,--yarnapplicationId <arg>       Attach to running YARN session
         -yj,--yarnjar <arg>                  Path to Flink jar file
         -yjm,--yarnjobManagerMemory <arg>    Memory for JobManager Container with
                                              optional unit (default: MB)
         -yn,--yarncontainer <arg>            Number of YARN container to allocate
                                              (=Number of Task Managers)
         -ynl,--yarnnodeLabel <arg>           Specify YARN node label for the YARN
                                              application
         -ynm,--yarnname <arg>                Set a custom name for the application
                                              on YARN
         -yq,--yarnquery                      Display available YARN resources
                                              (memory, cores)
         -yqu,--yarnqueue <arg>               Specify YARN queue.
         -ys,--yarnslots <arg>                Number of slots per TaskManager
         -yst,--yarnstreaming                 Start Flink in streaming mode
         -yt,--yarnship <arg>                 Ship files in the specified directory
                                              (t for transfer)
         -ytm,--yarntaskManagerMemory <arg>   Memory per TaskManager Container with
                                              optional unit (default: MB)
         -yz,--yarnzookeeperNamespace <arg>   Namespace to create the Zookeeper
                                              sub-paths for high availability mode
         -z,--zookeeperNamespace <arg>        Namespace to create the Zookeeper
                                              sub-paths for high availability mode
    
      Options for default mode:
         -m,--jobmanager <arg>           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.
         -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                         for high availability mode
    
    
    
    Action "info" shows the optimized execution plan of the program (JSON).
    
      Syntax: info [OPTIONS] <jar-file> <arguments>
      "info" action options:
         -c,--class <classname>           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 <parallelism>   The parallelism with which to run the
                                          program. Optional flag to override the
                                          default value specified in the
                                          configuration.
    
    
    Action "list" lists running and scheduled programs.
    
      Syntax: list [OPTIONS]
      "list" action options:
         -r,--running     Show only running programs and their JobIDs
         -s,--scheduled   Show only scheduled programs and their JobIDs
      Options for yarn-cluster mode:
         -m,--jobmanager <arg>            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.
         -yid,--yarnapplicationId <arg>   Attach to running YARN session
         -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                          sub-paths for high availability mode
    
      Options for default mode:
         -m,--jobmanager <arg>           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.
         -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                         for high availability mode
    
    
    
    Action "stop" stops a running program (streaming jobs only).
    
      Syntax: stop [OPTIONS] <Job ID>
      "stop" action options:
    
      Options for yarn-cluster mode:
         -m,--jobmanager <arg>            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.
         -yid,--yarnapplicationId <arg>   Attach to running YARN session
         -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                          sub-paths for high availability mode
    
      Options for default mode:
         -m,--jobmanager <arg>           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.
         -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                         for high availability mode
    
    
    
    Action "cancel" cancels a running program.
    
      Syntax: cancel [OPTIONS] <Job ID>
      "cancel" action options:
         -s,--withSavepoint <targetDirectory>   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:
         -m,--jobmanager <arg>            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.
         -yid,--yarnapplicationId <arg>   Attach to running YARN session
         -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                          sub-paths for high availability mode
    
      Options for default mode:
         -m,--jobmanager <arg>           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.
         -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                         for high availability mode
    
    
    
    Action "savepoint" triggers savepoints for a running job or disposes existing ones.
    
      Syntax: savepoint [OPTIONS] <Job ID> [<target directory>]
      "savepoint" action options:
         -d,--dispose <arg>       Path of savepoint to dispose.
         -j,--jarfile <jarfile>   Flink program JAR file.
      Options for yarn-cluster mode:
         -m,--jobmanager <arg>            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.
         -yid,--yarnapplicationId <arg>   Attach to running YARN session
         -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                          sub-paths for high availability mode
    
      Options for default mode:
         -m,--jobmanager <arg>           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.
         -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                         for high availability mode
    
    
    
    Action "modify" modifies a running job (e.g. change of parallelism).
    
      Syntax: modify <Job ID> [OPTIONS]
      "modify" action options:
         -h,--help                           Show the help message for the CLI
                                             Frontend or the action.
         -p,--parallelism <newParallelism>   New parallelism for the specified job.
         -v,--verbose                        This option is deprecated.
      Options for yarn-cluster mode:
         -m,--jobmanager <arg>            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.
         -yid,--yarnapplicationId <arg>   Attach to running YARN session
         -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper
                                          sub-paths for high availability mode
    
      Options for default mode:
         -m,--jobmanager <arg>           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.
         -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths
                                         for high availability mode
    
    
    
    Please specify an action.

使用*运行* 算子操作将作业提交给YARN。客户端能够确定JobManager的地址。在极少数问题的情况下，您还可以使用`-m`参数传递JobManager地址。JobManager地址在YARN控制台中可见。

./bin/flink run ./examples/batch/WordCount.jar\
       --input  hdfs://mycluster/user/flink/LICENSE-2.0.txt\
        --output  hdfs://mycluster/user/flink/wordcount-result.txt