大数据Flink组件安装部署

0 介绍

Flink起源于Stratosphere项目，Stratosphere是在2010~2014年由3所地处柏林的大学和欧洲的一些其他的大学共同进行的研究项目，2014年4月Stratosphere的代码被复制并捐赠给了Apache软件基金会，参加这个孵化项目的初始成员是Stratosphere系统的核心开发人员，2014年12月，Flink一跃成为Apache软件基金会的顶级项目。
在德语中，Flink一词表示快速和灵巧，项目采用一只松鼠的彩色图案作为logo，这不仅是因为松鼠具有快速和灵巧的特点，还因为柏林的松鼠有一种迷人的红棕色，而Flink的松鼠logo拥有可爱的尾巴，尾巴的颜色与Apache软件基金会的logo颜色相呼应，也就是说，这是一只Apache风格的松鼠

1 安装

1.1 standalone模式

1） 解压缩 flink-1.10.0-bin-scala_2.11.tgz，进入conf目录中。

2） 修改 flink/conf/flink-conf.yaml 文件：

3） 修改conf/slvaes文件

4) 分发到集群其他需要安装flink的节点
5) 启动start-cluster.sh

访问http://192.168.199.101:8081可以对flink集群和任务进行监控管理。

提交任务测试：
1)编写如下wordcount并打成jar包

package com.chen.flink.part01

import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._


// 流处理 word count
object StreamWordCount {
  def main(args: Array[String]): Unit = {
    // 创建流处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //    env.setParallelism(8)
    //    env.disableOperatorChaining()

    // 从程序运行参数中读取hostname和port
    val params: ParameterTool = ParameterTool.fromArgs(args)
    val hostname: String = params.get("host")
    val port: Int = params.getInt("port")

    // 接受socket文本流
    val inputDataStream: DataStream[String] = env.socketTextStream(hostname, port)
//        val inputDataStream: DataStream[String] = env.socketTextStream("master-2", 9999)

    // 定义转换操作，word count
    val resultDataStream: DataStream[(String, Int)] = inputDataStream.flatMap(_.split(" "))
      .map((_,1)).keyBy(0).sum(1)

    resultDataStream.print()
    env.execute("stream word count job")

  }
}

2)提交job执行

bin/flink run -c com.chen.flink.part01.StreamWordCount  /root/Flink2-1.0-SNAPSHOT-jar-with-dependencies.jar --host localhost --port 12345

在master-1机器执行nc –lk 12345并输入数据：

在flink的8081页面查看：

1.2 yarn模式

以Yarn模式部署Flink任务时，要求Flink是有Hadoop支持的版本，Hadoop环境需要保证版本在2.2以上，并且集群中安装有HDFS服务。
Flink提供了两种在yarn上运行的模式，分别为Session-Cluster和Per-Job-Cluster模式。

1.2.1 Session-cluster 模式

Session-Cluster模式需要先启动集群，然后再提交作业，接着会向yarn申请一块空间后，资源永远保持不变。如果资源满了，下一个作业就无法提交，只能等到yarn中的其中一个作业执行完成后，释放了资源，下个作业才会正常提交。所有作业共享dispatcher和ResourceManager；共享资源；适合规模小执行时间短的作业。
在yarn中初始化一个flink集群，开辟指定的资源，以后提交任务都向这里提交。这个flink集群会常驻在yarn集群中，除非手工停止。

1.2.2 Per-Job-Cluster 模式

一个Job会对应一个集群，每提交一个作业会根据自身的情况，都会单独向yarn申请资源，直到作业执行完成，一个作业的失败与否并不会影响下一个作业的正常提交和运行。独享dispatcher和ResourceManager，按需接受资源申请；适合规模大长时间运行的作业。
每次提交都会创建一个新的flink集群，任务之间互相独立，互不影响，方便管理。任务执行完成之后创建的集群也会消失。

1.3 启动Session Cluster

1. 启动hadoop集群（略）
2. 启动yarn-session

./yarn-session.sh -n 2 -s 2 -jm 1024 -tm 1024 -nm test -d

其中：
-n(–container)：TaskManager的数量。
-s(–slots)： 每个TaskManager的slot数量，默认一个slot一个core，默认每个taskmanager的slot的个数为1，有时可以多一些taskmanager，做冗余。
-jm：JobManager的内存（单位MB)。
-tm：每个taskmanager的内存（单位MB)。
-nm：yarn 的appName(现在yarn的ui上的名字)。
-d：后台执行。

查看yarn页面：

3) 执行任务

bin/flink run -c com.chen.flink.part01.StreamWordCount  /root/Flink2-1.0-SNAPSHOT-jar-with-dependencies.jar --host localhost --port 12345

4) 查看flink页面

5)取消yarn-session

yarn application --kill application_1622476234777_0004

1.3 启动 Per Job Cluster

1)启动hadoop集群（略）
2)不启动yarn-session，直接执行job

bin/flink run –m yarn-cluster -c com.chen.flink.part01.StreamWordCount  /root/Flink2-1.0-SNAPSHOT-jar-with-dependencies.jar --host master-1 --port 12345