带你认识FusionInsight Flink：既能批处理，又能流处理

摘要：本文主要介绍了FusionInsight Flink组件的基本原理、Flink任务提交的常见问题、以及最佳实践FAQ。

本文分享自华为云社区《FusionInsight HD Flink组件基本原理和常见问题解析》，作者：FI小粉丝 。

Flink是一个批处理和流处理结合的统一计算框架，其核心是一个提供数据分发以及并行化计算的流数据处理引擎。

它的最大亮点是流处理，是业界最顶级的开源流处理引擎。

Flink最适合的应用场景是低时延的数据处理（Data Processing）场景：高并发pipeline处理数据，时延毫秒级，且兼具可靠性。

本文主要介绍了FusionInsight Flink组件的基本原理、Flink任务提交的常见问题、以及最佳实践FAQ。

基本概念

基本原理

简介

Flink是一个批处理和流处理结合的统一计算框架，其核心是一个提供了数据分发以及并行化计算的流数据处理引擎。它的最大亮点是流处理，是业界最顶级的开源流处理引擎。

Flink最适合的应用场景是低时延的数据处理（Data Processing）场景：高并发pipeline处理数据，时延毫秒级，且兼具可靠性。

Flink技术栈如图所示：

Flink在当前版本中重点构建如下特性，其他特性继承开源社区，不做增强，具体请参考：slot优化。

FAQ

Flink如何加载其它目录的jar包

需求描述

Flink业务一般在运行过程中默认加载的jar包路径为：“xxx/Flink/flink/lib”的目录下，如果添加其它路径的jar包会报错，如何添加其它外部依赖。

实现方案

创建一个外部的lib目录，将部分依赖包放到外部lib目录下，如下图：

修改启动脚本的参数配置脚本，config.sh将jar包路径传给环境变量中。

此时正常启动任务即可，不需要加其它参数。

HDFS上也能看到第三方jar的目录。

如何收集任务taskmanager的jstack和pstree信息

需求描述

在任务运行过程中我们通常需要对taskmanager的进程进行查询和处理，例如：打jstack，jmap等操作，做这些操作的过程中需要获取任务的taskmanager信息。

实现方案

获取一个nodemanager节点上面所有taskmanager的进程信息的方法如下：

ps -ef | grep taskmanager | grep -v grep | grep -v "bash -c"

其中红框中的内容就是taskmanager的进程号，如果一个节点上面存在多个taskmanager那么这个地方会有多个进程号。获取到进程号后我们可以针对这个进程号收集jstack或者pstree信息。

收集jstack

1.通过上面流程获取到进程信息，然后从中获取进程ID和application id，如上图中进程id为“30047 applicationid为application_1623239745860_0001”。

2.从FI前台界面获取这个进程的启动用户。如下图为flinkuser。

3.在对应的nodemanager节点后台切换到这个用户，人机用户机机用户即可。

4. 进入到节点所在的jdk目录下

5. 给taskmanager进程打jstack。

不同用户提交的taskmanager只能由提交任务的用户打jstack。

收集pstree信息

使用pstree –p PID 的方式能够获取taskmanager的pstree信息，这个地方提供一个收集脚本。内容如下：

#!/bin/bash
searchPID() {
   local pids=`ps -ef | grep taskmanager | grep -v grep | grep -v "bash -c" | grep -v taskmanagerSearch.sh | awk '{print $2}'`;   
 time=$(date "+%Y-%m-%d %H:%M:%S")
 echo "checktime is  --------------------- $time" >> /var/log/Bigdata/taskManagerTree.log
  for i in $pids
    do
       local treeNum=$(pstree -p $i | wc -l)
       echo "$i 's pstree num is $treeNum" >> /var/log/Bigdata/taskManagerTree.log
    done
}
searchPID

该脚本的功能为获取节点上所有taskmanager pstree的数量，打印结果如下：

slot优化

需求描述

Slot可以认为是taskmanager上面一块独立分配的资源，是taskmanager并行执行的能力的体现。Taskmanager中有两种使用slot的方法：

• 一个taskmanager中设置了一个slot。
• 一个taskmanager中设置了多个slot。

每个task slot 表示TaskManager 拥有资源的一个固定大小的子集。假如一个taskManager 有三个slot，那么它会将其管理的内存分成三份给各个slot。资源slot化意味着一个subtask 将不需要跟来自其他job 的subtask 竞争被管理的内存，取而代之的是它将拥有一定数量的内存储备。需要注意的是，这里不会涉及到CPU 的隔离，slot 目前仅用来隔离task 的受管理的内存。通过调整task slot 的数量，允许用户定义subtask 之间隔离的方式。如果一个TaskManager 一个slot，那将意味着每个task group运行在独立的JVM 中（该JVM可能是通过一个特定的容器启动的），而一个TaskManager 多个slot 意味着更多的subtask 可以共享同一个JVM。而在同一个JVM 进程中的task 将共享TCP 连接（基于多路复用）和心跳消息。它们也可能共享数据集和数据结构。因此，对于资源密集型任务（尤其是对cpu使用较为密集的）不建议使用单个taskmanager中创建多个slot使用，否则容易导致taskmanager心跳超时，出现任务失败。如果需要设置单taskmanager多slot，参考如下操作。

单taskmanager多slot的设置方法

方式一：在配置文件中配置taskmanager.numberOfTaskSlots，通过修改提交任务的客户端配置文件中的配置flink-conf.yaml配置，如下图：将该值设置为需要调整的数值即可。

方式二：启动命令的过程中使用-ys命令传入，例如以下命令：

./flink run -m yarn-cluster -p 1 -ys 3 ../examples/streaming/WindowJoin.jar

在启动后在一个taskmanager中会启动3个slot。

单taskmanager多slot需要优化哪些参数

设置单taskmanager多slot需要优化以下参数

点击关注，第一时间了解华为云新鲜技术~