Flink搭建
standalone模式是最简单的一种集群模式，不需要yarn、mesos等资源调度平台，这里以1.12.1版本为例进行启动。
下载Flink
- 进入https://www.apache.org/dyn/closer.lua/flink/flink-1.12.1/flink-1.12.1-bin-scala_2.12.tgz，下载即可。
- 对下载下来的tgz包进行解压。
启动集群
-
进入bin目录，执行start-cluster.sh，如下所示：
- 执行jps命令，查看如下进程是否正常启动：
TaskManagerRunner
StandaloneSessionClusterEntrypoint -
浏览器启动localhost:8081，查看概览：
出现此画面，说明Flink启动成功。
提交任务
- 执行一个最简单的WordCount程序，代码如下：
public class WordCountStream {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
// nc -lk 7777
DataStreamSource<String> source = streamExecutionEnvironment.socketTextStream("localhost", 7777);
DataStream<Tuple2<String, Integer>> resultSet = source.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
public void flatMap(String str, Collector<Tuple2<String, Integer>> collector) throws Exception {
String[] sArr = str.split(" ");
for (String s : sArr) {
collector.collect(new Tuple2<String, Integer>(s, 1));
}
}
}).keyBy(0).sum(1);
resultSet.print();
streamExecutionEnvironment.execute();
}
}
maven依赖如下：
<dependencies>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>1.12.1</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>1.12.1</version>
</dependency>
</dependencies>
注意，本地工程引用Flink的jar包开发时，scala版本一定要和集群相对应，如上我们下载的是scala-2.12版本，那么使用java开发时要选用：flink-streaming-java_2.12，版本为1.12.1。
-
对程序进行打包
执行mvn package -DskipTests命令，生成如下jar包：
-
在web ui上对任务进行提交
-
点击jar名称，设置启动选项
点击submit即可提交。
注意
- socketTextStream方法只是connect到远程端口，本身并不会启动端口，因此按照上述流程job会执行失败，此时需要使用nc -l 7777先将对应端口进行启动。
- Flink的stream模式默认需要一次性申请所有的slot进行启动，standalone模式在默认情况下一个TaskManager只有一个slot，因此需要修改Flink的配置。
- 修改slot配置
进入conf目录，对flink-conf.yaml进行编辑，将对应的值修改为5（只要大于2即可，原理后面解释）。
# The number of task slots that each TaskManager offers. Each slot runs one parallel pipeline.
taskmanager.numberOfTaskSlots: 5
- 重新执行上述流程，任务即可正常执行。
