Project 1：从百万歌曲 Foundation 到 Compact BFS

1. 边生成规则

仓库事实：每条 foundation 记录中的 artist_id 是源点；similar_artists 数组中的每个非空元素生成一条出边。

foundation row:
  artist_id = "A"
  similar_artists = ["B", "C", "B", null, " "]

cleaned directed edges:
  ("A", "B")
  ("A", "C")

仓库 src/2/foundation_graph.py 中的可执行 PySpark 边投影为：

from pyspark.sql import functions as F

edges = (
    foundation.select(
        F.trim(F.col("artist_id")).alias("src_artist_id"),
        F.explode_outer("similar_artists").alias("dst_artist_id"),
    )
    .select(
        "src_artist_id",
        F.trim(F.col("dst_artist_id")).alias("dst_artist_id"),
    )
    .where("src_artist_id IS NOT NULL AND src_artist_id <> ''")
    .where("dst_artist_id IS NOT NULL AND dst_artist_id <> ''")
    .dropDuplicates(["src_artist_id", "dst_artist_id"])
)

说明：以上是从仓库函数中抽出的可执行 PySpark（只省略读取与按源点重分区），不是 Scala 源码。仓库实现为 PySpark 与 Java；本页不会虚构 Scala 实现。

• 有向：A → B 不推出 B → A。只有源数据也含反向关系时，反向边才存在。
• 无权：每条边 hop cost 恒为 1。距离是边数，不是音频相似度。
• 去重：相同 (src,dst) 只能保留一次；否则 frontier count、shuffle 量和性能都会失真。
• 允许自环但不产生新发现：spec 没有显式要求删除 A → A。visited 集会挡住它。

考试陷阱：对 src_artist_id 单列去重会错误删除一个艺人的其他邻居；必须对二元组 (src,dst) 去重。

2. 队列版 BFS 伪代码

BFS(G, source, target, maxDepth):
    visited := {source}
    predecessor[source] := NONE
    frontier := [source]

    if source = target:
        return distance 0, path [source]

    for depth := 1 .. maxDepth:
        next := empty sequence
        for u in frontier:
            for each directed neighbor v in G[u]:
                if v not in visited:
                    visited.add(v)
                    predecessor[v] := u
                    next.append(v)
                    if v = target:
                        return depth, reconstruct(predecessor, target)
        if next is empty:
            break
        frontier := next

    return unreachable

3. 四个核心不变量

1. 层不变量：进入第 d 轮时，frontier 中每个顶点与 source 的最短距离恰为 d−1。
2. visited 不变量：第 d 轮结束后，visited 恰包含距离不超过 d 的已发现顶点。
3. 互斥不变量：next_frontier ∩ visited_before = ∅。同一轮多个父节点发现同一目标时，目标也只能进入一次。
4. 前驱不变量：首次发现 v 时记录的 predecessor 位于上一层，因此沿 predecessor 回溯必然得到长度为发现深度的路径。

为什么第一次发现就是最短路

可验证推导：BFS 按 0、1、2、… hop 的顺序扩展。若顶点 v 首次在深度 d 出现，却存在长度小于 d 的路径，则该路径的倒数第二个顶点应在更早层被扩展，并更早发现 v，矛盾。因此首次发现深度就是最短 hop 数。

分布式实现对应关系

4. Exact-hop 与 within-hop 不能混用

若 exact[d] 表示恰在第 d 跳首次发现的节点数，则：

考试陷阱：“三跳内人数”是累计量；“第三跳新发现人数”是单层量。仓库 formal contract 保存 exact_new_by_hop，不能拿累计值与它比较。

1. 正式 MapReduce：每 hop 直接扫描 foundation

仓库事实：FoundationMrJob 根据格式选择原生输入类：

if (format.equals("avro")) {
    job.setInputFormatClass(AvroKeyInputFormat.class);
} else if (format.equals("parquet")) {
    job.setInputFormatClass(AvroParquetInputFormat.class);
}

每轮把 frontier.txt 与 visited.txt作为 cache files 分发。mapper 读取 foundation 的 artist_id 与 similar_artists；只有 source 位于当前 frontier 时才发出未访问邻居。reducer 以目标 artist 为 key 去重。

for depth = 1..D:
    write frontier and visited
    launch native Avro/Parquet MapReduce job
    scan every foundation record
    mapper: if record.artist_id in frontier:
                emit each nonempty unvisited similar_artist
    reducer: emit each destination artist once
    next := job output
    visited := visited ∪ next

结果 metadata 强制：

• input_semantics = direct_foundation_scan_per_hop；
• tsv_fallback_used = false；
• Avro/Parquet 的 input-format class 与路线一致；
• 深度为 D 时必须存在 D 个不同 application ID；
• exact_new_by_hop 必须等于冻结的正确性序列。

2. 正式 Spark：直接投影 foundation，一次缓存边

仓库事实：Spark launcher 的 manifest 写入：

input_semantics = direct_foundation_projection_cached_then_dataframe_bfs
timing_scope = python_main_start_through_spark_stop

它在同一个 application 中读取指定格式、清洗并去重边，然后 cache()。默认使用 eager_count 先物化 edge cache，再开始 hop 循环；frontier 与 visited 默认经 HDFS roundtrip 物化。

3. 旧 benchmark 脚本：格式标签只代表边的来源

历史实现：run-task2-expansion-benchmark.sh 先调用 build_edges.py，从 Avro 或 Parquet foundation 生成：

• 供 Spark 使用的 Parquet edge table；
• 供 Hadoop Streaming 使用的 TSV edge text。

旧 MapReduce 路线的真实输入是 avro_tsv 或 parquet_tsv。Python mapper/reducer 在 Hadoop Streaming 中处理这些 TSV 边，而不是原生读取 foundation。

考试陷阱：旧脚本并非“错误代码”。它可以验证图语义和探索工程瓶颈；问题在于它不能回答正式的 native MR × physical format 比较。

1. cache() 是声明，不是 action

edges = read_foundation_edges(...).cache()

# lazy_cache：到这里还没有读取/去重完整 foundation

edges.count()
# eager_count：触发 projection、dedup、shuffle，并填充 cache

可验证推导：若跳过 count()，第一个需要 edges 的 action 会一并承担首次计算成本。这样“BFS query time”包含多少 preparation，取决于计时边界。仓库正式 manifest 记录 edge materialization mode，避免暗中改变口径。

考试陷阱：第二次 action 是否完全命中 cache 取决于分区是否成功物化且未被驱逐。写了 cache() 不等于数据永久驻留。

2. 多个 action 会重复触发未缓存血缘

简化版 Spark BFS 常在同一轮调用：

next_frontier = expand(...).cache()
hit = next_frontier.filter(is_target).limit(1).collect()  # action
next_count = next_frontier.count()                        # action
visited_count = visited.count()                           # action

第一个 action 会物化 next_frontier；后续 action 可以复用它，但 visited 若只保留长 lineage，仍可能重算上游。性能分析必须看 action、job 与 stage，而不能只数 Python 循环次数。

3. 为什么 lineage 会增长

每一轮都构造：

visited_d = dedup(visited_(d-1) UNION frontier_d)

若不截断 lineage，第 d 轮的逻辑计划包含前 d−1 轮的 union/dedup 链。长 lineage 会增加计划、调度、序列化与故障重算成本。

仓库支持的物化策略

历史记录：仓库记录 Spark 4.1 DataFrame checkpoint 在该作业中遇到 lineage 问题，最终脚本采用每 hop 将 frontier 与 visited 写入 HDFS再读回的方式。

权衡：HDFS roundtrip 增加 I/O，却建立明确迭代边界并避免长 lineage。是否更快必须测量；它首先是一项可复现性与稳定性决策。

4. 缓存优化为何可能没有效果

历史记录：当前性能文档中 G4 “cached hybrid edges” 为 40.630 s，G2 为 40.172 s，被标为 NO EFFECT；G5 warm edge cache 为 41.033 s，也没有改善。

可能原因包括：边读取已不是主瓶颈；额外 action/物化抵消收益；cache 已在先前路径中复用；测量差异落在正常噪声内。正确结论是“在该匹配 workload 下无测得效果”，不是“cache 永远无用”。

1. 为什么需要 compact graph

Foundation/Parquet 适合存储与分析，却不是重复低延迟 BFS 的理想查询结构。每 hop 扫列式表、explode 数组、join frontier 和 shuffle，会让调度与数据移动成本远大于位运算本身。

仓库事实：优化路径从 canonical foundation Parquet 派生 compact artifact。README 记录一个 50,996,454-byte artifact，包含 354,307 artists、约 4.441 million directed edges 与 998,795 artist-to-song links。

注意：性能文档另列 format-v1 sparse bitmap 为 344,151 artists。两个 artist count 在同一 commit 中不一致，表明它们不是可无条件合并的同一 artifact；详见性能冲突章节。

2. Dense ID 与六个数据文件

artist_ids.txt          dense artist index → original artist_id
song_ids.txt            dense song index   → original song_id

artist_index.bin        per artist: <uint32 offset, uint8 count>
artist_neighbors.u32    packed dense neighbor IDs

song_index.bin          per artist: <uint32 offset, uint8 count>
artist_songs.u32        packed dense song IDs

仓库事实：index record 是 little-endian <IB，共 5 bytes；payload value 是 little-endian uint32。因此每行 offset 上限为 2³²−1，count 上限为 255。

artist_index.bin
dense artist 0 ── (offset=0, count=3) ─────────┐
dense artist 1 ── (offset=1, count=2) ──────┐  │
                                             ▼  ▼
artist_neighbors.u32 payload:               [3, 1, 0]
artist 0 neighbors = payload[0:3] = [3,1,0]
artist 1 neighbors = payload[1:3] =   [1,0]

第二行复用第一行的后缀，不再复制 [1,0]。

3. Containment compaction

仓库事实：ContainmentArrayWriter 先收集每个 source 的有序、无重复 row，再按“长度降序、字典序”处理唯一 row。若某 row 已作为 payload 中连续子序列存在，则 index 复用该 offset；否则追加。

logical rows:
  row 0 = (3, 1, 0)
  row 1 = (1, 0)

physical payload:
  (3, 1, 0)

logical_count = 5
physical_count = 3

可验证推导：复用连续子序列保持 row 的顺序与内容完全不变，因此 BFS 邻居枚举语义不变。它不是集合近似压缩，也不允许漏边。

考试陷阱：count 只有 8 bit。单个 artist 若有 256 个邻居，writer 必须抛出 overflow；不能静默截断。

4. mmap 与启动校验

Java CompactGraph.open 会：

1. 读取 metadata；
2. 验证每个文件的实际 bytes 与 metadata 相同；
3. 读取 artist/song ID 列表并核对 count；
4. 内存映射 index 与 payload；
5. 逐行验证 offset/count 不越界，dense ID 不超 catalog。

这些检查是性能数字的前置条件。若 graph checksum、文件大小或 artist count 不匹配，不能把运行称为同一 artifact 的复现。

5. Bitmap kernel

仓库事实：Java kernel 为 artist universe 分配三个 long[] 位集：

• visited：所有已发现 artist；
• frontier：当前 exact-hop 集合；
• next：下一层候选。

frontier := {source}
visited  := {source}
exact[0] := 1

for depth = 1..maxDepth:
    clear(next)
    for each set artist bit in frontier:
        for each 64-bit adjacency word (block, word):
            discovered := word AND NOT visited[block] AND NOT next[block]
            next[block] := next[block] OR word
            optionally record distance/predecessor for discovered bits

    for each block:
        next[block]    := next[block] AND NOT visited[block]
        visited[block] := visited[block] OR next[block]
        exact[depth]  += popcount(next[block])

    swap(frontier, next)

为什么快：64 个候选的集合并、差与计数可用一个 machine word 的 OR、AND-NOT 与 bitCount 完成；整数 dense ID 避免大量字符串 hash/join。

路径开销：counts-only 查询不分配 distances/predecessors；需要路径时才分配 byte[] distances 与 int[] predecessors。正式 one-shot contract 当前是 counts-only-v1，不能声称它输出了路径。

边界：底层 kernel 接受深度 0..15；正式 one-shot CLI 只允许 1..10。接口契约比内核能力更窄，调用者必须遵守 CLI 契约。

1. 正式比较必须固定的测量合同

仓库权威性能规则：

• 输入、source、hop、集群布局、资源限制与 application lifecycle 必须相同。
• 慢实验和失败实验也要记录为 REGRESSION 或 INVALID。
• 小于正常抖动的差异标为 NO EFFECT。
• 使用 timing 前，先核对 frontier counts、row counts、schema 与 output hash。
• one-shot Spark 与 persistent hot service 是不同 workload class，不得当作等价延迟。
• 每个数字必须能追溯到 command、commit、Spark application ID、配置快照与 raw resource log。

2. 当前权威四路线记录

历史记录：pre/performance.md 明确自称当前 authoritative performance record。五名艺人各跑十 hop，得到：

在该受控实验中选择 Spark + Parquet。正确解读是：“在此 workload 与资源口径下，它的均值最低”；不是“Parquet 对所有 BFS 永远最快”。

3. 同一 commit 中的历史数字冲突

未给定：外部 raw evidence 位于 /4721/project/perf_runs/，不在 Git 中；本页无法仅凭仓库重建每个历史数字的完整运行链。因此只报告仓库声明，不把它们重新认证为本机实测。

4. 这些数字为什么不能放在同一柱状图

毫秒级 kernel/service latency 与几十秒 fresh application wall 同时成立，因为它们计时边界不同。kernel 只测图遍历；fresh wall 还含提交、资源分配、JVM、Spark context、artifact 分发与结果写出。

5. D1/D2 与性能归因

比较 D1/D2 时应至少同时记录：

launcher_process_wall_nanos
query_wall_nanos
kernel_nanos
executor_bridge_nanos
job_ids
execution_steps
resource_envelope
exact_new_by_hop

若 D2 wall 更低但 kernel 相同，收益主要来自 job 融合与 bridge 减少。若 exact counts 不同，则该 timing 直接失效。