Minimal PoC

core/src/main/scala/Block.scala

@@ -47,7 +47,7 @@ class Block(val records: LazyList[Record]) extends AnyVal {
 
   def writeTo(path: Path): File = {
     val file = File(path)
-    val writer = file.bufferedWriter()
+    val writer = file.bufferedOutput()
 
     try {
       toChars.foreach(writer.write(_))
@@ -65,7 +65,7 @@ class Block(val records: LazyList[Record]) extends AnyVal {
       .zipWithIndex
       .map({ case (records, index) =>
         val file = File(directory / s"partition.$index")
-        val writer = file.bufferedWriter()
+        val writer = file.bufferedOutput()
 
         try {
           records.foreach(_.toChars.foreach(writer.write(_)))

core/src/main/scala/Key.scala

@@ -8,17 +8,23 @@ object Key {
   def fromByteString(byteString: ByteString): Key = new Key(
     byteString.toByteArray
   )
+
+  def min(size: Int = 10) = new Key(Array.fill[Byte](size)(0x00.toByte))
+
+  def max(size: Int = 10) = new Key(Array.fill[Byte](size)(0xff.toByte))
 }
 
-class Key(val underlying: Array[Byte]) extends AnyVal with Ordered[Key] {
+case class Key(underlying: Array[Byte]) extends AnyVal with Ordered[Key] {
   def is(that: Key): Boolean = underlying sameElements that.underlying
 
   def hex: String = underlying.map("%02x" format _).mkString
 
+  def prior: Key = new Key(underlying.init :+ (underlying.last - 1).toByte)
+
   override def compare(that: Key): Int = underlying
     .zip(that.underlying)
     .map { case (a, b) =>
-      a - b
+      a.toChar - b.toChar
     }
     .find(_ != 0)
     .getOrElse(0)

docker-compose.yml

@@ -5,14 +5,33 @@ services:
       dockerfile: docker/master/Dockerfile
       args:
         - NUMBER_OF_WORKERS=2
-  worker:
+  worker-0:
     build:
       context: .
       dockerfile: docker/worker/Dockerfile
       args:
+        - WORKER_ID=0
         - MASTER_HOST=master
         - MASTER_PORT=50051
-    deploy:
-      replicas: 2
     depends_on:
       - master
+  worker-1:
+    build:
+      context: .
+      dockerfile: docker/worker/Dockerfile
+      args:
+        - WORKER_ID=0
+        - MASTER_HOST=master
+        - MASTER_PORT=50051
+    depends_on:
+      - master
+  worker-2:
+    build:
+      context: .
+      dockerfile: docker/worker/Dockerfile
+      args:
+        - WORKER_ID=0
+        - MASTER_HOST=master
+        - MASTER_PORT=50051
+    depends_on:
+      - master

docker/worker/Dockerfile

@@ -1,5 +1,7 @@
 FROM sbtscala/scala-sbt:eclipse-temurin-jammy-20.0.2_9_1.9.6_2.13.12
 
+ARG WORKER_ID
+
 ARG MASTER_HOST
 ARG MASTER_PORT
 
@@ -10,7 +12,7 @@ ENV SBT_OPTS="-Xmx2G -Xss2M"
 
 RUN mkdir -p /app /data /output
 
-COPY docker/worker/data /data
+COPY docker/worker/data/${WORKER_ID} /data
 
 WORKDIR /app
 
@@ -32,4 +34,4 @@ COPY . .
 
 RUN sbt --batch compile
 
-ENTRYPOINT sbt --batch -v "worker/run ${MASTER_HOST}:${MASTER_PORT} -I /data/0 /data/1 /data/2 -O /output/"
+ENTRYPOINT sbt --batch -v "worker/run ${MASTER_HOST}:${MASTER_PORT} -I /data/0 /data/1 -O /output/"

master/src/main/scala/Master.scala

@@ -13,19 +13,25 @@ import scala.concurrent.ExecutionContextExecutor
 object Master extends Logging {
   private def workersWithKeyRange(
       keys: List[Key],
-      workers: List[WorkerMetadata]
-  ): List[WorkerMetadata] =
-    keys
-      .sliding(
-        keys.size / workers.size,
-        keys.size / workers.size
+      workers: List[WorkerMetadata],
+      min: Key,
+      max: Key
+  ): List[WorkerMetadata] = {
+    val keysWithLowerBound = keys :+ min
+
+    val startKeys = keysWithLowerBound.sorted
+      .grouped(
+        (keysWithLowerBound.size.toDouble / workers.size.ceil).ceil.toInt
       )
       .toList
-      .map(keys => KeyRange.tupled(keys.head, keys.last))
-      .zip(workers)
-      .map { case (keyRange, worker) =>
-        worker.copy(keyRange = Some(keyRange))
-      }
+      .map(_.head)
+
+    val pairs = startKeys.zip(startKeys.tail.map(_.prior) :+ max)
+
+    workers.zip(pairs).map { case (worker, (min, max)) =>
+      worker.copy(keyRange = Some(KeyRange(min, max)))
+    }
+  }
 
   def main(args: Array[String]): Unit = {
     val masterArguments = new MasterArguments(args)
@@ -73,11 +79,10 @@ object Master extends Logging {
       .flatMap(_.sampledKeys)
       .map(Key.fromByteString)
 
-    logger.info("[Master] Sampled")
-
-    val sortedSampledKeys = sampledKeys.sorted
+    logger.info(s"[Master] Sampled $sampledKeys")
 
-    val workers = workersWithKeyRange(sortedSampledKeys, workerInfo)
+    val workers =
+      workersWithKeyRange(sampledKeys, workerInfo, Key.min(), Key.max())
 
     logger.info(s"[Master] Key ranges with worker: $workers")