Flink 专题1 ： 搭建Flink 及Flink 简介

图片来源于网络

Flink 简介

Apache Flink® – 基于数据流的有状态计算

Flink 的优势:

• 流场景使用案例
  数据驱动的应用
  批流数据分析
  数据通道和ETL
• 正确性保证
  Exactly-once状态一致性保证
  事件时间处理
  复杂的late date处理 更多
• API分层体系
  统一SQL支持Stream和Batch数据处理
  DataStream API & DataSet API
  ProcessFunction (Time & State)
• Operational Focus
  部署灵活
  高可用配置
  Savepoints
• 适用于各种应用场景Scales to any use case#
  架构可扩展
  超大state支持
  增量checkpointing
• 高性能
  低延时
  高吞吐
  内存计算

Flink 安装

安装地址:
flink ： http://mirror.bit.edu.cn/apache/flink/flink-1.6.2/flink-1.6.2-bin-hadoop27-scala_2.11.tgz
hadoop : https://archive.apache.org/dist/hadoop/common/hadoop-2.7.6/hadoop-2.7.6.tar.gz

hadoop 安装略过

flink 安装步骤

flink 安装包含单点模式，集群模式，flink on yarn 模式 ，flink on k8s 等模式 ，flink 通过是基于jvm 进行操作， 通过代码可以在单机情况下模拟 集群模式数据 ，以此可以实现本地化的degug 操作。
下面介绍一下集群模式部署:

flink 集群模式 结构 :

Flink 集群模式 包含 JobManager /TaskManager

配置文件设置:

flink-conf.yaml

jobmanager.rpc.address: test-hadoop01
jobmanager.rpc.port: 6123
 // 设置jobManager 的内存大小 
jobmanager.heap.size: 2048m
 // 设置每个taskManager 的内存大小 
taskmanager.heap.size: 3072m
// 设置每个TaskManager 所占槽位 (最好和当前 机器的 可用核数相同（注意要排除预留给自己自身的核数）)
taskmanager.numberOfTaskSlots: 8 
parallelism.default: 3  // 默认并行度 
// hdfs 地址 
fs.default-scheme: hdfs://test-hadoop02:9000/  
fs.hdfs.hadoopconf: hdfs:///flink/data/     
state.checkpoints.dir: hdfs:///checkpoints/  
//设置checkpoint 保留版本数量（选择）
state.checkpoints.num-retained: 20
// 设置savepoint 地址 （选择 ）
state.savepoints.dir: hdfs://namenode01.td.com/flink-1.5.3/flink-savepoints
//该参数控制了 Flink 是否该重新分配失败的 TaskManager 容器。默认值：true  （选择 ）
yarn.reallocate-failed:true
//ApplicationMaster 能接受最多的失败 container 数，直到 YARN 会话失败。默认：初始请求的 TaskManager 数（-n） （选择 ）
yarn.maximum-failed-containers:10
//ApplicationMaster（以及 TaskManager containers）重试次数。此参数默认值为1，如果 Application master 失败，那么整个 YARN session 会失败。如果想增大 ApplicationMaster 重启次数，可以把该参数的值调大一些。 （选择 ） 
yarn.application-attempts:5

slaves
将集群的所有节点均写入该文件中

test-hadoop01
test-hadoop02
test-hadoop03

添加jobManager/TaskManager

可以使用 bin/jobmanager.sh 和 bin/taskmanager.sh 两个脚本把 JobManager 和 TaskManager 实例添加到正在运行的集群中。
添加 JobManager

./bin/jobmanager.sh ((start|start-foreground) [host] [webui-port])|stop|stop-all

添加 TaskManager

./bin/taskmanager.sh start|start-foreground|stop|stop-all

启动集群

1 集群模式启动

/bin/start-cluster.sh

2. yarn 模式启动

./bin/yarn-session.sh

Usage:
   Required
     -n,--container <arg>   Number of YARN container to allocate (=Number of Task Managers)
   Optional
     -D <arg>                        Dynamic properties
     -d,--detached                   Start detached
     -jm,--jobManagerMemory <arg>    Memory for JobManager Container with optional unit (default: MB)
     -nm,--name                      Set a custom name for the application on YARN
     -q,--query                      Display available YARN resources (memory, cores)
     -qu,--queue <arg>               Specify YARN queue.
     -s,--slots <arg>                Number of slots per TaskManager
     -tm,--taskManagerMemory <arg>   Memory per TaskManager Container with optional unit (default: MB)
     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths for HA mode

Flink 基于 YARN 的恢复机制
Flink 的 YARN 客户端通过下面的配置参数来控制容器的故障恢复。这些参数可以通过 conf/flink-conf.yaml 或者在启动 YARN session 的时候通过 -D 参数来指定。
yarn.reallocate-failed：该参数控制了 Flink 是否该重新分配失败的 TaskManager 容器。默认值：true
yarn.maximum-failed-containers：ApplicationMaster 能接受最多的失败 container 数，直到 YARN 会话失败。默认：初始请求的 TaskManager 数（-n）
yarn.application-attempts：ApplicationMaster（以及 TaskManager containers）重试次数。此参数默认值为1，如果 Application master 失败，那么整个 YARN session 会失败。如果想增大 ApplicationMaster 重启次数，可以把该参数的值调大一些。

参考地址:
https://flink.apache.org/