小红书数据平台工程师1面

大家好，我是土哥。

周五晚上过来卷大家了，今天为大家带来一位读者面试小红书的 数据平台工程师（实时方向）面经。

面试时间：72 分钟

面试方向：数据平台工程师（实时方向）

面试工具：赛码网

面试难度 : ⭐⭐⭐⭐

岗位要求：如下图。

流计算平台

面试官： 不用自我介绍了，直接介绍一下流计算平台

纳尼？好高冷啊！！

具体自我介绍请查看： 58同城大数据开发社招面经(附答案)

面试官：1 你们的 UDF 是怎么管理的？

当自定义 UDF jar 后，

1. 如果是 sql 类型以来的 jar , UDF JAR 文件会被上传到 OSS Bucket 中的 sql-artifacts 目录下。此外，sql 执行时，会自动提取类名，填充到 Function Name 字段中。

2. 对于 Java 类型的 UDF，其依赖可以打包到 UDF JAR 包中，也可以通过依赖文件项进行上传；对于 Python 类型的 UDF，其依赖推荐通过的单独上传依赖文件方式上传。

3. 在 UDF Artifact 文件或者其依赖文件比较大时，推荐通过外部 URL 的方式进行上传。需要注意的是，如果外部 URL 是 OSS Bucket 地址，其依赖文件必须位于 sql-artifacts/namespaces/{namespace}目录下。

面试官： 2 任务和那些 jar 包的依赖关系是怎么管理的？

当提交一个 Flink 任务，如果该任务依赖一些 jar 包，我们专门设计了一个资源管理模块，会通过该资源管理模块上传 jar 包，该 jar 包上传后，会被放入 HDFS 目录下，在任务提交时，通过 CLIFrontend 获取资源配置项进行解析。

面试官：3 Flink 应用提交包含哪些模式？

远程提交方式：分为 Standalone 方式、yarn 方式、K8s 方式

Standalone：包含 session 模式

Yarn 方式分为三种提交模式：Yarn-perJob 模式、Yarn-Sessionmo 模式、YarnApplication 模式。

K8s 方式：包含 session 模式

面试官：4 yarn per job 模式和 yarn session 模式以及 application 模式的优缺点说一下？

1 Yarn-Session 模式：所有作业共享集群资源，隔离性差，JM 负载瓶颈，main 方 法在客户端执行。适合执行时间短，频繁执行的短任务，集群中的所有作业只有一个 JobManager，另外，Job 被随机分配给 TaskManager。

特点：

Session-Cluster 模式需要先启动集群，然后再提交作业。

2 Yarn-Per-Job 模式：每个作业单独启动集群，隔离性好，JM 负载均衡，main 方法在客户端执行。在 per-job 模式下，每个 Job 都有一个 JobManager，每个 TaskManager 只有单个 Job。

特点：

一个任务会对应一个 Job，每提交一个作业会根据自身的情况，都会单独向 yarn 申请资源，独享 Dispatcher 和 ResourceManager，按需接受资源申请；适合 规模大长时间运行的作业。

3 application 模式： main 方法在 JM 中执行，入口点位于 ApplicationClusterEntryPoint，客户端只需要负责发起部署请求。

优点： 减轻客户端的压力，避免客户端资源成为瓶颈。

面试官：5 State 可以介绍一下吗？

（1） 按照由 Flink 管理 还是 用户管理，状态可以分为 原始状态（Raw State） 和 托管状态（ManagedState）。

托管状态（ManagedState）:由 Flink 自行进行管理的 State。

原始状态（Raw State）：由用户自行进行管理。

两者区别：

1. Managed State 由 Flink Runtime 管理，自动存储，自动恢复，在内存管理上有优化；而 Raw State 需要用户自己管理，需要自己序列化。

2. Managed State 支持已知的数据结构，如 Value、List、Map 等。而 Raw State 只支持 字节 数组，所有状态都要转换为二进制字节数组才可以。

（2）State 按照是否有 key 划分为 KeyedState 和 OperatorState 两种。

KeyedState 只能用在 keyedStream 上的算子中，状态跟特定的 key 绑定。keyStream 流上的每一个 key 对应一个 state 对象。若一个 operator 实例处理多个 key，访问相应的多个 State，可对应多个 state。

keyedState 保存在 StateBackend 中，通过 RuntimeContext 访问，实现 Rich Function 接口。支持多种数据结构：ValueState、ListState、ReducingState、 AggregatingState、MapState。

OperatorState 可以用于所有算子，但整个算子只对应一个 state。

OperatorState 实现 CheckpointedFunction 或者 ListCheckpointed 接口。目前只支持 ListState 数据结构。

（3）在 Flink 中， state 一般被存储在 StateBackend 里面。

总共包含三种存储方式，内存、文件、RocksDB 等。

面试官：6 什么样的业务场景你会选择 FsBackend 或者 RocksDB ?

1 MemoryStateBackend，运行时所需的 State 数据全部保存在 TaskManager JVM 堆上内存中，执行检查点的时候，会把 State 的快照数据保存到 JobManager 进程 的内存中。基于内存的 Stateßackend 在生产环境下不建议使用，可以在本地开发调试测试 。

1. State 存储在 JobManager 的内存中.受限于 JobManager 的内存大小。
2. 每个 State 默认 5MB,可通过 MemoryStateBackend 构造函数调整。
3. 每个 Stale 不能超过 Akka Frame 大小。

2 FSStateBackend，运行时所需的 State 数据全部保存在 TaskManager 的内存 中， 执行检查点的时候，会把 State 的快照数据保存到配置的文件系统中。

适用场景：

FSStateBackend 适用于处理大状态、长窗口、或者大键值状态的有状态处理任务。

1. State 数据首先被存在 TaskManager 的内存中。
2. State 大小不能超过 TM 内存。
3. TM 异步将 State 数据写入外部存储。

3 RocksDBStateBackend 使用嵌入式的本地数据库 RocksDB 将流计算数据状态存 储在本地磁盘中。在执行检查点的时候，再将整个 RocksDB 中保存的 State 数据全量或者增量持久化到配置的文件系统中。

适用场景：

1. 最适合用于处理大状态、长窗口，或大键值状态的有状态处理任务。
2. RocksDBStateBackend 非常适合用于高可用方案。
3. RocksDBStateBackend 是目前唯一支持增量检查点的后端。 增量检查点非常适用于超大状态的场景。

技术知识点

面试官： 1 在 java 高并发时，会用到锁，你说一下 synchronized 和 ReenTrantLock的区别？

1 synchronized 是java内置关键字，在JVM层面，ReentrantLock是个java类

2 ReentrantLock包含等待可中断锁，意味着持有锁的线程长期不释放锁，正在等待的线程可以选择放弃等待，可以避免死锁

3 synchronized 自动释放锁 ReentrantLock需要在finally中手工释放锁，否则容易造成线程死锁 Lock.unlock()

4 ReentrantLock可以实现公平锁，公平锁就是先等待的线程先获得锁。

面试官：2 一个任务消费延迟了，你会怎么解决？

如果 一个任务消费延迟了，可以在 Flink WEBUI 上查看是否存在反压情况。

若看到 Flink 的哪个算子和 task 出现了反压。可以从资源调优和算子调优等方面进行解决。

资源调优即是对作业中的 Operator 的并发数（parallelism）、CPU（core）、堆内存（heap_memory）等参数进行调优。

作业参数调优包括：并行度的设置，State 的设置，checkpoint 的设置、Buffer 个数设置、Buffer 缓冲设置等。

面试官：3 你是怎么判断哪个算子存在反压呢？

在 Flink WEBUI 中，通过颜色和数值来判断任务的繁忙和反压情况，若颜色为红色，表示任务繁忙，若算子指标大于 0.5，表现为 High,证明该算子存在高度反压情况。

面试官：4 Flink 反压检测逻辑了解吗？（内部使用什么算法？） 深层次提问

Flink 1.13 版本以前，使用的堆栈采样式方式判断 反压，在 1.13 版本开始使用基于任务 Mailbox 计时方式判断反压。

面试官：5 反压有哪些危害呢？

反压如果不能得到正确的处理，可能会影响到 checkpoint 时长和 state 大小，甚至 可能会导致资源耗尽甚至系统崩溃。

1）影响 checkpoint 时长：barrier 不会越过普通数据，数据处理被阻塞也会导致 checkpoint barrier 流经整个数据管道的时长变长，导致 checkpoint 总体时间（End to End Duration）变长。

2）影响 state 大小：barrier 对齐时，接受到较快的输入管道的 barrier 后，它后面数 据会被缓存起来但不处理，直到较慢的输入管道的 barrier 也到达，这些被缓存的数据会 被放到 state 里面，导致 checkpoint 变大。

这两个影响对于生产环境的作业来说是十分危险的，因为 checkpoint 是保证数据一 致性的关键，checkpoint 时间变长有可能导致 checkpoint 超时失败，而 state 大小同样可能拖慢 checkpoint 甚至导致 OOM （使用 Heap-based StateBackend）或者物理内存使用超出容器资源（使用 RocksDBStateBackend）的稳定性问题。

面试官：6 反压如何解决，排查方式有哪些？

1. 解决反压首先要做的是定位到造成反压的节点,排查的时候，先把 operator chain 禁用，方便定位到具体算子。

因为 Flink 默认是开启 operator chain 的，他会将多个 operator 串在一起作为一个 operator chain, 以便提高程序的性能，禁用后 每个 operator 都会执行。

2. 在 WEBUI 上看到某个算子出现反压，若该节点的发送速率跟不上它的产生数据速率。这一般会发生在一条输入多条输出的 operator 上，该节点是反压的根源节点，它是从 Source Task 到 Sink Task 的第一个出现反压的节点。 若下游的节点接受速率较慢，通过反压机制限制了该节点的发送速率。这种情况，需要继续排查下游节点，一直找到第一个为 OK 的一般就是根源节点。

3. 利用 Metrics 定位，因为数据在传输过程中，会和 Channel 接受端的 Buffer 使用率有关。 例如 发送端 Buffer 使用率、接收端 Buffer 的使用率等。

如果 Subtask 发送端 Buffer 使用率高，代表被下游反压限速了，如果 Subtask 接收端 Buffer 使用率高，表明将反压传导至上游。

具体看如下图：

4. 使用火焰图功能判断用户的代码是否存在性能问题，，Flink 1.13 直接在 WebUI 提供 JVM 的 CPU 火焰图，默认是不开启的，需要修改参数：rest.flamegraph.enabled: true #默认 false。修改后，用来分析 Task Thread 是否跑满一个 CPU 核。

火焰图是通过对堆栈跟踪进行多次采样来构建的，火焰图纵向代表调用链，横向代表样本出现次数。看顶层的哪个函数占据的宽度最大。只要有"平顶"（plateaus），就表示该函数可能存在性能问题，具体如下图。

5. 分析 GC 情况，TaskManager 的内存以及 GC 问题也可能会导致反压，包括 TaskManager JVM 各区内存不合理导致的频繁 Full GC 甚至失联。通常建议使用默认的 G1 垃圾回收器。可以通过打印 GC 日志 使用 GC 分析器（GCViewer 工具）来验证这种情况。

6. 也有可能是数据倾斜所造成的反压。

面试官：7 为什么数据倾斜会造成反压？

首先，数据倾斜 是由于不同的 key 对应的数据量不同，而导致不同 task 所处理的数据量不同的问题。如果在 Flink 相同 Task 的多个 Subtask 中，个别 Subtask 接收到的数据量明显大于其他 Subtask 接收到的数据量，就会造成数据倾斜，这样就会导致任务执行过慢，从而引起反压情况。

Flink Web UI 可以精确地看到每个 Subtask 处理了多少数据，即可判断出 Flink 任务是否存在数据倾斜。

面试官：8 Flink 中数据倾斜有哪些解决办法？

先分析是 keyBy 前还是 keyBy 后发生数据倾斜。

1. keyBy 后的聚合操作存在数据倾斜。使用 LocalKeyBy 的思想，在 keyBy 上游算子数据发送之前，首先在上游算子的本地对数据进行聚合后，再发送到下游，使下游接收到的数据量大大减少，从而使得 keyBy 之后的聚合操作不再是任务的瓶颈。

2. keyBy 之前发生数据倾斜。产生该情况可能是因为数据源的数据本身就不均匀，例如由于某些原因 Kafka 的 topic 中某些 partition 的数据量较大，某些 partition 的数据量较少。 解决办法：需要让 Flink 任务强制进行 shuffle。使用 shuffle、rebalance 或 rescale 算子即可将数据均匀分配，从而解决数据倾斜的问题。

3. keyBy 后的窗口聚合操作存在数据倾斜。因为使用了窗口，变成了有界数据（攒批）的处理，窗口默认是触发时才会输出一条结果发往下游，所以可以使用两阶段聚合的方式。

解决办法：

第一阶段聚合：key 拼接随机数前缀或后缀，进行 keyby、开窗、聚合。

第二阶段聚合：按照原来的 key 及 windowEnd 作 keyby、聚合

算法题

面试官：1 写一个算法吧，求 两数之和

解题思路：

创建一个哈希表，对于每一个 x，我们首先查询哈希表中是否存在 target - x，然后将 x 插入到哈希表中，即可保证不会让 x 和自己匹配。

使用哈希表，可以将寻找 target - x 的时间复杂度降低到从 O(N) 降低到 O(1)。

class Solution {
   public int[] twoSum(int[] nums, int target) {
       Map<Integer, Integer> hashtable = new HashMap<Integer, Integer>();
       for (int i = 0; i < nums.length; ++i) {
           if (hashtable.containsKey(target - nums[i])) {
               return new int[]{hashtable.get(target - nums[i]), i};
           }
           hashtable.put(nums[i], i);
       }
       return new int[0];
   }
}

复杂度分析:

时间复杂度：O(N)，其中 N 是数组中的元素数量。对于每一个元素 x，我们可以O(1) 地寻找 target - x。

空间复杂度：O(N)，其中 N 是数组中的元素数量。主要为哈希表的开销。

面试官：2 再写个 三数之和

解题思路：

ArrayList 集合 + 排序 + 双指针

1. 先创建集合 ArrayList。
2. 判断数组长度小于3，返回定义的集合。
3. 对数组排序 Arrays.sort(nums);
4. for循环遍历数组；
5. 判断，当nums[i] >0 返回 break；因为nums[i] 大于0了，后面的都大于0了
6. 判断 当nums[i] ==nums[i+1],continue;
7. 定义双指针 L = i+1; R = len-1;
8. while(R>L) 定义 sum = nums[i] + nums[L] + nums[R];
9. 判断 当sum == 0, 将三个数封装成集合，装进 listArray.
10. 当 sum > 0,证明R 往后更大，所以 R--；
11. 当 sum < 0,证明 R已经是最大了，L应该往右移，L++。
12. 返回 listArray

class Solution {
    
    //nums = [-1,0,1,2,-1,-4] ,[[-1,-1,2],[-1,0,1]]
    
    public List<List<Integer>> threeSum(int[] nums) {
        
        // 1 新建 list 集合
        List<List<Integer>> listArray = new ArrayList<>();
        
        // 2 条件判断
        
        if(nums ==null || nums.length<3){
            return listArray;
        }
        
        //3 排序 [-4,-1,-1,0,1,2]
        Arrays.sort(nums);
        
        int len = nums.length;
        
        // 4 for 循环遍历
        for(int i =0;i<len;i++){
            if(nums[i]>0) break;
            if(i>0 && nums[i] == nums[i-1]) continue;
            // 5 定义双指针
             int L = i+1;
             int R = len-1;
            while(R>L){
                int sum = nums[i] + nums[L] + nums[R];
                // 6 判断 sum ==0
                if(sum ==0){
                    listArray.add(Arrays.asList(nums[i],nums[L],nums[R]));
                    while(L<R && nums[L] ==nums[L+1]) L++;
                    while(L<R && nums[R] == nums[R-1]) R--;
                    L++;
                    R--;
                }
                if(sum >0){
                 R--;   
                }
                if(sum <0){
                    L++;
                }
            }
        }
         return listArray;   
    }
}

时间复杂度：O(n^2)