Repartition RDDs before writing to Alternator

scylladb · Mar 12, 2024 · 5a03158 · 5a03158
1 parent 6a0895d
commit 5a03158
Showing 1 changed file with 5 additions and 4 deletions.
diff --git a/migrator/src/main/scala/com/scylladb/migrator/writers/DynamoStreamReplication.scala b/migrator/src/main/scala/com/scylladb/migrator/writers/DynamoStreamReplication.scala
@@ -65,10 +65,11 @@ object DynamoStreamReplication {
         case _ => None
       }
       .foreachRDD { msgs =>
-        val rdd = msgs.collect {
-          case Some(item) => (new Text(), new DynamoDBItemWritable(item))
-        }
-        // FIXME re-partition by columns? msgs.repartition(???)
+        val rdd = msgs
+          .collect {
+            case Some(item) => (new Text(), new DynamoDBItemWritable(item))
+          }
+          .repartition(Runtime.getRuntime.availableProcessors() * 2)
 
         log.info("Changes to be applied:")
         rdd