Spark and Hadoop Basics Questions

import org.apache.spark.Partitioner
import java.util.zip.CRC32

class HotKeyPartitioner(val numPartitions: Int, hotKeyToPartition: Map[String, Int]) extends Partitioner {
  require(numPartitions > 0)
  // Required by Spark
  override def numPartitions: Int = this.numPartitions

  // key: composite key represented as string or tuple
  override def getPartition(key: Any): Int = {
    // check hot-key first (string representation or explicit type)
    val keyStr = key match {
      case s: String => s
      case (a, b) => s"$a|$b"
      case other => other.toString
    }

    hotKeyToPartition.get(keyStr) match {
      case Some(p) => Math.floorMod(p, numPartitions)
      case None =>
        // stable 32-bit hash using CRC32 for determinism across JVMs
        val crc = new CRC32()
        crc.update(keyStr.getBytes("UTF-8"))
        (crc.getValue.toInt & 0x7fffffff) % numPartitions
    }
  }

  override def equals(other: Any): Boolean = other match {
    case h: HotKeyPartitioner => h.numPartitions == numPartitions && h.hotKeyToPartition == hotKeyToPartition
    case _ => false
  }
}