def scalding()

in src/main/scala/com/spotify/bdrc/pipeline/PageRank.scala [34:54]

• 18 lines of code
• 3 McCabe index (conditional complexity)

  def scalding(input: TypedPipe[(String, String)]): TypedPipe[(String, Double)] = {
    val links = input.group.toList // (src URL, list of dst URL)
    var ranks = input.keys.distinct.map((_, 1.0)) // (src URL, 1.0)

    for (i <- 1 to 10) {
      val contribs = links
        .join(ranks)
        .toTypedPipe
        .values
        // re-distribute rank of src URL among collection of dst URLs
        .flatMap { case (urls, rank) =>
          val size = urls.size
          urls.map((_, rank / size))
        }
      ranks = contribs.group.sum
        .mapValues((1 - dampingFactor) + dampingFactor * _)
        .toTypedPipe
    }

    ranks
  }