字节跳动大数据开发面试常见问题及技术解析

在生成 Operation 后，Flink 会进一步将其转化为物理执行计划，主要包含以下步骤：

1. TranslateToRel

   • 将 ModifyOperation 转换为 Calcite RelNode 逻辑计划树。
   • 映射为 FlinkLogicalRel，保留逻辑语义。

2. Optimize（优化）

   • 调用 optimize() 方法，将逻辑计划优化为物理计划（FlinkPhysicalRel）。
   • RBO（基于规则的优化）：包括谓词下推、Join 顺序重排、列裁剪、分区裁剪等。
   • CBO（基于代价的优化）：根据统计信息估算成本，选择最优 Join 策略。

3. TranslateToExecNodeGraph

   • 将物理计划转换为执行图（execGraph），定义算子间的依赖与数据流转。

4. TranslateToPlan

   • 将 execGraph 转换为最终的 Transformation 对象，准备提交给 JobManager。

RBO 规则优化详解

常见的 RBO 规则包括：

• 谓词下推（Predicate Pushdown）：将过滤条件尽可能下推到存储层或扫描层，减少数据传输量。
• Join 优化：调整 Join 顺序，优先连接小表以减少中间结果集大小。
• 列裁剪（Column Pruning）：仅读取查询所需的列，避免全表扫描。
• 分区裁剪（Partition Pruning）：
  • 静态裁剪：编译期根据常量分区键直接排除无关分区。
  • 动态裁剪：运行期根据变量分区键动态判断需访问的分区范围。
  • 作用：显著降低 IO 请求，提升查询性能。

Spark 3.0 AQE 优化特性

Spark 3.0 引入了自适应查询执行（AQE），主要包含三项核心优化：

1. 动态合并 Shuffle 分区

   • 背景：Shuffle 前分区数过多会导致大量小文件 IO，过少则导致单个分区数据倾斜。
   • 机制：在 Map 端输出后，若相邻分区数据量过小（低于阈值），自动合并为一个分区，平衡负载。

2. 动态调整 Join 策略

   • 策略类型：Broadcast Hash Join、Hash Join、SortMerge Join。
   • 优化逻辑：运行时根据实际数据量动态选择最优策略。例如，当小表数据量小于广播阈值时，强制使用 Broadcast Hash Join。

3. 动态优化数据倾斜 Join

   • 问题：Key 分布不均导致部分 Task 处理时间过长。
   • 方案：检测到倾斜 Key 后，自动拆分大 Key，引入随机前缀打散数据，聚合后再去重，消除倾斜。

Spark 3.0 如何达到 Broadcast Hash Join 要求？

当两张表 Join 时，若 A 表数据量超过广播阈值，通常无法使用 Broadcast Hash Join。但 Spark 3.0 提供了灵活的处理方式：

• Filter 预过滤：如果 A 表存在无用数据，且 B 表无对应无用数据，可通过 Filter 条件提前过滤 A 表。
• 动态调整：过滤后若 A 表剩余数据量小于广播阈值，AQE 会自动将 Join 策略调整为 Broadcast Hash Join，从而获得最佳性能。

Flink Checkpoint 失败原因及解决方案

Checkpoint 失败通常与反压（Backpressure）密切相关，主要原因如下：

1. 数据流动缓慢导致超时

• 现象：Barrier 在 Buffer 中排队等待，因反压导致到达算子时间过长，超过 Timeout 设置。
• Barrier 对齐阻塞：在多输入源算子中，若一个 Barrier 到达，其他输入的数据会被阻塞，直到所有 Barrier 对齐。这严重影响了实时性。
• 解决方案：
  • FLIP-183（动态 Buffer 调整）：限制 Buffer 缓存量，只缓存配置时间内可处理的数据，加速 Barrier 流动。
  • FLIP-76（Unaligned Checkpoint）：允许 Barrier 不对齐，先处理数据再异步等待 Barrier，牺牲少量一致性换取高性能，适用于对实时性要求高、允许少量重复数据的场景。

2. 状态数据过大

• 现象：State Backend 写入持久化存储耗时过长，IO 压力大，导致超时。
• 解决方案：
  • FLIP-158（增量快照）：
    • 状态变化不仅写入 State Backend，同时写入预写日志（Changelog）。
    • Changelog 细粒度记录状态变更，上传至持久化存储后确认完成。
    • 定期物化状态（State Table），清理旧日志，减少单次 Snapshot 数据量。

Flink 滑动窗口特点

Flink 窗口由两个属性定义：窗口长度（Size）和滑动间隔（Interval）。

• 滚动窗口：Size = Interval，无重叠。
• 滑动窗口：Size > Interval，数据有重叠。

示例：

• timeWindow(Time.seconds(10), Time.seconds(5))：每 5 秒触发一次，计算过去 10 秒内的数据。
• countWindow(10, 5)：每 5 条数据触发一次，计算最近 10 条数据。

算法题实战

1. 旋转数组最小值

题目：给定一个有序数组，前 n 位往后移，求其中的最小值。 思路：利用二分查找，时间复杂度 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;
            // 如果中间元素小于右边界，说明最小值在左半部分（含 mid）
            if (nums[mid] < nums[high]) {
                high = mid;
            } else {
                // 否则最小值在右半部分（不含 mid）
                low = mid + 1;
            }
        }
        return nums[low];
    }
}

2. LRU 缓存算法

原理：最近最少使用（Least Recently Used），淘汰最久未使用的数据。 实现：哈希表 + 双向链表。

• 哈希表：O(1) 定位节点。
• 双向链表：维护访问顺序，头部为最近使用，尾部为最久未使用。
• 操作：Get/Put 均为 O(1)。

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;
        }
        moveToHead(node);
        return node.value;
    }

    public void put(int key, int value) {
        DLinkedNode node = cache.get(key);
        if (node == null) {
            DLinkedNode newNode = new DLinkedNode(key, value);
            cache.put(key, newNode);
            addToHead(newNode);
            ++size;
            if (size > capacity) {
                DLinkedNode tailNode = removeTail();
                cache.remove(tailNode.key);
                --size;
            }
        } else {
            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)