Flink学习笔记：3、Flink分布式模式（Standalone）
Flink学习笔记：3、Flink分布式模式（Standalone）
在https://cloud.tencent.com/developer/article/1011015，我们已经在node1节点上下载了Flink软件包。
3.1 配置文件
3.1.1 flink-conf.yaml
参数 | 默认值 | 修改值 | 说明 |
|---|---|---|---|
jobmanager.rpc.address | localhost | | 配置JobManager进行RPC通信的地址 |
jobmanager.rpc.port | 6123 | | 配置JobManager进行RPC通信的端口 |
taskmanager.numberOfTaskSlots | 1 | 2 | 配置每一个slave节点上task的数目 |
taskmanager.memory.preallocate | false | | 配置是否在Flink集群启动时候给TaskManager分配内存，默认不进行预分配，这样在我们不使用flink集群时候不会占用集群资源 |
parallelism.default | 1 | 2 | 用于未指定的程序的并行性和其他并行性 |
jobmanager.web.port | 8081 | | 指定JobManger的可视化端口，尽量配置一个不容易冲突的端口 |
编辑conf/flink-conf.yaml配置文件
[root@node1 flink-1.3.2]# vi conf/flink-conf.yaml 内容如下
jobmanager.rpc.address: node1
jobmanager.rpc.port: 6123
jobmanager.heap.mb: 1024
taskmanager.heap.mb: 1024
taskmanager.numberOfTaskSlots: 2
taskmanager.memory.preallocate: false
parallelism.default: 2
jobmanager.web.port: 8081注意：flink-conf.yaml中配置key/value时候在“:”后面需要有一个空格，否则配置不会生效。
3.1.2 slaves
将所有的 worker 节点 （TaskManager）的IP或者主机名（一行一个）填入conf/slaves 文件中。
此处建议写入主机名，与IP地址解耦。
[root@node1 flink-1.3.2]# vi conf/slaves
[root@node1 flink-1.3.2]# cat conf/slaves
node1
node2
node3
[root@node1 flink-1.3.2]#3.2 分布软件包
[root@node1 flink-1.3.2]# scp -r /opt/flink-1.3.2/ node2:/opt
[root@node1 flink-1.3.2]# scp -r /opt/flink-1.3.2/ node3:/opt3.3 启动Flink集群
[root@node1 flink-1.3.2]# bin/start-cluster.sh
Starting cluster.
Starting jobmanager daemon on host node1.
Starting taskmanager daemon on host node1.
Starting taskmanager daemon on host node2.
Starting taskmanager daemon on host node3.
[root@node1 flink-1.3.2]# jps3.4 查看Flink进程
[root@node1 flink-1.3.2]# jps
4518 TaskManager
4583 Jps
4170 JobManager
[root@node1 flink-1.3.2]#[root@node2 ~]# jps
3136 TaskManager
3167 Jps
[root@node2 ~]#[root@node3 ~]# jps
3411 Jps
3389 TaskManager
[root@node3 ~]#3.5 WebUI
If all the configurations are good, then you would see that the cluster is up and running. You can
check the web UI at http://<job-manager-ip>:8081/ .
The following are some snapshots of the Flink Web UI:
以下是Flink Web UI的一些快照：
直接打开JobManager所在节点：http://192.168.80.131:8081
You can click on the Job Manager link to get the following view:
您可以单击“作业管理器”链接以获取以下视图：
Similarly, you can check out the Task Managers view as follows:
同样，您可以按如下方式签出任务管理器视图：
3.6 Shell
最后我们可以启动一个shell连接到集群上运行一个job试一试了，可以使用start-scala-shell.sh启动shell控制台进行写程序，但是如果不跟参数的话，则启动的是一个本地的shell。所以我们需要输入我们的 Master URL。
[root@node1 flink-1.3.2]# bin/start-scala-shell.sh remote node1 6123
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.
Connecting to Flink cluster (host: node1, port: 6123).
▒▓██▓██▒
▓████▒▒█▓▒▓███▓▒
▓███▓░░ ▒▒▒▓██▒ ▒
░██▒ ▒▒▓▓█▓▓▒░ ▒████
██▒ ░▒▓███▒ ▒█▒█▒
░▓█ ███ ▓░▒██
▓█ ▒▒▒▒▒▓██▓░▒░▓▓█
█░ █ ▒▒░ ███▓▓█ ▒█▒▒▒
████░ ▒▓█▓ ██▒▒▒ ▓███▒
░▒█▓▓██ ▓█▒ ▓█▒▓██▓ ░█░
▓░▒▓████▒ ██ ▒█ █▓░▒█▒░▒█▒
███▓░██▓ ▓█ █ █▓ ▒▓█▓▓█▒
░██▓ ░█░ █ █▒ ▒█████▓▒ ██▓░▒
███░ ░ █░ ▓ ░█ █████▒░░ ░█░▓ ▓░
██▓█ ▒▒▓▒ ▓███████▓░ ▒█▒ ▒▓ ▓██▓
▒██▓ ▓█ █▓█ ░▒█████▓▓▒░ ██▒▒ █ ▒ ▓█▒
▓█▓ ▓█ ██▓ ░▓▓▓▓▓▓▓▒ ▒██▓ ░█▒
▓█ █ ▓███▓▒░ ░▓▓▓███▓ ░▒░ ▓█
██▓ ██▒ ░▒▓▓███▓▓▓▓▓██████▓▒ ▓███ █
▓███▒ ███ ░▓▓▒░░ ░▓████▓░ ░▒▓▒ █▓
█▓▒▒▓▓██ ░▒▒░░░▒▒▒▒▓██▓░ █▓
██ ▓░▒█ ▓▓▓▓▒░░ ▒█▓ ▒▓▓██▓ ▓▒ ▒▒▓
▓█▓ ▓▒█ █▓░ ░▒▓▓██▒ ░▓█▒ ▒▒▒░▒▒▓█████▒
██░ ▓█▒█▒ ▒▓▓▒ ▓█ █░ ░░░░ ░█▒
▓█ ▒█▓ ░ █░ ▒█ █▓
█▓ ██ █░ ▓▓ ▒█▓▓▓▒█░
█▓ ░▓██░ ▓▒ ▓█▓▒░░░▒▓█░ ▒█
██ ▓█▓░ ▒ ░▒█▒██▒ ▓▓
▓█▒ ▒█▓▒░ ▒▒ █▒█▓▒▒░░▒██
░██▒ ▒▓▓▒ ▓██▓▒█▒ ░▓▓▓▓▒█▓
░▓██▒ ▓░ ▒█▓█ ░░▒▒▒
▒▓▓▓▓▓▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒▒░░▓▓ ▓░▒█░
F L I N K - S C A L A - S H E L L
NOTE: Use the prebound Execution Environments to implement batch or streaming programs.
Batch - Use the 'benv' variable
* val dataSet = benv.readTextFile("/path/to/data")
* dataSet.writeAsText("/path/to/output")
* benv.execute("My batch program")
HINT: You can use print() on a DataSet to print the contents to the shell.
Streaming - Use the 'senv' variable
* val dataStream = senv.fromElements(1, 2, 3, 4)
* dataStream.countWindowAll(2).sum(0).print()
* senv.execute("My streaming program")
HINT: You can only print a DataStream to the shell in local mode.
Scala-Flink>