由于最近公司需要进行流式计算方面的研究，派给我研究flink框架的任务，由于之前完全没有接触过流式计算的相关内容，在接到任务后还是有点捉急的，下面将我自己在入门使用的一些心得体会记录下，以便如果有需要的新同学共同的学习。下面我将重点介绍下flink如何安装配置：

1：环境准备

单机版：

   操作系统：Macos

 java版本：1.8

集群版：

   操作系统：Ubuntu14

   Hadoop：2.7.3

java：1.8

  flink版本：Apache Flink 1.8.1 for Scala 2.12 (asc, sha512)（ 下载地址：下载地址）

注：1.8版本的flink的Hadoop作为一个组件分离出来了，所以如果需要使用yarn集群启动的话，还需要下载对应Hadoop版本的bin包。下载页面都有提供。

当下载好了之后，我们就可以对其进行配置了。我这里主要说明下单机版的，伪分布式具体操作流程也和这个类似，网上的教程也很多。

1，首先加压下载包：

2，进入解压目录后：

执行： ./bin/start-cluster.sh

可以看到启动输出，提示是后台启动，然后我们通过jps命令可以看到服务进程；

启动进程分别是：

TaskManagerRunner的任务进程和StandaloneSessionClusterEntrypoint 的jobManager

任务启动后，我们也可以通过web UI进行访问，webUI的默认访问端口是8081。

可以通过localhost：8081

至此我们单机版的服务就启动了，你就可以进行任务的提交啦！

对于集群版本：

集群主要有两种启动模式，一种是standalone，一种是yarn方式。

假设我们有三台机器，三台机器的机器名分别是master ，slave01  ，slave02。

然后我们三台机器需要设置免密登录。（不会设置的同学可自行Google Ubuntu 如何设置免密登录，很简单的）

flink也是采用的主从模式，配置信息也和Hadoop类似，我们主要修改的文件有以下三个：

1，flink的启动配置文件：flink-conf.yaml ，此文件中我们最主要修改的参数是主机地址，也就是你选择的集群的JobManager的机器。

jobmanager.rpc.address: master（注意冒号后面需要有空格配置才会生效，不然不会生效）

其他的都可以采用默认值，这里提一下，因为我使用的机器8081端口被人占用了，所以要修改我的webUI端口，

rest.port: 9091 （可以通过设置rest.port端口来修改我们的web UI访问端口）；

2，master文件

在master文件中我们配置好我们的主机就可以

master:8081 

或者如果想配置多台主机，每行一台即可:

master01:8081

master02:8081

说明：前面是主机名，后面是web ui的端口号，如果要要修改web ui的端口，我们此处的端口号应该和flink-conf.yaml文件中的rest.port值一致。

3，slaves文件，该文件主要是配置我们的taskManager节点。

slave01

slave02

通过配置我们的配置文件就完成了，然后每台机器的配置需要一样，我们可以通过scp将三个配置都复制三台机器，还需要注意的是我们flink的在每台机器的目录应该也是一样的。

当配置完成后，我们可以通过start-cluster.sh命令来启动我们的服务，此时启动的就是standalone模式，我们可以在通过：master:8081访问我们的服务。

我们也可以通过yarn集群来启动我们的服务，在flink里面提供了一个命令：

yarn-session.sh 来启动。  

常用的参数有：

-d  表示是否后台启动

-n 启动的任务数量

-s  每个任务的slot数量

-jm 每个jobManager的内存

-tm 每个taskManager的内容

例： yarn-session.sh -d -n 2 -s 4 -jm 1024 -tm 1024

注意在yarn模式下，我们需要先启动yarn服务，并且1.8需要下载对应的Hadoop bin文件 ，下载的文件需要放到flink的lib目录下。

至此，flink的相关启动我们就完成了，当然这是是最简单的启动，flink的还有很多其他的高级内容，如HA的配置等，我们在之后会进行分享。

下面是我在服务启动中的一些问题：

在伪分布式用standalone模式启动的时候，我的slave机器的taskManager 任务一直无法启动，并且日志总是报：

Exception in thread "main" java.lang.UnsupportedClassVersionError: words/Thesaurus : Unsupported major.minor version 51.0  at java.lang.ClassLoader.defineClass1(Native Method)  at java.lang.ClassLoader.defineClassCond(ClassLoader.java:631)  at java.lang.ClassLoader.defineClass(ClassLoader.java:615)  at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:141)  at java.net.URLClassLoader.defineClass(URLClassLoader.java:283)  at java.net.URLClassLoader.access$000(URLClassLoader.java:58)  at java.net.URLClassLoader$1.run(URLClassLoader.java:197)  at java.security.AccessController.doPrivileged(Native Method)  at java.net.URLClassLoader.findClass(URLClassLoader.java:190)  at java.lang.ClassLoader.loadClass(ClassLoader.java:306)  at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:301)  at java.lang.ClassLoader.loadClass(ClassLoader.java:247)  Could not find the main class: words.Thesaurus.  Program will exit. 

然后通过手动的添加taskmanager又没有问题，莫名其妙的问题，我机器配置的JAVA_HOME都没有问题，而且java版本都是1.8.0_191，对于这个问题，我们可以在flink-conf.yaml文件中配置java路径：

在文件中添加你的java路径：  env.java.home： XXX/jdk_1.8.0

如有问题，欢迎随时交流！

WeChat：359502980