大数据架构面试核心问题解析：Flink、Spark 与算法

3. 优化策略

面试官：RBO 里面都了解哪些规则优化？

RBO 规则优化中包含谓词下推、Join 优化、列裁剪、分区裁剪等等。

面试官：分区裁剪主要解决什么问题？

分区剪裁就是对于分区表或者分区索引来说，优化器可以自动从 from 和 where 中根据分区键直接提取出需要访问的分区，从而避免扫描所有的分区，降低了 IO 请求。

分区剪裁可以细分为静态分区剪裁和动态分区剪裁。其中静态分区剪裁发生在 SQL 语句编译阶段，而动态分区剪裁则发生在 SQL 语句执行阶段。对于分区键是常量值优化器会走静态分区剪裁的，如果分区键是变量形式优化器只会走动态分区剪裁。

4. Join 类型

面试官：那在 Flink SQL 中，join 都包含哪些类型？（引擎层的实现）

在 Join 中包含 Regular Join、Interval Join、Temporal Join、Lookup Join。

• Regular Join：包含 Left Join、Right Join、Inner Join、Full Join。
• Interval Join：时间区间 Join，表示两条流之间一段时间的 Join。

三、Spark 优化特性

面试官：Spark 3.0 优化特性了解不？

了解 Spark 3.0 AQE（Adaptive Query Execution）自适应查询优化。

Spark 3.0 AQE 自适应查询里面包含 3 种优化，如动态合并 Shuffle 分区、动态调整 Join 策略、动态优化数据倾斜 Join 等。

(1) 动态合并 Shuffle 分区 在 Spark 中，Shuffle 前后的分区不同。如果分区数太少，那么每个分区处理的数据大小可能非常大，导致大分区处理时需要落盘，查询效率太低；如果分区过多，导致每个分区处理数据较少，这也会导致 IO 请求增多降低查询效率。 动态合并 Shuffle 的含义就是当 Map 端的两个分区经过 Shuffle 操作后，本来产生五个分区的，但因为有两个分区数据过小，所以直接对其进行合并，最终输出 3 个分区。

(2) 动态调整 Join 策略 总共包含 3 种 Join 策略：Broadcast Hash Join、Hash Join、SortMergeJoin。

(3) 动态优化数据倾斜 Join 针对数据倾斜场景，AQE 能够识别出倾斜的 Key，并将倾斜的 Task 单独拆分或广播小表来优化。

面试官：假如两张表 Join，但目前达不到 Broadcast Hash Join 的要求，Spark 3.0 是怎么处理的可以让其达到要求？

在 Spark 3.0 AQE 中通过动态调整 Join 策略，其中 Broadcast Hash Join 性能是最好的，前提是参加 Join 的一张表的数据能够装入内存。由于这个原因，当 Spark 估计参加 Join 的表数据量小于广播大小的阈值时，其会将 Join 策略调整为 Broadcast Hash Join。 所以当两张表 Join 时，如果 A 表的数据量大于广播大小的阈值，此时不能选择 Broadcast Hash Join，但恰好可以通过 Filter 条件将 A 表无用数据过滤掉，且 B 表不包含无用数据，这时候过滤掉后的 A 表数据量小于广播大小的阈值，此时就可以选择 Broadcast Hash Join。

四、Checkpoint 与容错

面试官：Checkpoint 失败有遇到过吗，什么原因导致的？

有遇到过，Checkpoint 失败一般都和反压相结合。导致 Checkpoint 失败的原因主要有两个：

1. 数据流动缓慢，Checkpoint 执行时间过长 我们知道，Flink Checkpoint 机制是基于 Barrier 的。在数据处理过程中，Barrier 也需要像普通数据一样，在 Buffer 中排队，等待被处理。当 Buffer 较大或者数据处理较慢时，Barrier 需要很久才能够到达算子，触发 Checkpoint。尤其是当存在反压时，Barrier 需要在 Buffer 中流动数个小时，从而导致 Checkpoint 执行时间过长，超过了 Timeout 还没有完成，从而导致失败。 当算子需要 Barrier 对齐时，如果一个输入的 Barrier 已经到达，那么该输入的 Barrier 后面的数据会被阻塞住，不能被处理，需要等到其他输入 Barrier 到达之后，才能继续处理。在 Barrier 对齐过程中，其他输入数据处理都要暂停，将严重导致应用实时性，从而让 Checkpoint 执行时间过长，超过了 Timeout 还没有完成，导致执行失败。

2. 状态数据过大 当状态数据过大，会影响每次 Checkpoint 的时间，并且在 Checkpoint 时 IO 压力也会很大，执行时间过长，导致超时还没有执行成功，从而导致执行失败。

面试官：怎么解决的上述问题？

对于数据流动缓慢解决思路是： 让 Buffer 中的数据变少，让 Barrier 能跳过 Buffer 中存储的数据。 这对应社区提出的 FLIP-183 Dynamic buffer size adjustment，其解决思路是只缓存配置时间内可以处理的数据量，这可以很好的控制 Checkpoint。 对于 Barrier 对齐问题，社区提出 FLIP-76 Unaligned Checkpoint。其解决思路是对于实时性要求很好，但数据重复性要求低的，可采用 Barrier 不对齐模式。当还有其他流的 Barrier 还没到达时，为了不影响性能，不用理会，直接处理 Barrier 之后的数据。等到所有流的 Barrier 的都到达后，就可以对该 Operator 做 CheckPoint 了。

对于状态数据过大问题： FLIP-158 提出通用的增量快照方案，其核心思想是基于 State Changelog。Changelog 能够细粒度地记录状态数据的变化。具体描述如下：

• 有状态算子除了将状态变化写入状态后端外，再写一份到预写日志中。
• 预写日志上传到持久化存储后，Operator 确认 Checkpoint 完成。
• 独立于 Checkpoint 之外，State Table 周期性上传，这些上传到持久存储中的数据被称为物化状态。
• State Stable 上传后，之前部分预写日志就没用了，可以被裁剪。

五、窗口计算

面试官：滑动窗口有啥特点？

Flink 支持的窗口包含两个重要属性（窗口长度 Size 和滑动间隔 Interval），通过窗口长度和滑动间隔来区分滚动窗口和滑动窗口。 滑动窗口（Sliding Window）数据有重叠，即 Size(1min) > Interval(30s)。

• timeWindow(Time.seconds(10), Time.seconds(5)) --- 基于时间的滑动窗口
• countWindow(10, 5) --- 基于数量的滑动窗口

六、常见算法题

面试官：我们写两道算法吧，先看看第一道 给定一个有序数组，前 n 位往后移，例如 {1,2,3,4,5} -> {3,4,5,1,2}，求其中的最小值。 该题其实就是让你用最优解找一个数组中的最小值，可以使用二分查找法。

时间复杂度 O(log n)，空间复杂度 O(1)

public class Main {
    public static void main(String[] args) {
        int[] nums = {4, 5, 6, 7, 1, 2, 3};
        System.out.println(test(nums));
    }

    public static int test(int[] nums) {
        int low = 0;
        int high = nums.length - 1;

        while (low < high) {
            int mid = (low + high) / 2;
            if (nums[mid] < nums[high]) {
                high = mid;
            } else {
                low = mid + 1;
            }
        }
        return nums[low];
    }
}

面试官：LRU 算法，先说一下原理，然后介绍一下实现思路 LRU 被称作最近最少使用算法，是一种页面置换算法。其核心思想是将最近最久未使用的页面予以淘汰。就是一种缓存淘汰算法。

实现思路： LRU 缓存机制可以通过哈希表 + 双向链表实现，我们用一个哈希表和一个双向链表维护所有在缓存中的键值对。 双向链表按照被使用的顺序存储了这些键值对，靠近头部的键值对是最近使用的，而靠近尾部的键值对是最久未使用的。 哈希表即为普通的哈希映射（HashMap），通过缓存数据的键映射到其在双向链表中的位置。 这样以来，我们首先使用哈希表进行定位，找出缓存项在双向链表中的位置，随后将其移动到双向链表的头部，即可在 O(1) 的时间内完成 get 或者 put 操作。

public class LRUCache {
    class DLinkedNode {
        int key;
        int value;
        DLinkedNode prev;
        DLinkedNode next;
        public DLinkedNode() {}
        public DLinkedNode(int _key, int _value) { key = _key; value = _value; }
    }

    private Map<Integer, DLinkedNode> cache = new HashMap<>();
    private int size;
    private int capacity;
    private DLinkedNode head, tail;

    public LRUCache(int capacity) {
        this.size = 0;
        this.capacity = capacity;
        // 使用伪头部和伪尾部节点
        head = new DLinkedNode();
        tail = new DLinkedNode();
        head.next = tail;
        tail.prev = head;
    }

    public int get(int key) {
        DLinkedNode node = cache.get(key);
        if (node == null) {
            return -1;
        }
        // 如果 key 存在，先通过哈希表定位，再移到头部
        moveToHead(node);
        return node.value;
    }

    public void put(int key, int value) {
        DLinkedNode node = cache.get(key);
        if (node == null) {
            // 如果 key 不存在，创建一个新的节点
            DLinkedNode newNode = new DLinkedNode(key, value);
            // 添加进哈希表
            cache.put(key, newNode);
            // 添加至双向链表的头部
            addToHead(newNode);
            ++size;
            if (size > capacity) {
                // 如果超出容量，删除双向链表的尾部节点
                DLinkedNode tail = removeTail();
                // 删除哈希表中对应的项
                cache.remove(tail.key);
                --size;
            }
        } else {
            // 如果 key 存在，先通过哈希表定位，再修改 value，并移到头部
            node.value = value;
            moveToHead(node);
        }
    }

    private void addToHead(DLinkedNode node) {
        node.prev = head;
        node.next = head.next;
        head.next.prev = node;
        head.next = node;
    }

    private void removeNode(DLinkedNode node) {
        node.prev.next = node.next;
        node.next.prev = node.prev;
    }

    private void moveToHead(DLinkedNode node) {
        removeNode(node);
        addToHead(node);
    }

    private DLinkedNode removeTail() {
        DLinkedNode res = tail.prev;
        removeNode(res);
        return res;
    }
}

时间复杂度 O(1) 空间复杂度 O(capacity)

七、架构设计补充

面试官：如何处理数据倾斜？

数据倾斜通常指在分布式计算中，某些节点处理的数据量远大于其他节点，导致整体任务进度受限于慢节点。 解决方案包括：

1. Key 加盐：给倾斜的 Key 加上随机前缀，打散到不同分区，聚合时再去掉前缀。
2. 广播小表：如果是 Join 倾斜，尝试将小表广播到所有节点。
3. 调整并行度：增加倾斜任务的并行度，分担负载。
4. 自定义 Partitioner：根据数据特征自定义分区策略。

面试官：Kafka 如何保证消息不丢失？

1. Producer 端：设置 acks=all，确保所有副本确认收到；开启重试机制 retries。
2. Broker 端：设置 min.insync.replicas 至少为 2，确保至少有一个副本存活；关闭 unclean.leader.election.enable。
3. Consumer 端：关闭自动提交 Offset，手动提交 Offset，确保消费成功后再提交。

八、总结

本文涵盖了大数据架构面试的核心知识点，包括 Flink SQL 解析、Ranger 鉴权、Checkpoint 机制、Spark AQE 优化、窗口计算以及常见算法题解。建议面试者在准备时不仅关注 API 的使用，更要深入理解底层原理和优化策略，结合项目经验进行阐述。