项目1：百万歌曲基础数据与紧凑图BFS

1.边生成规则

仓库事实：每条基础数据记录中的 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。只有源数据也含反向关系时，反向边才存在。
• 无权：每条边的代价恒为1。距离表示边数，不表示音频相似度。
• 去重：相同 (src,dst) 只能保留一次。否则，当前边界计数、混洗数据量和性能结果都会失真。
• 允许自环但不产生新发现：仓库规格没有明确要求删除 A → A。已访问集合会过滤该自环。

需要区分：对 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 中每个顶点与源点的最短距离恰为 d−1。
2. 已访问集合不变量：第 d 轮结束后，visited 恰好包含距离不超过 d 的已发现顶点。
3. 互斥不变量：next_frontier ∩ visited_before = ∅。同一轮多个父节点发现同一目标时，目标也只能进入一次。
4. 前驱不变量：首次发现 v 时记录的前驱位于上一层。因此，沿前驱回溯会得到长度为发现深度的路径。

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

可验证推导：BFS按0、1、2、… 跳的顺序扩展。如果顶点 v 首次在深度 d 出现，但存在长度小于 d 的路径，则该路径的倒数第二个顶点会在更早的层中扩展，并更早发现 v。这与首次在深度 d 发现 v 矛盾。因此，首次发现深度就是最少跳数。

分布式实现对应关系

4.精确层计数与累计层计数不能混用

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

需要区分：“三层内的人数”是累计量。“第三层新发现的人数”是单层量。仓库的正式约束保存 exact_new_by_hop。不得用累计值与该字段比较。

1.正式MapReduce：每一跳直接扫描基础数据

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

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

每轮把 frontier.txt 和 visited.txt 作为缓存文件分发。映射器读取基础数据中的 artist_id 和 similar_artists。仅当源点位于当前边界时，映射器才输出未访问邻居。归约器按目标艺人键去重。

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

结果元数据强制：

• input_semantics = direct_foundation_scan_per_hop；
• tsv_fallback_used = false；
• Avro/Parquet的输入-格式类与路线一致；
• 深度为 D 时必须存在 D 个不同应用程序ID；
• exact_new_by_hop 必须等于冻结的正确性序列。

2.正式Spark：直接投影基础数据并缓存一次边表

仓库事实：Spark启动器的运行清单写入：

input_semantics = direct_foundation_projection_cached_then_dataframe_bfs
timing_scope = python_main_start_through_spark_stop

Spark在同一个应用中读取指定格式，并清洗和去重边。然后Spark调用 cache()。默认模式 eager_count 先物化边缓存，再开始逐跳循环。frontier 和 visited 默认通过HDFS写入后重读来物化。

3.旧基准测试脚本：格式标签只代表边的来源

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

• 供Spark使用的Parquet边表；
• 供Hadoop Streaming使用的TSV边文本。

旧MapReduce路线的真实输入是 avro_tsv 或 parquet_tsv。Python映射任务/归约任务在Hadoop Streaming中处理这些TSV边，而不是原生读取基础数据。

需要区分：旧脚本并非“错误代码”。它可以验证图语义和探索工程瓶颈；问题在于它不能回答正式的原生MR × physical格式比较。

1. cache() 是声明，不是Spark动作

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

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

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

可验证推导：若跳过 count()，第一个需要边的动作会一并承担首次计算成本。这样“BFS query时间”包含多少准备，取决于计时边界。仓库正式清单记录边物化模式，避免暗中改变口径。

需要区分：第二次Spark动作能否完全复用缓存，取决于分区是否已经物化以及分区是否仍在缓存中。调用 cache() 不表示数据会永久驻留。

2.多个Spark动作会重复触发未缓存的上游计算

简化版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

第一个动作会物化 next_frontier；后续动作可以复用它，但 visited 若只保留长血缘关系，仍可能重算上游。性能分析必须看动作、作业与阶段，而不能只数Python循环次数。

3.血缘关系为什么会增长

每一轮都构造：

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

若不截断血缘关系，第 d 轮的逻辑计划包含前 d−1 轮的并集/去重链。长血缘关系会增加计划、调度、序列化与故障重算成本。

仓库支持的物化策略

历史记录：仓库记录Spark 4.1 DataFrame检查点在该作业中遇到血缘关系问题，最终脚本采用每跳数将当前边界与已访问写入HDFS再读回的方式。

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

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

历史记录：当前性能文档中G4 “已缓存混合边” 为40.630 s，G2为40.172 s，被标为 NO EFFECT；G5预热边缓存为41.033 s，也没有改善。

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

1.为什么需要紧凑图

基础数据/Parquet适合存储与分析，却不是重复低延迟BFS的理想查询结构。每跳数扫列式表、展开数组、连接当前边界和混洗，会让调度与数据移动成本远大于位运算本身。

仓库事实：优化路径从规范基础数据Parquet派生紧凑产物。README记录一个50,996,454-字节产物，包含354,307艺人、约4.441百万有向边与998,795艺人-到歌曲链接。

注意：性能文档另列格式-v1稀疏位图为344,151艺人。两个艺人计数在同一提交中不一致，表明它们不是可无条件合并的同一产物；详见性能冲突章节。

2.稠密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

仓库事实：索引记录是小端序 <IB，共5字节；载荷值是小端序 uint32。因此每行偏移量上限为2³²−1，计数上限为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.复用相同的连续邻接片段

仓库事实：ContainmentArrayWriter 先收集每个来源的有序、无重复行，再按“长度降序、字典序”处理唯一行。若某行已作为载荷中连续子序列存在，则索引复用该偏移量；否则追加。

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

physical payload:
  (3, 1, 0)

logical_count = 5
physical_count = 3

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

需要区分：计数只有8位。单个艺人若有256个邻居，写入器必须抛出溢出；不能静默截断。

4. mmap与启动校验

Java CompactGraph.open 会：

1. 读取元数据；
2. 验证每个文件的实际字节与元数据相同；
3. 读取艺人/歌曲ID列表并核对计数；
4. 内存映射索引与载荷；
5. 逐行验证偏移量/计数不越界，稠密ID不超目录。

这些检查是性能数字的前置条件。若图校验和、文件大小或艺人计数不匹配，不能把运行称为同一产物的复现。

5.位图核心算法

仓库事实：Java核心算法为艺人全集分配三个 long[] 位集：

• visited：所有已发现艺人；
• frontier：当前精确跳数集合；
• 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个候选的集合并、差与计数可用一个机器字的OR、AND-NOT与 bitCount 完成；整数稠密ID避免大量字符串哈希/连接。

路径开销：计数-仅查询不分配距离/前驱；需要路径时才分配 byte[] distances 与 int[] predecessors。正式单次契约当前是 counts-only-v1，不能声称它输出了路径。

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

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

仓库权威性能规则：

• 输入、来源、跳数、集群布局、资源限制与应用程序生命周期必须相同。
• 慢实验和失败实验也要记录为 REGRESSION 或 INVALID。
• 小于正常抖动的差异标为 NO EFFECT。
• 使用计时前，先核对当前边界计数、行计数、模式与输出哈希。
• 单次Spark与持久化热点服务是不同工作负载类，不得当作等价延迟。
• 每个数字必须能追溯到命令、提交、Spark应用程序ID、配置快照与原始资源日志。

2.当前权威四路线记录

历史记录：pre/performance.md 明确自称当前权威性能记录。五名艺人各跑十跳数，得到：

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

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

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

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

毫秒级核心算法/服务延迟与几十秒全新应用程序墙钟同时成立，因为它们计时边界不同。核心算法只测图遍历；全新墙钟还含提交、资源分配、JVM、Spark上下文、产物分发与结果写出。

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墙钟更低但核心算法相同，收益主要来自作业融合与桥接减少。若精确计数不同，则该计时直接失效。