期末 Scala 算法实现核心训练

题面

实现一个教学版 HLLLite。64-bit hash 的最高 p 位选择 bucket；剩余 q=64−p 位中，第一个 1 之前的零数加一为 rho。实现：

1. offerHashed 与 add；
2. 兼容性检查和 in-place merge；
3. raw HLL estimate，并在小基数时使用 linear counting；
4. 解释为何 MapReduce combiner/reducer 可以逐寄存器取 max。

hash 函数由题目提供。不同 sketch 只有在 p 与 hash 标识都相同时才允许 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 的正偏值。

评分点

完整 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")
  }
}

测试向量

复杂度、不变量与边界

• 时间：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 分钟策略

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

题面

1. 20 分：实现 out-of-place Cholesky。输入必须为实对称正定矩阵；输出下三角 L，使 A≈LL^T。
2. 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；负深度拒绝。

评分点

完整 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

• 时间 Θ(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 自己。

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

题面

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

1. 写纯 Scala Seq[String] => Map[String,Int]；
2. 写 Spark RDD[String] => RDD[(String,Int)]；
3. 说明 transformation/action、reduceByKey 与 groupByKey 的区别；
4. 给出正确 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"

评分点

完整 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()
    }
  }
}

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

• 本地时间：Θ(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 分钟策略

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

1. Spark batch gradient descent

接口：

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 为：

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。