Amazon NG 四轮 VO 面经：今年变化总结 + 三道业务场景 Coding + Bar Raiser 深挖

先说说这两年的区别

整体感觉，今年 Amazon NG 面试和去年相比变化挺明显的：

	去年	今年
招聘方式	很多岗位是 general hiring，面完再定组	基本都是组招，面试官大多直接来自目标 team
轮数	三轮	逐渐变成四轮，Bar Raiser 存在感明显增强
Coding	高频题多，刷题库收益高	越来越偏真实业务场景包装，很少一眼认出原题
AI 话题	几乎没人问	高频话题：会聊 Cursor、Claude、ChatGPT，以及如何验证 AI 生成代码的正确性

今年的 VO 流程

• 一般四轮，可能一天面完，也可能拆成两天。
• 通常三轮来自 hiring team，一轮是 Bar Raiser。
• 技术轮基本是半小时 BQ + 半小时 Coding；BR 大概率全程 BQ。
• 面试结束后一周内出结果。

第一轮：BQ + Promotion Eligibility Service

面试官来自目标组，是一位做广告平台的 SDE。开场自我介绍后直接进入 BQ。

BQ

1. 讲一次主动发现项目风险并推动解决的经历。 我讲的是实习期间发现推荐服务的缓存命中率持续下降，如果继续下去可能导致线上延迟暴涨，于是提前分析数据，推动团队优化了缓存策略。
2. 讲一次和 teammate 出现技术分歧的经历。 追问得比较细：不仅问双方方案的优缺点，还问如果最终事实证明自己错了，会怎么处理。

Coding：Promotion Eligibility Service

平台会定期推出各种促销活动，一个用户可能同时参与多个活动。系统需要支持：

• 用户加入活动、退出活动
• 查询用户当前参与了哪些活动
• 快速统计某个活动当前覆盖了多少用户

Follow-up：

• 活动数量达到几十万怎么办？
• 如果需要实时显示参与人数最多的前十个活动怎么办？
• 如何避免同一个用户被重复加入同一个活动？

整体是 HashMap + HashSet 的组合题，但包装得很像真实业务系统。

import heapq
from collections import defaultdict

class PromotionService:
    def __init__(self):
        self.user_promos = defaultdict(set)     # user -> 参与的活动
        self.promo_count = defaultdict(int)     # 活动 -> 参与人数

    def join(self, user, promo):
        if promo in self.user_promos[user]:
            return False                        # 去重：同一用户不会被重复加入
        self.user_promos[user].add(promo)
        self.promo_count[promo] += 1
        return True

    def leave(self, user, promo):
        if promo not in self.user_promos[user]:
            return False
        self.user_promos[user].remove(promo)
        self.promo_count[promo] -= 1
        return True

    def promos_of(self, user):
        return set(self.user_promos[user])

    def user_count(self, promo):
        return self.promo_count[promo]

    def top_promos(self, k=10):
        """参与人数最多的 k 个活动，O(P log k)"""
        return heapq.nlargest(k, (p for p, c in self.promo_count.items() if c > 0),
                              key=lambda p: (self.promo_count[p], p))

Follow-up 参考思路：

• 活动几十万：每个活动只存一个计数，统计是 O(1)，不需要遍历用户。
• 实时 Top 10：上面的 nlargest 每次扫描所有活动；查询非常频繁时，可以维护一个按人数排序的结构（比如 Redis ZSET），每次 join / leave 时 O(log P) 更新分数，查询直接取前 10。
• 去重：用户维度的 HashSet 先检查再写入；分布式场景下用 (user_id, promo_id) 唯一约束兜底。

第二轮：BQ + Vendor Contract Registry

BQ：Deliver Results

讲一个时间特别紧张但最终成功上线的项目。面试官没怎么关注项目本身，而是一直追问决策过程：

• 为什么优先做某些功能？
• 为什么放弃另外一些功能？
• 如果时间再少一半，你会怎么做？

Coding：Vendor Contract Registry

公司维护大量供应商合同，系统需要支持：

• 新增合同、删除合同、更新合同状态
• 查询某个供应商当前有效合同数量

Follow-up：

• 如何快速找到未来三十天内即将到期的合同？
• 如何处理自动续约的合同？
• 多个运营人员同时修改同一个合同时，如何保证数据一致性？

面试官对代码细节看得不算特别细，反而花了很多时间讨论数据结构选型和复杂度分析。

from bisect import bisect_left, insort

class ContractRegistry:
    def __init__(self):
        self.contracts = {}        # id -> {vendor, status, end, version}
        self.active_count = {}     # vendor -> 有效合同数
        self.by_end = []           # 按到期日（整数天数）排序的 (end, id)，用于查即将到期

    def _adjust(self, vendor, status, delta):
        if status == "ACTIVE":
            self.active_count[vendor] = self.active_count.get(vendor, 0) + delta

    def add(self, cid, vendor, end, status="ACTIVE"):
        if cid in self.contracts:
            return False
        self.contracts[cid] = {"vendor": vendor, "status": status, "end": end, "version": 0}
        self._adjust(vendor, status, +1)
        insort(self.by_end, (end, cid))
        return True

    def remove(self, cid):
        c = self.contracts.pop(cid, None)
        if c is None:
            return False
        self._adjust(c["vendor"], c["status"], -1)
        self.by_end.remove((c["end"], cid))
        return True

    def update_status(self, cid, status, expected_version):
        """乐观锁：版本号不一致说明别人先改过，拒绝这次更新"""
        c = self.contracts.get(cid)
        if c is None or c["version"] != expected_version:
            return False
        self._adjust(c["vendor"], c["status"], -1)
        self._adjust(c["vendor"], status, +1)
        c["status"], c["version"] = status, c["version"] + 1
        return True

    def active_contracts(self, vendor):
        return self.active_count.get(vendor, 0)

    def expiring_within(self, today, days=30):
        """到期日在 [today, today + days] 内的有效合同，二分定位 O(log n + 结果数)。
        日期用整数天数表示（例如距 1970-01-01 的天数）；(d,) 比任何 (d, id) 都小"""
        lo = bisect_left(self.by_end, (today,))
        hi = bisect_left(self.by_end, (today + days + 1,))
        return [cid for _, cid in self.by_end[lo:hi] if self.contracts[cid]["status"] == "ACTIVE"]

Follow-up 参考思路：

• 即将到期：维护按到期日排序的索引（上面的 by_end，生产环境用数据库在 end_date 上建索引），二分找到时间范围。
• 自动续约：给合同加 auto_renew 标记，由定时任务扫描即将到期且需要续约的合同，更新到期日（同时更新索引），并记录续约历史。
• 并发修改：用乐观锁，每次更新带上读取时的 version，版本不一致就拒绝，让用户刷新后重试（上面的 update_status）。

第三轮：Bar Raiser（全程 BQ，约 45 分钟）

这一轮完全没有 Coding，三组问题，每个 story 都被连续深挖十分钟以上：

问题	追问方向
影响最大的一个项目	你具体做了什么、影响怎么衡量
一次失败经历	失败原因是什么？为什么当时没提前发现？团队有没有提过不同意见？事后具体采取了哪些改进？
一次 Ownership 的例子	哪些工作原本不属于你的职责？为什么决定主动承担？

整体感觉这一轮更像是在验证你过去做事情的方式，而不是验证技术能力。

第四轮：BQ + Return Request Processing System

BQ

1. 讲一次快速学习新技术并完成交付的经历。
2. 讲一次需求发生重大变化后如何调整项目计划。

Coding：Return Request Processing System

电商平台每天会收到大量退货申请，系统需要支持：

• 创建退货单、取消退货单
• 查询用户的历史退货记录
• 统计某个商品当前未处理的退货数量

from collections import defaultdict

class ReturnService:
    def __init__(self):
        self.requests = {}                       # rid -> {user, item, status}
        self.user_history = defaultdict(list)    # user -> [rid]，按创建顺序
        self.pending = defaultdict(int)          # item -> 未处理退货数

    def create(self, rid, user, item):
        if rid in self.requests:                 # 幂等：重复提交同一退货单直接忽略
            return False
        self.requests[rid] = {"user": user, "item": item, "status": "PENDING"}
        self.user_history[user].append(rid)
        self.pending[item] += 1
        return True

    def _close(self, rid, status):
        r = self.requests.get(rid)
        if r is None or r["status"] != "PENDING":
            return False
        r["status"] = status
        self.pending[r["item"]] -= 1
        return True

    def cancel(self, rid):
        return self._close(rid, "CANCELLED")

    def complete(self, rid):
        return self._close(rid, "COMPLETED")

    def history(self, user):
        return [(rid, self.requests[rid]["status"]) for rid in self.user_history[user]]

    def pending_count(self, item):
        return self.pending[item]

聊 AI（约 10 分钟）

Coding 结束后专门聊了十分钟 AI：

• 平时是否使用 Cursor 或 Claude？
• 如果 AI 给出了一段复杂代码，你会如何验证其正确性？
• 有没有遇到过 AI 给出的方案看起来合理、实际上有逻辑漏洞？最终是怎么发现的？

准备建议：提前准备一个真实例子，讲清楚你验证 AI 代码的方法，比如写测试用例、和暴力解对拍、读懂每一行再合入、code review。面试官想听的是你对代码质量负责的态度，而不是会不会用工具。

总结

• Coding：今年题目都是「HashMap / HashSet / 堆 + 业务包装」，难点不在算法，而在把需求拆成清晰的数据结构，并能讨论扩展、并发和复杂度。
• BQ：每个 story 都会被追问决策过程和细节，一定要准备真实、经得起深挖的例子，并能对应到具体的 Leadership Principles。
• AI：已经是高频话题，提前准备「如何验证 AI 生成代码」的回答。