Flink04--double_happy

面试题：谈谈你对Flink ON YARN执行流程的理解

acks=all 数据一定不会丢失吗
如何保证Kafka消费者数据全局有序？ 伪命题
    producer
    storage
    consumer



10分区 3个消费者
    哪个分区被谁消费

C0
C1

T0
    0 1 2
T1
    0 1 2

6/2
We then divide the number of partitions by the total number of
consumers to determine the number of partitions to assign to each consumer. If it does not evenly
divide, then the first few consumers will have one extra partition.




t0p0, t1p0, t1p1, t2p0, t2p1, t2p2

C0 is subscribed to t0;           t0p0
C1 is subscribed to t0, t1;       t0p0, t1p0, t1p1
C2 is subscribed to t0, t1, t2.   t0p0, t1p0, t1p1, t2p0, t2p1, t2p2

Sinl 到 HBase：
	1种.自定义 RichSinkFunction  或者 带并行的
	2种 .  Hadoop Compatibility Beta
			You can:
			use Hadoop’s Writable data types in Flink programs.
			use any Hadoop InputFormat as a DataSource.
			use any Hadoop OutputFormat as a DataSink.
			use a Hadoop Mapper as FlatMapFunction.
			use a Hadoop Reducer as GroupReduceFunction.

也就是说  ：
	写入HBase 有两种方式
		1.自定义 Sink
		2.Flink-Hadoop 的Api

部署：
	1.本地的 
	2.yarn
作为重点。

1.启动集群：

[double_happy@hadoop101 flink]$ bin/start-cluster.sh 
Starting cluster.
Starting standalonesession daemon on host hadoop101.
Starting taskexecutor daemon on host hadoop101.

注意：
	1.Starting standalonesession 
	2.Starting taskexecutor


[double_happy@hadoop101 flink]$ jps
4688 TaskManagerRunner
4722 Jps
4258 StandaloneSessionClusterEntrypoint
[double_happy@hadoop101 flink]$ 


2.看页面 端口 8081

按照官网给的案例：

object SocketWindowWordCount {
  def main(args: Array[String]) : Unit = {
    val port: Int = try {
      ParameterTool.fromArgs(args).getInt("port")
    } catch {
      case e: Exception => {
        System.err.println("No port specified. Please run 'SocketWindowWordCount --host <host> --port <port>'")
        return
      }
    }

    val host: String = try {
      ParameterTool.fromArgs(args).get("host")
    } catch {
      case e: Exception => {
        System.err.println("No port specified. Please run 'SocketWindowWordCount --host <host> --port <port>'")
        return
      }
    }
    

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val text = env.socketTextStream(host, port)
    val windowCounts = text
      .flatMap { w => w.split(",") }
      .map { w => WordWithCount(w, 1) }
      .keyBy("word")
      .sum("count")

    windowCounts.print().setParallelism(1)

    env.execute("Socket Window WordCount")
  }

  case class WordWithCount(word: String, count: Long)

}


注意：
	打包上传到本地

1.启动 端口 nc ：nc -lk 9998

2.提交flink jar 包

[double_happy@hadoop101 bin]$ ./flink
2020-01-05 23:20:00,751 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:20:00,751 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
./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.
     -py,--python <python>                Python script with the program entry
                                          point. The dependent resources can be
                                          configured with the `--pyFiles`
                                          option.
     -pyfs,--pyFiles <pyFiles>            Attach custom python files for job.
                                          Comma can be used as the separator to
                                          specify multiple files. The standard
                                          python resource file suffixes such as
                                          .py/.egg/.zip are all supported.(eg:
                                          --pyFiles
                                          file:///tmp/myresource.zip,hdfs:///$na
                                          menode_address/myresource2.zip)
     -pym,--pyModule <pyModule>           Python module with the program entry
                                          point. This option must be used in
                                          conjunction with `--pyFiles`.
     -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.
     -yat,--yarnapplicationType <arg>     Set a custom application type for the
                                          application on YARN
     -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 with a savepoint (streaming jobs only).

  Syntax: stop [OPTIONS] <Job ID>
  "stop" action options:
     -d,--drain                           Send MAX_WATERMARK before taking the
                                          savepoint and stopping the pipelne.
     -p,--savepointPath <savepointPath>   Path to the savepoint (for example
                                          hdfs:///flink/savepoint-1537). If no
                                          directory is specified, the configured
                                          default will be used
                                          ("state.savepoints.dir").
  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>   **DEPRECATION WARNING**: Cancelling
                                            a job with savepoint is deprecated.
                                            Use "stop" instead.
                                            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



Please specify an action.
[double_happy@hadoop101 bin]$

./flink  run [OPTIONS] <jar-file> <arguments> 

注意：  提交jar  运行 命令

./flink run \
--class com.sx.flink04.SocketWindowWordCount \
/home/double_happy/lib/Flink-1.0.jar \
--port 9998 \
--host hadoop101

提交结果：
[double_happy@hadoop101 bin]$ ./flink run \
> --class com.sx.flink04.SocketWindowWordCount \
> /home/double_happy/lib/Flink-1.0.jar \
> --port 9998 \
> --host hadoop101
Starting execution of program

nc 生产一些数据：

[double_happy@hadoop101 ~]$ nc -lk 9998
a,a,a,a,bb,c


那么这些数据在Flink 哪里呢？

点击 Task Managers： 能看到很多参数的 ***
	1.能看到 jvm的 参数 
	2.Network 相关的

这些日志哪里来的？
	就是从文件系统读进来的

结果：
	1.通过 页面可以看的到的 （就是从本地结果 传到页面上的 ）
	2.通过本地也能看到

本地查看：
	[double_happy@hadoop101 log]$ tail -200f flink-double_happy-taskexecutor-1-hadoop101.out
	WordWithCount(a,1)
	WordWithCount(a,2)
	WordWithCount(a,3)
	WordWithCount(a,4)
	WordWithCount(bb,1)
	WordWithCount(c,1)


注意：
	$ tail -f log/flink-*-taskexecutor-*.out  
就可以看到本地 结果

如何停止这个作业：
	1.页面 上 running jos 点进去 
	2.本地命令

本地命令：
	 ./flink  cancel [OPTIONS] <Job ID>

[double_happy@hadoop101 bin]$ ./flink cancel 710a899ba9589d6a688c14931f7ef6cd
Cancelling job 710a899ba9589d6a688c14931f7ef6cd.
Cancelled job 710a899ba9589d6a688c14931f7ef6cd.
[double_happy@hadoop101 bin]$

关闭集群：


[double_happy@hadoop101 bin]$ jps
4688 TaskManagerRunner
4258 StandaloneSessionClusterEntrypoint
9444 Jps

[double_happy@hadoop101 bin]$ ./stop-cluster.sh 
Stopping taskexecutor daemon (pid: 4688) on host hadoop101.
Stopping standalonesession daemon (pid: 4258) on host hadoop101.

[double_happy@hadoop101 bin]$ jps
10457 Jps
[double_happy@hadoop101 bin]$

注意：
	页面也就打不开了

以上：
	就是最简单的local模式 运行jar  job

[double_happy@hadoop101 bin]$ ./start-scala-shell.sh 
Starting Flink Shell:
log4j:WARN No appenders could be found for logger (org.apache.flink.configuration.GlobalConfiguration).
log4j:WARN Please initialize the log4j system properly.
log4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.
Error: please specify execution mode:
[local | remote <host> <port> | yarn]
[double_happy@hadoop101 bin]$ 

注意：
	Error: please specify execution mode: 指定跑在什么模式下 
	Spark和Flink都可以跑在多个模式上的

[double_happy@hadoop101 bin]$ ./start-scala-shell.sh local
Starting Flink Shell:

Starting local Flink cluster (host: localhost, port: 8081).


Connecting to Flink cluster (host: localhost, port: 8081).


                         ▒▓██▓██▒
                     ▓████▒▒█▓▒▓███▓▒
                  ▓███▓░░        ▒▒▒▓██▒  ▒
                ░██▒   ▒▒▓▓█▓▓▒░      ▒████
                ██▒         ░▒▓███▒    ▒█▒█▒
                  ░▓█            ███   ▓░▒██
                    ▓█       ▒▒▒▒▒▓██▓░▒░▓▓█
                  █░ █   ▒▒░       ███▓▓█ ▒█▒▒▒
                  ████░   ▒▓█▓      ██▒▒▒ ▓███▒
               ░▒█▓▓██       ▓█▒    ▓█▒▓██▓ ░█░
         ▓░▒▓████▒ ██         ▒█    █▓░▒█▒░▒█▒
        ███▓░██▓  ▓█           █   █▓ ▒▓█▓▓█▒
      ░██▓  ░█░            █  █▒ ▒█████▓▒ ██▓░▒
     ███░ ░ █░          ▓ ░█ █████▒░░    ░█░▓  ▓░
    ██▓█ ▒▒▓▒          ▓███████▓░       ▒█▒ ▒▓ ▓██▓
 ▒██▓ ▓█ █▓█       ░▒█████▓▓▒░         ██▒▒  █ ▒  ▓█▒
 ▓█▓  ▓█ ██▓ ░▓▓▓▓▓▓▓▒              ▒██▓           ░█▒
 ▓█    █ ▓███▓▒░              ░▓▓▓███▓          ░▒░ ▓█
 ██▓    ██▒    ░▒▓▓███▓▓▓▓▓██████▓▒            ▓███  █
▓███▒ ███   ░▓▓▒░░   ░▓████▓░                  ░▒▓▒  █▓
█▓▒▒▓▓██  ░▒▒░░░▒▒▒▒▓██▓░                            █▓
██ ▓░▒█   ▓▓▓▓▒░░  ▒█▓       ▒▓▓██▓    ▓▒          ▒▒▓
▓█▓ ▓▒█  █▓░  ░▒▓▓██▒            ░▓█▒   ▒▒▒░▒▒▓█████▒
 ██░ ▓█▒█▒  ▒▓▓▒  ▓█                █░      ░░░░   ░█▒
 ▓█   ▒█▓   ░     █░                ▒█              █▓
  █▓   ██         █░                 ▓▓        ▒█▓▓▓▒█░
   █▓ ░▓██░       ▓▒                  ▓█▓▒░░░▒▓█░    ▒█
    ██   ▓█▓░      ▒                    ░▒█▒██▒      ▓▓
     ▓█▒   ▒█▓▒░                         ▒▒ █▒█▓▒▒░░▒██
      ░██▒    ▒▓▓▒                     ▓██▓▒█▒ ░▓▓▓▓▒█▓
        ░▓██▒                          ▓░  ▒█▓█  ░░▒▒▒
            ▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓  ▓░▒█░

              F L I N K - S C A L A - S H E L L

NOTE: Use the prebound Execution Environments and Table Environment to implement batch or streaming programs.

  Batch - Use the 'benv' and 'btenv' variable

    * val dataSet = benv.readTextFile("/path/to/data")
    * dataSet.writeAsText("/path/to/output")
    * benv.execute("My batch program")
    *
    * val batchTable = btenv.fromDataSet(dataSet)
    * btenv.registerTable("tableName", batchTable)
    * val result = btenv.sqlQuery("SELECT * FROM tableName").collect
    HINT: You can use print() on a DataSet to print the contents or collect()
    a sql query result back to the shell.

  Streaming - Use the 'senv' and 'stenv' variable

    * val dataStream = senv.fromElements(1, 2, 3, 4)
    * dataStream.countWindowAll(2).sum(0).print()
    *
    * val streamTable = stenv.fromDataStream(dataStream, 'num)
    * val resultTable = streamTable.select('num).where('num % 2 === 1 )
    * resultTable.toAppendStream[Row].print()
    * senv.execute("My streaming program")
    HINT: You can only print a DataStream to the shell in local mode.
      

scala>

注意：
	  Batch - Use the 'benv' and 'btenv' variable
	  Streaming - Use the 'senv' and 'stenv' variable

批处理：
scala> val text = benv.fromElements("kairis,kairis,kairis","double_happy,sxwang")
text: org.apache.flink.api.scala.DataSet[String] = org.apache.flink.api.scala.DataSet@26a004ed

scala> text.flatMap(_.split(",")).map((_,1)).groupBy(0).sum(1).print
(double_happy,1)
(kairis,3)
(sxwang,1)

scala>

面试题：谈谈你对Flink ON YARN执行流程的理解？

	和Spark mr 都差不多 只是 ApplicaitonMaster 名字变了

The YARN client needs to access the Hadoop configuration to connect to the YARN resource manager and HDFS

注意：
	YARN client 要配置gateway

但是：
	Flink跑yarn 跟Spark 跑yarn 有点区别

为什么呢？
	Flink有两种模式跑yarn
		1.Start a long-running Flink cluster on YARN
			启动一个长服务
		2.Run a Flink job on YARN

Spark里面 就是 右边这种  

注意：
	Spark申请资源是耗时的 还记得吗 依赖的 jar包需要传到hdfs上 也不然每次都要传 还记得么

这两种方式会带来什么好处：
	1.长服务 
			省资源申请的时间

 ./yarn-session.sh  运行的时候有坑的 ：
 	1.你自己源码编译好的是没有问题的
 	2.不是编译的 要在flink lib包下面 加一个 lib哦


[double_happy@hadoop101 bin]$ ./yarn-session.sh --help
2020-01-05 22:32:14,068 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, hadoop101
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2020-01-05 22:32:14,070 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
2020-01-05 22:32:14,071 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.execution.failover-strategy, region
2020-01-05 22:32:14,071 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: rest.port, 8081
2020-01-05 22:32:14,667 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2020-01-05 22:32:14,923 INFO  org.apache.flink.runtime.security.modules.HadoopModule        - Hadoop user set to double_happy (auth:SIMPLE)
Usage:
   Required
     -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers)
   Optional
     -at,--applicationType <arg>     Set a custom application type for the application on YARN
     -D <property=value>             use value for given property
     -d,--detached                   If present, runs the job in detached mode
     -h,--help                       Help for the Yarn session CLI.
     -id,--applicationId <arg>       Attach to running YARN session
     -j,--jar <arg>                  Path to Flink jar file
     -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)
     -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.
     -n,--container <arg>            Number of YARN container to allocate (=Number of Task Managers)
     -nl,--nodeLabel <arg>           Specify YARN node label for the YARN application
     -nm,--name <arg>                Set a custom name for the application on YARN
     -q,--query                      Display available YARN resources (memory, cores)
     -qu,--queue <arg>               Specify YARN queue.
     -s,--slots <arg>                Number of slots per TaskManager
     -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.
     -st,--streaming                 Start Flink in streaming mode
     -t,--ship <arg>                 Ship files in the specified directory (t for transfer)
     -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)
     -yd,--yarndetached              If present, runs the job in detached mode (deprecated; use non-YARN specific option instead)
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths for high availability mode
[double_happy@hadoop101 bin]$

  Required
     -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers)

注意：
	必填的 
	-n   就是 container   =  Number of Task Managers 

不是必填的：
  -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)
 -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)
    -d,--detached                   If present, runs the job in detached mode     运行你的job 以什么方式

运行起来：

[double_happy@hadoop101 bin]$ ./yarn-session.sh -n 2 -jm 1024 -tm 1024 -d 

2020-01-05 22:37:43,651 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.address, hadoop101
2020-01-05 22:37:43,656 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.rpc.port, 6123
2020-01-05 22:37:43,656 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.heap.size, 1024m
2020-01-05 22:37:43,657 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.heap.size, 1024m
2020-01-05 22:37:43,657 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: taskmanager.numberOfTaskSlots, 1
2020-01-05 22:37:43,657 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: parallelism.default, 1
2020-01-05 22:37:43,658 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: jobmanager.execution.failover-strategy, region
2020-01-05 22:37:43,658 INFO  org.apache.flink.configuration.GlobalConfiguration            - Loading configuration property: rest.port, 8081
2020-01-05 22:37:44,328 WARN  org.apache.hadoop.util.NativeCodeLoader                       - Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
2020-01-05 22:37:44,505 INFO  org.apache.flink.runtime.security.modules.HadoopModule        - Hadoop user set to double_happy (auth:SIMPLE)
2020-01-05 22:37:44,611 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 22:37:44,761 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The argument n is deprecated in will be ignored.
2020-01-05 22:37:44,930 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=2, slotsPerTaskManager=1}
2020-01-05 22:37:45,573 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/double_happy/app/flink-1.9.1/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
2020-01-05 22:37:47,853 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1578233120766_0002
2020-01-05 22:37:48,255 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1578233120766_0002
2020-01-05 22:37:48,255 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
2020-01-05 22:37:48,257 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
2020-01-05 22:37:58,915 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
2020-01-05 22:37:58,915 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The Flink YARN client has been started in detached mode. In order to stop Flink on YARN, use the following command or a YARN web interface to stop it:
yarn application -kill application_1578233120766_0002
Please also note that the temporary files of the YARN session in the home directory will not be removed.
2020-01-05 22:37:59,734 INFO  org.apache.flink.runtime.rest.RestClient                      - Rest client endpoint started.
Flink JobManager is now running on hadoop101:8081 with leader id 00000000-0000-0000-0000-000000000000.
JobManager Web Interface: http://hadoop101:8081
2020-01-05 22:37:59,757 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - The Flink YARN client has been started in detached mode. In order to stop Flink on YARN, use the following command or a YARN web interface to stop it:
yarn application -kill application_1578233120766_0002
[double_happy@hadoop101 bin]$

运行一个 例子试试：
	Example

wget -O LICENSE-2.0.txt http://www.apache.org/licenses/LICENSE-2.0.txt
hadoop fs -copyFromLocal LICENSE-2.0.txt hdfs:/// ...
./bin/flink run ./examples/batch/WordCount.jar \
       --input hdfs:///..../LICENSE-2.0.txt --output hdfs:///.../wordcount-result.txt


注意：
	这是啥意思呀：
		就是把数据 上传到hdfs上 然后运行

[double_happy@hadoop101 data]$ hadoop fs -put ./LICENSE-2.0.txt /flink/input/
put: `/flink/input/': No such file or directory

[double_happy@hadoop101 data]$ hadoop fs -mkdir -p  /flink/input/

[double_happy@hadoop101 data]$ hadoop fs -put ./LICENSE-2.0.txt /flink/input/    

[double_happy@hadoop101 data]$ hadoop fs -ls /flink/input/
Found 1 items
-rw-r--r--   1 double_happy supergroup      11358 2020-01-05 22:43 /flink/input/LICENSE-2.0.txt
[double_happy@hadoop101 data]$

运行：

[double_happy@hadoop101 flink]$ ./bin/flink run ./examples/batch/WordCount.jar \
> --input hdfs://hadoop101:8020/flink/input/LICENSE-2.0.txt \
> --output hdfs://hadoop101:8020/flink/output
2020-01-05 22:46:46,517 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 22:46:46,517 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 22:46:47,119 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - YARN properties set default parallelism to 2
2020-01-05 22:46:47,119 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - YARN properties set default parallelism to 2
YARN properties set default parallelism to 2
2020-01-05 22:46:47,238 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 22:46:47,449 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 22:46:47,449 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 22:46:47,650 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Found application JobManager host name 'hadoop101' and port '8081' from supplied application id 'application_1578233120766_0002'
Starting execution of program

结果：
[double_happy@hadoop101 flink]$ hadoop fs -ls /flink/output
-rw-r--r--   1 double_happy supergroup       4499 2020-01-05 22:47 /flink/output
[double_happy@hadoop101 flink]$ hadoop fs -text /flink/output
0 3
1 2
2 4
2004 1
3 1
4 1
5 1
50 1
6 1
7 1
8 1
9 2
a 22
above 1
acceptance 1
accepting 3
act 1
acting 1
acts 1
add 2
addendum 1
additional 5
additions 1


注意：
	Flink输出的 是文件 
	Spark是文件夹

注意：
	这个模式 他怎么知道是提交到 哪个session上去呢？

看控制台log：
2020-01-05 22:46:46,517 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli              
   - Found Yarn properties file under /tmp/.yarn-properties-double_happy.

打开： /tmp/.yarn-properties-double_happy


[double_happy@hadoop101 flink]$ cat /tmp/.yarn-properties-double_happy
#Generated YARN properties file
#Sun Jan 05 22:37:59 CST 2020
parallelism=2
dynamicPropertiesString=
applicationID=application_1578233120766_0002     //yarn 的 application的 id
[double_happy@hadoop101 flink]$


注意：
	这块 如果你已经使用了 这个模式跑 之后 还想使用 local跑 
	需要把/tmp/.yarn-properties-double_happy  这个 干掉 也不然 本地模式会 跑到yarn上 报错

官方案例：

[double_happy@hadoop101 flink]$ ./bin/flink run -m yarn-cluster ./examples/batch/WordCount.jar
2020-01-05 23:06:06,559 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:06:06,559 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:06:07,151 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 23:06:07,324 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:06:07,324 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:06:07,520 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=1}
2020-01-05 23:06:08,127 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/double_happy/app/flink-1.9.1/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
2020-01-05 23:06:10,927 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1578233120766_0003
2020-01-05 23:06:10,965 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1578233120766_0003
2020-01-05 23:06:10,966 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
2020-01-05 23:06:10,971 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
2020-01-05 23:06:18,402 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
Starting execution of program
Executing WordCount example with default input data set.
Use --input to specify file input.
Printing result to stdout. Use --output to specify output path.
(a,5)
(action,1)
(after,1)
(against,1)
(all,2)
(and,12)
(arms,1)
(arrows,1)
(awry,1)
(ay,1)
(bare,1)
(be,4)
(bear,3)
(bodkin,1)
(bourn,1)
(but,1)
(by,2)
(calamity,1)
(cast,1)
(coil,1)
(come,1)
(conscience,1)
(consummation,1)
(contumely,1)
(country,1)
(cowards,1)
(currents,1)
(d,4)
(death,2)
(delay,1)
(despis,1)
(devoutly,1)
(die,2)
(does,1)
(dread,1)
(dream,1)
(dreams,1)
(end,2)
(enterprises,1)
(er,1)
(fair,1)
(fardels,1)
(flesh,1)
(fly,1)
(for,2)
(fortune,1)
(from,1)
(give,1)
(great,1)
(grunt,1)
(have,2)
(he,1)
(heartache,1)
(heir,1)
(himself,1)
(his,1)
(hue,1)
(ills,1)
(in,3)
(insolence,1)
(is,3)
(know,1)
(law,1)
(life,2)
(long,1)
(lose,1)
(love,1)
(make,2)
(makes,2)
(man,1)
(may,1)
(merit,1)
(might,1)
(mind,1)
(moment,1)
(more,1)
(mortal,1)
(must,1)
(my,1)
(name,1)
(native,1)
(natural,1)
(no,2)
(nobler,1)
(not,2)
(now,1)
(nymph,1)
(o,1)
(of,15)
(off,1)
(office,1)
(ophelia,1)
(opposing,1)
(oppressor,1)
(or,2)
(orisons,1)
(others,1)
(outrageous,1)
(pale,1)
(pangs,1)
(patient,1)
(pause,1)
(perchance,1)
(pith,1)
(proud,1)
(puzzles,1)
(question,1)
(quietus,1)
(rather,1)
(regard,1)
(remember,1)
(resolution,1)
(respect,1)
(returns,1)
(rub,1)
(s,5)
(say,1)
(scorns,1)
(sea,1)
(shocks,1)
(shuffled,1)
(sicklied,1)
(sins,1)
(sleep,5)
(slings,1)
(so,1)
(soft,1)
(something,1)
(spurns,1)
(suffer,1)
(sweat,1)
(take,1)
(takes,1)
(than,1)
(that,7)
(the,22)
(their,1)
(them,1)
(there,2)
(these,1)
(this,2)
(those,1)
(thought,1)
(thousand,1)
(thus,2)
(thy,1)
(time,1)
(tis,2)
(to,15)
(traveller,1)
(troubles,1)
(turn,1)
(under,1)
(undiscover,1)
(unworthy,1)
(us,3)
(we,4)
(weary,1)
(what,1)
(when,2)
(whether,1)
(whips,1)
(who,2)
(whose,1)
(will,1)
(wish,1)
(with,3)
(would,2)
(wrong,1)
(you,1)
Program execution finished
Job with JobID 0af373b2d338811166328e7e1fca1fe1 has finished.
Job Runtime: 10995 ms
Accumulator Results: 
- fed80ecbe6107c17759a1be247abbdd4 (java.util.ArrayList) [170 elements]

[double_happy@hadoop101 flink]$

运行一下 之前的local的 程序：
[double_happy@hadoop101 flink]$ ./bin/flink run \
> --class com.sx.flink04.SocketWindowWordCount \
> -m yarn-cluster \
> /home/double_happy/lib/Flink-1.0.jar \
> --port 9998 \
> --host hadoop101
2020-01-05 23:08:37,619 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:08:37,619 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - Found Yarn properties file under /tmp/.yarn-properties-double_happy.
2020-01-05 23:08:38,076 INFO  org.apache.hadoop.yarn.client.RMProxy                         - Connecting to ResourceManager at /0.0.0.0:8032
2020-01-05 23:08:38,294 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:08:38,294 INFO  org.apache.flink.yarn.cli.FlinkYarnSessionCli                 - No path for the flink jar passed. Using the location of class org.apache.flink.yarn.YarnClusterDescriptor to locate the jar
2020-01-05 23:08:38,467 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Cluster specification: ClusterSpecification{masterMemoryMB=1024, taskManagerMemoryMB=1024, numberTaskManagers=1, slotsPerTaskManager=1}
2020-01-05 23:08:39,079 WARN  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - The configuration directory ('/home/double_happy/app/flink-1.9.1/conf') contains both LOG4J and Logback configuration files. Please delete or rename one of them.
2020-01-05 23:08:40,769 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Submitting application master application_1578233120766_0004
2020-01-05 23:08:41,006 INFO  org.apache.hadoop.yarn.client.api.impl.YarnClientImpl         - Submitted application application_1578233120766_0004
2020-01-05 23:08:41,007 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Waiting for the cluster to be allocated
2020-01-05 23:08:41,010 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - Deploying cluster, current state ACCEPTED
2020-01-05 23:08:46,914 INFO  org.apache.flink.yarn.AbstractYarnClusterDescriptor           - YARN application has been deployed successfully.
Starting execution of program

输入数据：
[double_happy@hadoop101 data]$ nc -lk 9998
啊，a,a,a,v,b,c
c,c,d,d,,b,b


查看结果：

这是yarn的两种方式：

我使用的是第二种方式 ：
	如果需要一个MySQL jar包该怎么传？
		我接触flink 时间很短  我身边同事也不知道 他们经常搞Flink的 竟然不知道？ 就知道打胖包？ 绝对是不可以的