1 安装基于CDH6.2.0环境.
hadoop版本3.0
spark版本2.4， 自带scala版本2.11.12
2 下载flink1.9 http://mirrors.tuna.tsinghua.edu.cn/apache/flink/flink-1.9.0/flink-1.9.0-bin-scala_2.11.tgz
3解压 tar -vxf flink-1.9.0-bin-scala_2.11.tgz， 到/myflink/flink1.9.0
4 每个节点修改
4.1修改conf/flink-conf.yaml
设置jobmanager.rpc.address 配置项为该master节点的IP 或者主机名
4.2 修改masters
cdh00
4.3修改slaves
cdh01
cdh02
cdh03
5 master节点启动： bin/start-cluster.sh
6 常用命令
6.1 起服务
./flink run /root/helloworld.jar
6.2、取消任务
./flink cancel jobId
6.3、停止集群服务
./stop_cluster.sh
7 Flink示例代码测试（单词计数）。代码使用 idea打包。
7.1 数组输入测试。任务执行可以在任一个节点
提交命令： ./flink run -c WordCount /home/hadoop/flinktest/FlinkStream_-1.0-SNAPSHOT.jar
jar包如果没指定入口类(manifest声明)，需要通过-c指定入口类。
7.2 本地文件输入测试：
./flink run -c WordCount /home/hadoop/flinktest/FlinkStream_-1.0-SNAPSHOT.jar --input /home/hadoop/flinktest/inputwords.txt --output /home/hadoop/flinktest/output
或者：
./flink run -c WordCount /home/hadoop/flinktest/FlinkStream_-1.0-SNAPSHOT.jar --input file:///home/hadoop/flinktest/inputwords.txt --output file:///home/hadoop/flinktest/output
因为任务执行可以在任一个节点，需要保证每个节点都存在输入文件。
7.3 HDFS文件输入测试：
./flink run -c WordCount /home/hadoop/flinktest/FlinkStream_-1.0-SNAPSHOT.jar --input hdfs://cdh00:8020/flinktest/inputwords.txt --output hdfs:///home/hadoop/flinktest/output 或者
./flink run -c WordCount /home/hadoop/flinktest/FlinkStream_-1.0-SNAPSHOT.jar --input hdfs:///flinktest/inputwords.txt --output hdfs:///home/hadoop/flinktest/output
注意：由于flink默认没有绑定hadoop相关的类文件，刚开始测试的时候总是报错：
Caused by: org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 'hdfs'.
解决方案： 设置环境变量：
export HADOOP_CLASSPATH=`hadoop classpath`重启Flink即可。
网上其它说的在flink的lib下添加相关的jar包: flink-shaded-hadoop2.jar包的方案未解决问题，或许CDH版本使用的是hadoop3.0， jar包版本不对吧。 不过这种方案肯定不如上述方案完美。
7.4 kafka连接测试：
kafka connector导入，如下。
注： 不要使用flink-connector-0.10-kafka_2.11，此连接器是为1.0版本及以下的kafka使用。
使用时会报错：Caused by: java.lang.NoSuchMethodError: org.apache.kafka.clients.consumer.KafkaConsumer.assign
8 日志查看： standalone日志查看， 到task manager所在节点的日志路径查看out 日志即可。
日志路径。在cdh01节点，
启动日志 /myflink/flink-1.9.0/log/ **********.log
程序输出日志： /myflink/flink-1.9.0/log/ **********.out
9 standalone存在的问题
9.1）job subtask分布使用不均衡的问题
早些Flink版本standalone cluster在发布具体的job时，会有一定的随机性。举个例子，如果当前集群总共有2台8核的机器用以部署TaskManager，每台机器上一个TaskManager实例，每个TaskManager的TaskSlot为8，而我们的job的并行度为12，那么就有可能会出现下图的现象：
第一个TaskManager的slot全被占满，而第二个TaskManager只使用了一半的资源！资源严重不平衡，随着job处理的流量加大，一定会造成TM1上的task消费速度慢，而TM2上的task消费速度远高于TM1的task的情况。假设业务量的增长迫使我们不得不扩大job的并行度为24，并且扩容2台性能更高的机器（12核），在新的机器上，我们分别部署slot数为12的TaskManager。经过扩容后，新扩容的配置高的机器并没有去承担更多的Task，老机器的负担仍然比较严重，资源本质上还是不均匀！
9.2） 资源隔离差的问题。
在一个cluster中往往会部署不止一个job，而这些job在每台机器上都共用JVM，会造成资源的竞争。
起初，我们为了解决这些问题，采用了如下的解决方法:
- 将TaskManager的粒度变小，即一台机器部署多个实例，每个实例持有的slot数较少；
- 将大的业务job隔离到不同的集群上。
这些解决方法增加了实例数和集群数，进而增加了维护成本。因此我们决定要迁移到on yarn上，目前看Flink on yarn的资源分配和资源隔离确实比standalone模式要优秀一些。
10 standalone模式tm数量按机器slave节点数来计算， slot总数= tm(节点数)* 每tm槽位数
yarn模式下，tm数=max(子任务并行书)/每tm槽位数
参考：
