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

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 ：
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/