Scala 算法实现：三套 45 分自测训练

训练顺序与材料依据

使用边界：本页根据课程材料安排训练顺序。本页不预测任何考试内容。当前材料不包含正式题面，也不包含官方 45 分评分规则。

等级	材料依据	可使用的结论
A	当前材料提供完整接口、算法或计算要求。	可以据此建立完整实现训练。
B	当前材料提供可执行流程，但没有提供完整接口。	可以据此建立推导或半实现训练。
C	历史训练材料提供同类代码。	只能称为历史训练证据。
D	扩展资料提供相邻主题。	只能作为补充训练，不表示官方要求。

实现训练顺序
顺序	训练内容	依据等级	安排理由
1	HyperLogLog（HLL）	C	`prob_struct.html` 提供完整的短 Scala 实现。该实现适合训练状态、位运算、`merge` 和估计器。材料没有把它声明为官方考核要求。
2	Cholesky	A+B	当前材料提供 SPD 和 Cholesky 分解的公式、循环不变量和失败条件。
3	无权有向图 BFS	A	当前材料提供最短距离、路径和分布式当前边界的语义。
4	Scala Collection / Spark RDD 聚合	A	当前材料提供 Scala、WordCount 和 `flatMap → map → reduceByKey` 数据流。
5	Spark 批量梯度下降	B	当前材料提供梯度下降的数据流和通信要求，适合半实现训练。
6	PCA/SVD Gram	B	当前材料提供 Gram 矩阵、SVD、PCA 和高而窄矩阵聚合。
扩展	蓄水池抽样（Reservoir Sampling）/计数最小摘要（Count-Min Sketch，CMS）/布隆过滤器（Bloom filter）	D	可用于读码迁移；不应挤占前六项训练时间。

Scala 2.x 实现语法速查

三套参考实现采用 Scala 2.13 语法。Spark 自测训练固定使用 Scala 2.12.18 和 Apache Spark 3.5.1，因为 Spark 构件带有 Scala 二进制版本后缀。不得混合编译两个版本的类文件。

// Array 与二维数组
val a = Array(1.0, 2.0)
val matrix = Array.ofDim[Double](3, 3)
val copy = a.clone()

// 现场算法常用 mutable 容器
import scala.collection.mutable
val q = mutable.Queue.empty[(String, Int)]
val seen = mutable.Set.empty[String]
val parent = mutable.Map.empty[String, String]

// Option
def lookup(k: String): Option[Int] =
  if (k.nonEmpty) Some(k.length) else None

// 迭代器适合一次流过
val words = lines.iterator
  .flatMap(_.toLowerCase.split("[^a-z0-9]+"))
  .filter(_.nonEmpty)

六项静态自查

检查每个数组下标。确认下标范围与数组形状一致。
整数除法已显式转换为 Double。
循环更新的是当前状态，不是误用上一题变量。
检查 merge 和归约操作。确认操作满足算法要求的结合性。
在进入核心循环之前处理边界输入。
区分 Spark 转换和动作。区分驱动程序与执行器的状态位置。

自测训练 A（45 分）：流式概率结构 HLL

题面

实现教学版 HLLLite。64 位哈希值的最高 p 位选择桶。剩余 q=64−p 位中，第一个 1 之前的零数加一为 rho。实现以下接口：

实现 offerHashed 和 add。
实现兼容性检查和原地 merge。
实现未经修正的 HLL 估计值。小基数时使用线性计数修正。
说明 MapReduce 合并器和归约器为什么可以逐寄存器取最大值。

哈希函数由训练题提供。只有当两个摘要的 p 和哈希标识同时相同时，才允许执行 merge。

接口合同

final class HLLLite(
  val p: Int,
  val hashId: String,
  hash64: String => Long
) {
  def add(value: String): Unit
  def offerHashed(hash: Long): Unit
  def merge(that: HLLLite): Unit
  def estimate(): Double
  def registers: Vector[Int]
}

4 ≤ p ≤ 20，m=2^p。
offerHashed 每次只更新一个 register，并且 register 只能增大。
merge 对不同 p/hashId 必须抛 IllegalArgumentException。
空 sketch 的小基数修正必须返回 0，而不是 raw estimate 的正偏值。

自测检查项

部分	自测分值	可观察检查项
状态与构造	7	p 检查、m、零寄存器、hash 标识。
hash 拆分与 rho	14	无符号右移取 bucket；rho 有全零 remainder 上界；max 更新。
merge	8	兼容性；逐项 max；交换/结合/幂等。
estimate	9	调和和、alpha、Double、小基数 linear counting。
复杂度与不变量	7	add O(1)，merge/estimate O(m)，state O(m)，MapReduce 合并理由。
总分	45

完整 Scala 2.13 参考实现

以下单文件不依赖第三方库，可保存为 MockHLL.scala，用 Scala 2.13 编译。

final class HLLLite(
    val p: Int,
    val hashId: String,
    hash64: String => Long
) {
  require(p >= 4 && p <= 20, "p must be in [4, 20]")
  require(hashId.nonEmpty, "hashId must be non-empty")

  private val m: Int = 1 << p
  private val q: Int = 64 - p
  private val reg: Array[Int] = Array.fill(m)(0)

  def add(value: String): Unit = offerHashed(hash64(value))

  def offerHashed(hash: Long): Unit = {
    val bucket = (hash >>> q).toInt
    val shiftedRemainder = hash << p
    val rawRho = java.lang.Long.numberOfLeadingZeros(shiftedRemainder) + 1
    val rho = math.min(rawRho, q + 1)
    if (rho > reg(bucket)) reg(bucket) = rho
  }

  def merge(that: HLLLite): Unit = {
    require(this.p == that.p, "precision mismatch")
    require(this.hashId == that.hashId, "hash function mismatch")
    val other = that.registers
    var i = 0
    while (i < m) {
      reg(i) = math.max(reg(i), other(i))
      i += 1
    }
  }

  def estimate(): Double = {
    var inversePowerSum = 0.0
    var zeroRegisters = 0
    var i = 0
    while (i < m) {
      inversePowerSum += math.pow(2.0, -reg(i).toDouble)
      if (reg(i) == 0) zeroRegisters += 1
      i += 1
    }

    val alpha = m match {
      case 16 => 0.673
      case 32 => 0.697
      case 64 => 0.709
      case _  => 0.7213 / (1.0 + 1.079 / m.toDouble)
    }
    val raw = alpha * m.toDouble * m.toDouble / inversePowerSum

    if (raw <= 2.5 * m && zeroRegisters > 0)
      m.toDouble * math.log(m.toDouble / zeroRegisters.toDouble)
    else
      raw
  }

  def registers: Vector[Int] = reg.toVector
}

object MockHLL {
  private def hashed(p: Int, bucket: Int, rho: Int): Long = {
    val q = 64 - p
    require(bucket >= 0 && bucket < (1 << p))
    require(rho >= 1 && rho <= q)
    (bucket.toLong << q) | (1L << (q - rho))
  }

  def main(args: Array[String]): Unit = {
    val noHash: String => Long = _ => 0L
    val left = new HLLLite(4, "test-v1", noHash)
    left.offerHashed(hashed(4, 0, 2))
    left.offerHashed(hashed(4, 0, 4))
    left.offerHashed(hashed(4, 3, 1))
    assert(left.registers(0) == 4)
    assert(left.registers(3) == 1)

    val right = new HLLLite(4, "test-v1", noHash)
    right.offerHashed(hashed(4, 1, 5))
    right.offerHashed(hashed(4, 3, 3))
    left.merge(right)
    assert(left.registers(0) == 4)
    assert(left.registers(1) == 5)
    assert(left.registers(3) == 3)

    val before = left.registers
    left.offerHashed(hashed(4, 3, 3))
    assert(left.registers == before)

    val empty = new HLLLite(4, "test-v1", noHash)
    assert(empty.estimate() == 0.0)
    println("MockHLL tests passed")
  }
}

测试向量

输入	预期	检查目的
同 bucket 的 rho 2、4	register 保留 4	max 不变量。
bucket 3 的 rho 1	只更新 register 3	bucket 拆分。
merge 两个 sketch	逐项 max	union summary。
重复 offer 同一 hash	状态不变	幂等。
空 sketch	estimate = 0	linear counting。
p/hashId 不同	拒绝 merge	兼容性。

复杂度、不变量与边界

时间：add 为 O(1)；merge 与 estimate 为 O(m)。
空间：O(m)，m=2^p。
核心不变量：M[j] 是所有落入 bucket j 的观测中最大 rho。
merge 代数：逐项 max 交换、结合、幂等，所以适合 combiner/reducer。
误差：典型相对标准误差约 1.04/√m；这是概率估计，不是逐输入确定误差界。
全零 remainder：rho 必须封顶为 q+1，避免超出可表示范围。
hash 前提：质量依赖近似均匀的 64-bit hash；示例的测试 hash 只用于确定性单元测试。

45 分钟策略

0–5 分钟：写 p、m、register、接口和兼容性。
5–18 分钟：完成 bucket/rho/max，并先写全零 remainder 上界。
18–26 分钟：完成 merge，口述交换/结合/幂等。
26–36 分钟：完成 raw estimate 与 linear counting。
36–42 分钟：跑三个手工 hash 状态。
42–45 分钟：补复杂度与“同一 hashId”边界。

自测训练 B（45 分）：Cholesky + BFS

题面

20 分：实现 out-of-place Cholesky。输入必须为实对称正定矩阵；输出下三角 L，使 A≈LL^T。
25 分：对无权有向邻接表实现深度受限 BFS，返回一条最短路径；再写出 Spark frontier 一步的关系代数。

接口合同

def cholesky(
  a: Array[Array[Double]],
  eps: Double = 1e-12
): Array[Array[Double]]

def shortestPath(
  adjacency: Map[String, Seq[String]],
  source: String,
  target: String,
  maxDepth: Int
): Option[List[String]]

Cholesky：空矩阵返回空；非方阵、不对称或 pivot residual≤eps 时拒绝。
BFS：source==target 返回单点路径；缺失邻接视为无出边；不可达为 None；负深度拒绝。

自测检查项

部分	自测分值	检查项
Cholesky 公式与循环	10	对角 residual、sqrt、下方 dot product/division。
验证与复杂度	5	square/symmetry/SPD、Θ(n³)。
测试与重构	5	3×3 例、非 SPD 失败、LLᵀ。
BFS 状态	14	queue、入队即 seen、parent、first discovery。
边界与回溯	6	深度、同点、不可达、环。
Spark frontier 语义	5	join、dedup、left-anti、visited union。

完整 Scala 2.13 参考实现

以下单文件不依赖第三方库，可保存为 MockLinearGraph.scala。

import scala.collection.mutable

object MockLinearGraph {
  def cholesky(
      a: Array[Array[Double]],
      eps: Double = 1e-12
  ): Array[Array[Double]] = {
    require(eps >= 0.0, "eps must be nonnegative")
    val n = a.length
    require(a.forall(_.length == n), "square matrix required")

    var i = 0
    while (i < n) {
      var j = 0
      while (j < n) {
        require(
          math.abs(a(i)(j) - a(j)(i)) <= eps,
          "symmetric matrix required"
        )
        j += 1
      }
      i += 1
    }

    val l = Array.ofDim[Double](n, n)
    var column = 0
    while (column < n) {
      var diagonalSum = 0.0
      var k = 0
      while (k < column) {
        diagonalSum += l(column)(k) * l(column)(k)
        k += 1
      }

      val residual = a(column)(column) - diagonalSum
      require(residual > eps, "matrix is not SPD")
      l(column)(column) = math.sqrt(residual)

      i = column + 1
      while (i < n) {
        var cross = 0.0
        k = 0
        while (k < column) {
          cross += l(i)(k) * l(column)(k)
          k += 1
        }
        l(i)(column) =
          (a(i)(column) - cross) / l(column)(column)
        i += 1
      }
      column += 1
    }
    l
  }

  def shortestPath(
      adjacency: Map[String, Seq[String]],
      source: String,
      target: String,
      maxDepth: Int
  ): Option[List[String]] = {
    require(maxDepth >= 0, "maxDepth must be nonnegative")
    if (source == target) return Some(List(source))

    val queue = mutable.Queue.empty[(String, Int)]
    val seen = mutable.Set.empty[String]
    val parent = mutable.Map.empty[String, String]
    queue.enqueue((source, 0))
    seen += source

    while (queue.nonEmpty) {
      val (u, depth) = queue.dequeue()
      if (depth < maxDepth) {
        val neighbors = adjacency.getOrElse(u, Seq.empty).distinct.sorted
        neighbors.foreach { v =>
          if (!seen(v)) {
            seen += v
            parent(v) = u
            if (v == target)
              return Some(reconstruct(parent.toMap, source, target))
            queue.enqueue((v, depth + 1))
          }
        }
      }
    }
    None
  }

  private def reconstruct(
      parent: Map[String, String],
      source: String,
      target: String
  ): List[String] = {
    var path = List(target)
    var current = target
    while (current != source) {
      current = parent(current)
      path = current :: path
    }
    path
  }

  private def multiplyLLT(l: Array[Array[Double]]): Array[Array[Double]] = {
    val n = l.length
    val out = Array.ofDim[Double](n, n)
    var i = 0
    while (i < n) {
      var j = 0
      while (j < n) {
        var k = 0
        while (k < n) {
          out(i)(j) += l(i)(k) * l(j)(k)
          k += 1
        }
        j += 1
      }
      i += 1
    }
    out
  }

  private def close(x: Double, y: Double): Boolean =
    math.abs(x - y) <= 1e-9

  def main(args: Array[String]): Unit = {
    val a = Array(
      Array(25.0, 15.0, -5.0),
      Array(15.0, 18.0, 0.0),
      Array(-5.0, 0.0, 11.0)
    )
    val l = cholesky(a)
    val expected = Array(
      Array(5.0, 0.0, 0.0),
      Array(3.0, 3.0, 0.0),
      Array(-1.0, 1.0, 3.0)
    )
    for (i <- a.indices; j <- a.indices)
      assert(close(l(i)(j), expected(i)(j)))
    val rebuilt = multiplyLLT(l)
    for (i <- a.indices; j <- a.indices)
      assert(close(rebuilt(i)(j), a(i)(j)))

    val graph = Map(
      "A" -> Seq("B", "C"),
      "B" -> Seq("D"),
      "C" -> Seq("D"),
      "D" -> Seq("E")
    )
    assert(shortestPath(graph, "A", "E", 3).contains(List("A", "B", "D", "E")))
    assert(shortestPath(graph, "A", "E", 2).isEmpty)
    assert(shortestPath(Map("A" -> Seq("A", "B")), "A", "B", 1)
      .contains(List("A", "B")))
    assert(shortestPath(graph, "A", "A", 0).contains(List("A")))
    println("MockLinearGraph tests passed")
  }
}

测试向量

算法	输入	预期
Cholesky	3×3 SPD 示例矩阵	`[[5,0,0],[3,3,0],[-1,1,3]]`
Cholesky	`[[25,-50],[-50,101]]`	`[[5,0],[-10,1]]`
Cholesky	`[[1,2],[2,1]]`	第二 pivot 失败。
Cholesky	`[[1,2],[0,1]]`	不对称拒绝。
BFS	A→B/C→D→E，depth 3	A-B-D-E（排序固定 tie）。
BFS	同图，depth 2	`None`。
BFS	A→A、A→B	终止并返回 A-B。

复杂度与循环不变量

Cholesky

时间 Θ(n³)，输出状态 Θ(n²)，主导 flop 约 n³/3。
完成第 j 列后，前 j+1 个 leading rows/columns 已满足对应的 A=LL^T 元素方程。
上三角保持 0；每个已写 diagonal 严格为正。

BFS

邻接表时间 Θ(V+E)，空间 Θ(V)；深度限制时只遍历半径内子图。
queue 中 depth 非降；顶点入队时立刻 seen，因此首次发现层即最短距离。
parent(v) 指向上一层，回溯路径无环。

Spark frontier 一步

// 关系语义；列名按实际 schema 替换
val candidates = frontier
  .join(edges, frontier("artist") === edges("src"))
  .select(frontier("source"), edges("dst").as("artist"))
  .dropDuplicates("source", "artist")

val next = candidates.join(
  visited,
  Seq("source", "artist"),
  "left_anti"
)

val visitedNext = visited.union(next).dropDuplicates("source", "artist")

以上是 DataFrame 关系骨架，不是本地 BFS 的必要依赖。核心顺序是 join → 轮内去重 → left-anti visited → union。

边界条件与 45 分钟策略

空 Cholesky 输入返回 0×0；ragged rows 必须拒绝。
对称性检查不是 SPD 证明；pivot residual 才在分解过程中验证正定。
BFS 在入队时 seen，不能等 dequeue 才标记，否则同层重复入队。
maxDepth=0 只允许 source 自己。

0–15 分钟：Cholesky shape、验证、两条公式与循环。
15–20 分钟：手算 3×3 的第一、二列检查。
20–35 分钟：BFS queue/seen/parent/depth。
35–40 分钟：回溯与环/同点/不可达。
40–45 分钟：Spark 四步语义与复杂度。

自测训练 C（45 分）：Scala Collection + Spark RDD

题面

输入日志每行包含自然语言 token。实现大小写无关词频，去除非字母数字分隔产生的空 token，只保留次数至少 2 的词：

写纯 Scala Seq[String] => Map[String,Int]；
写 Spark RDD[String] => RDD[(String,Int)]；
说明 transformation/action、reduceByKey 与 groupByKey 的区别；
给出正确 persist 位置与小型测试。

接口与 Spark 依赖合同

def localCounts(lines: Seq[String]): Map[String, Int]
def counts(logs: RDD[String]): RDD[(String, Int)]

参考实现固定：

Scala 2.12.18；
Apache Spark 3.5.1；
sbt："org.apache.spark" %% "spark-core" % "3.5.1" % "provided"；
本地测试运行时不能把 dependency 标成 provided，或需由 spark-submit 提供 Spark runtime。

// build.sbt
ThisBuild / scalaVersion := "2.12.18"
libraryDependencies +=
  "org.apache.spark" %% "spark-core" % "3.5.1" % "provided"

自测检查项

部分	自测分值	检查项
tokenize/normalize	8	小写、正则、过滤空 token、规则一致。
RDD 键值处理流程	9	`flatMap`、`(word, 1)`、`reduceByKey`。
filter/action	10	阈值过滤；不在算法函数中 collect 全量。
lazy/persist	7	action 触发；复用 base RDD 才 persist。
Collection 版本	6	foldLeft/mutable 计数正确。
复杂度/shuffle	5	线性 token 化；map-side combine；避免 materialize values。

完整 Spark/Scala 2.12 参考实现

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object MockWordCount {
  private val SplitPattern = "[^a-z0-9]+"

  private def tokens(line: String): Iterator[String] =
    line.toLowerCase
      .split(SplitPattern)
      .iterator
      .filter(_.nonEmpty)

  def localCounts(lines: Seq[String]): Map[String, Int] =
    lines.iterator
      .flatMap(tokens)
      .foldLeft(Map.empty[String, Int]) { (acc, word) =>
        acc.updated(word, acc.getOrElse(word, 0) + 1)
      }
      .filter { case (_, count) => count >= 2 }

  def counts(logs: RDD[String]): RDD[(String, Int)] =
    logs
      .flatMap(tokens)
      .map(word => (word, 1))
      .reduceByKey(_ + _)
      .filter { case (_, count) => count >= 2 }

  def main(args: Array[String]): Unit = {
    val input = Seq("To be, or not to be", "to")
    assert(localCounts(input) == Map("to" -> 3, "be" -> 2))
    assert(localCounts(Seq.empty).isEmpty)

    val conf = new SparkConf()
      .setAppName("MockWordCount")
      .setMaster("local[2]")
    val sc = new SparkContext(conf)
    try {
      val logs = sc.parallelize(input, 2).persist()
      val actual = counts(logs).collect().toMap
      assert(actual == Map("to" -> 3, "be" -> 2))
      logs.unpersist(blocking = false)
      println("MockWordCount tests passed")
    } finally {
      sc.stop()
    }
  }
}

测试向量、复杂度与不变量

输入	阈值后输出	覆盖
`"To be, or not to be", "to"`	`to→3, be→2`	大小写、标点、阈值。
空 Seq/RDD	空	空输入。
`"a---a", "A"`	`a→3`	连续分隔、小写。

本地时间：Θ(total characters + tokens)，状态 Θ(unique words)。
RDD：map 端扫描 token；reduceByKey 可先做 map-side combine，再按 key shuffle。
不变量：每个规范化 token 贡献一个 (word,1)；同 key 的和值等于全局出现次数。
为什么不用 groupByKey：它先搬运并保存同一 key 的全部 1，而求和只需可结合的局部部分和。
action：collect 只用于小测试；生产输出应 saveAsTextFile 或交给下游。
persist：只有同一个 RDD 会被多个 action/迭代复用时才有收益；persist 本身仍是 lazy。

45 分钟策略

0–6 分钟：固定 token 规则和接口。
6–15 分钟：完成本地 iterator/foldLeft。
15–25 分钟：平移为 flatMap/map/reduceByKey/filter。
25–32 分钟：写 transformation/action 与 groupByKey 对比。
32–39 分钟：加入 persist 使用点和小型 local Spark test。
39–45 分钟：空输入、标点、复杂度与 collect 边界。

补充训练：批量梯度下降与 PCA/SVD Gram 矩阵

这两项适合训练半实现、推导、复杂度和 Spark 数据流。本页的安排不表示官方考核要求。

1. Spark 批量梯度下降

接口：

def batchGD(
  data: RDD[(Array[Double], Double)],
  initial: Array[Double],
  learningRate: Double,
  iterations: Int
): Array[Double]

平方误差梯度：g=(1/n)Σ(x^Tw−y)x。每一轮所有样本必须读取同一个 w_t；driver 只更新一次得到 w_t+1。

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel

def batchGD(
    data: RDD[(Array[Double], Double)],
    initial: Array[Double],
    learningRate: Double,
    iterations: Int
): Array[Double] = {
  require(learningRate > 0.0, "learningRate must be positive")
  require(iterations >= 0, "iterations must be nonnegative")
  val dimension = initial.length
  val cached = data.persist(StorageLevel.MEMORY_AND_DISK)
  val count = cached.count()
  require(count > 0L, "data must be nonempty")
  require(cached.filter(_._1.length != dimension).take(1).isEmpty,
    "feature dimension mismatch")

  var weights = initial.clone()
  var iteration = 0
  while (iteration < iterations) {
    val broadcast = data.context.broadcast(weights)
    val gradient = cached.treeAggregate(Array.fill(dimension)(0.0))(
      seqOp = (sum, row) => {
        val (x, y) = row
        val w = broadcast.value
        var prediction = 0.0
        var j = 0
        while (j < dimension) {
          prediction += x(j) * w(j)
          j += 1
        }
        val error = prediction - y
        j = 0
        while (j < dimension) {
          sum(j) += error * x(j)
          j += 1
        }
        sum
      },
      combOp = (left, right) => {
        var j = 0
        while (j < dimension) {
          left(j) += right(j)
          j += 1
        }
        left
      }
    )
    broadcast.destroy(blocking = false)
    var j = 0
    while (j < dimension) {
      weights(j) -= learningRate * gradient(j) / count.toDouble
      j += 1
    }
    iteration += 1
  }
  cached.unpersist(blocking = false)
  weights
}

复杂度：每轮 Θ(nd)，总 Θ(Tnd)；每轮 broadcast d 个参数并 reduce d 维梯度。学习率过大不保证 loss 下降。

测试：一维点 (1→1),(2→2)、w0=0、小正学习率时，第一轮 gradient 为负，w 应向 1 增大；全零 feature 的 gradient 为 0。

2. Gram：PCA/SVD 的小矩阵聚合

对 rows x_i∈ℝ^d，Gram 为：

G=X^TX=Σ_ix_ix_i^T

def gram(rows: Iterable[Array[Double]]): Array[Array[Double]] = {
  val iterator = rows.iterator
  if (!iterator.hasNext) return Array.empty[Array[Double]]
  val first = iterator.next()
  val dimension = first.length
  val out = Array.ofDim[Double](dimension, dimension)

  def add(row: Array[Double]): Unit = {
    require(row.length == dimension, "row length mismatch")
    var j = 0
    while (j < dimension) {
      var k = j
      while (k < dimension) {
        out(j)(k) += row(j) * row(k)
        k += 1
      }
      j += 1
    }
  }

  add(first)
  iterator.foreach(add)
  var j = 0
  while (j < dimension) {
    var k = j + 1
    while (k < dimension) {
      out(k)(j) = out(j)(k)
      k += 1
    }
    j += 1
  }
  out
}

测试：rows [1,2]、[3,4] 应得 [[10,14],[14,20]]。

不变量：处理任意前缀后，G=Σx_ix_i^T，因此 G 对称且半正定。只计算上三角时必须恰好镜像一次。

复杂度：dense 时间 Θ(nd²)，状态 Θ(d²)。当 d² 无法放入单机内存时，这条基线不成立。Spark 版可发出上三角 key ((j,k), x(j)*x(k)) 并 reduceByKey，但会产生 Θ(nd²) pairs；稀疏数据应只枚举非零 pair。

训练顺序、边界与静态核查

三轮最小训练

计时 45 分钟完成训练 A。检查 rho 上界、逐项 max、Double 和空摘要。
计时 45 分钟完成训练 B。先手算一列 Cholesky。BFS 使用 parent，不要复制整条路径。
计时 45 分钟完成训练 C。标注每个转换和动作，并解释混洗。

代码静态核查清单

代码	接口	括号/结构	关键类型	依赖边界
HLL	完整	单文件 object/class	Long unsigned shift、Double	无第三方依赖
Cholesky/BFS	完整	单文件 object	Array[Double]、Option[List]	Scala 标准库
WordCount	完整	单文件 object	RDD[(String,Int)]	Spark 3.5.1 / Scala 2.12.18
Batch GD	核心函数完整	可放入 Spark object	RDD、broadcast、Array[Double]	同上
Gram	完整	标准库函数	Iterable、二维 Array	无第三方依赖

最终完成判定

能不看资料写出 HLL 的状态、更新、合并和估计，并解释合并操作的代数性质。
能写 Cholesky 两个元素公式、循环不变量和非 SPD 失败。
能写 BFS 的 seen 更新时机、深度限制、parent 回溯与 Spark 前沿集合的关系。
能把同一按键聚合在 Seq 与 RDD 间转换，并说明惰性求值、persist 与混洗。
能读写批量梯度下降的“广播变量 → 梯度聚合 → 驱动进程更新”过程。
能写 Gram 前缀不变量、对称镜像与 Θ(nd²) 边界。
能明确说出：候选排序是训练判断，不是课程出题事实。