Dynamic upload parallelism factor

snowplow-incubator · Nov 17, 2024 · b1f1019 · b1f1019
1 parent 4a3a929
commit b1f1019
Show file tree

Hide file tree

Showing 11 changed files with 70 additions and 42 deletions.
diff --git a/config/config.azure.reference.hocon b/config/config.azure.reference.hocon
@@ -75,8 +75,10 @@
     # - Events are emitted to Snowflake for a maximum of this duration, even if the `maxBytes` size has not been reached
     "maxDelay": "1 second"
 
-    # - How many batches can we send simultaneously over the network to Snowflake.
-    "uploadConcurrency":  1
+    # - Controls ow many batches can we send simultaneously over the network to Snowflake.
+    # -- E.g. If there are 4 available processors, and uploadParallelismFactor = 2.5, then we send up to 10 batches in parallel
+    # -- Adjusting this value can cause the app to use more or less of the available CPU.
+    "uploadParallelismFactor":  2.5
   }
 
   # -- Controls how the app splits the workload into concurrent batches which can be run in parallel.

diff --git a/config/config.kinesis.reference.hocon b/config/config.kinesis.reference.hocon
@@ -92,8 +92,10 @@
     # - Events are emitted to Snowflake for a maximum of this duration, even if the `maxBytes` size has not been reached
     "maxDelay": "1 second"
 
-    # - How many batches can we send simultaneously over the network to Snowflake.
-    "uploadConcurrency":  1
+    # - Controls ow many batches can we send simultaneously over the network to Snowflake.
+    # -- E.g. If there are 4 available processors, and uploadParallelismFactor = 2.5, then we send up to 10 batches in parallel
+    # -- Adjusting this value can cause the app to use more or less of the available CPU.
+    "uploadParallelismFactor":  2.5
   }
 
   # -- Controls how the app splits the workload into concurrent batches which can be run in parallel.

diff --git a/config/config.pubsub.reference.hocon b/config/config.pubsub.reference.hocon
@@ -81,8 +81,10 @@
     # - Events are emitted to Snowflake for a maximum of this duration, even if the `maxBytes` size has not been reached
     "maxDelay": "1 second"
 
-    # - How many batches can we send simultaneously over the network to Snowflake.
-    "uploadConcurrency":  1
+    # - Controls ow many batches can we send simultaneously over the network to Snowflake.
+    # -- E.g. If there are 4 available processors, and uploadParallelismFactor = 2.5, then we send up to 10 batches in parallel
+    # -- Adjusting this value can cause the app to use more or less of the available CPU.
+    "uploadParallelismFactor":  2.5
   }
 
   # -- Controls how the app splits the workload into concurrent batches which can be run in parallel.

diff --git a/modules/core/src/main/resources/reference.conf b/modules/core/src/main/resources/reference.conf
@@ -21,7 +21,7 @@
   "batching": {
     "maxBytes": 16000000
     "maxDelay": "1 second"
-    "uploadConcurrency":  3
+    "uploadParallelismFactor":  2.5
   }
   "cpuParallelismFraction": 0.75
 

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Config.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Config.scala
@@ -70,7 +70,7 @@ object Config {
   case class Batching(
     maxBytes: Long,
     maxDelay: FiniteDuration,
-    uploadConcurrency: Int
+    uploadParallelismFactor: BigDecimal
   )
 
   case class Metrics(

diff --git a/modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Environment.scala b/modules/core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/Environment.scala
@@ -63,13 +63,14 @@ object Environment {
       badSink <- toSink(config.output.bad.sink).onError(_ => Resource.eval(appHealth.beUnhealthyForRuntimeService(RuntimeService.BadSink)))
       metrics <- Resource.eval(Metrics.build(config.monitoring.metrics))
       tableManager <- Resource.eval(TableManager.make(config.output.good, appHealth, config.retries))
-      channelProviders <- Vector.range(0, config.batching.uploadConcurrency).traverse { index =>
+      cpuParallelism    = chooseCpuParallelism(config)
+      uploadParallelism = chooseUploadParallelism(config)
+      channelProviders <- Vector.range(0, uploadParallelism).traverse { index =>
                             for {
                               channelOpener <- Channel.opener(config.output.good, config.retries, appHealth, index)
                               channelProvider <- Channel.provider(channelOpener, config.retries, appHealth)
                             } yield channelProvider
                           }
-      cpuParallelism = chooseCpuParallelism(config)
     } yield Environment(
       appInfo        = appInfo,
       source         = sourceAndAck,
@@ -95,4 +96,15 @@ object Environment {
     (Runtime.getRuntime.availableProcessors * config.cpuParallelismFraction)
       .setScale(0, BigDecimal.RoundingMode.UP)
       .toInt
+
+  /**
+   * See the description of `uploadParallelism` on the [[Environment]] class
+   *
+   * For bigger instances (more cores) we produce batches more quickly, and so need higher upload
+   * parallelism so that uploading is not the bottleneck
+   */
+  private def chooseUploadParallelism(config: Config[Any, Any]): Int =
+    (Runtime.getRuntime.availableProcessors * config.batching.uploadParallelismFactor)
+      .setScale(0, BigDecimal.RoundingMode.UP)
+      .toInt
 }
diff --git a/.../core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/processing/Processing.scala b/.../core/src/main/scala/com.snowplowanalytics.snowplow.snowflake/processing/Processing.scala
@@ -363,24 +363,34 @@ object Processing {
       env.metrics.addGood(batch.origBatchCount - countBad) *> env.metrics.addBad(countBad)
     }
 
-  private implicit def batchableTokens: BatchUp.Batchable[BatchAfterTransform, Vector[Unique.Token]] =
-    new BatchUp.Batchable[BatchAfterTransform, Vector[Unique.Token]] {
-      def combine(b: Vector[Unique.Token], a: BatchAfterTransform): Vector[Unique.Token] =
-        b ++ a.tokens
+  /**
+   * Batches up checkpointing tokens, so we checkpoint every 10 seconds, instead of once per batch
+   *
+   * TODO: This should move to common-streams because it will be helpful for other loaders. The
+   * delay is helpful for Kinesis and Kafka, but not for PubSub.
+   */
+  private def emitTokens[F[_]: Async]: Pipe[F, BatchAfterTransform, Unique.Token] = {
+    implicit val batchableTokens: BatchUp.Batchable[BatchAfterTransform, Vector[Unique.Token]] =
+      new BatchUp.Batchable[BatchAfterTransform, Vector[Unique.Token]] {
+        def combine(b: Vector[Unique.Token], a: BatchAfterTransform): Vector[Unique.Token] =
+          b ++ a.tokens
 
-      def single(a: BatchAfterTransform): Vector[Unique.Token] =
-        a.tokens
+        def single(a: BatchAfterTransform): Vector[Unique.Token] =
+          a.tokens
 
-      def weightOf(a: BatchAfterTransform): Long =
-        0L
-    }
+        def weightOf(a: BatchAfterTransform): Long =
+          0L
+      }
+
+    // This will become configurable when we migrate it to common-streams
+    val checkpointDelay = 10.seconds
 
-  private def emitTokens[F[_]: Async]: Pipe[F, BatchAfterTransform, Unique.Token] =
-    BatchUp.withTimeout[F, BatchAfterTransform, Vector[Unique.Token]](Long.MaxValue, 10.seconds).andThen {
+    BatchUp.withTimeout[F, BatchAfterTransform, Vector[Unique.Token]](Long.MaxValue, checkpointDelay).andThen {
       _.flatMap { tokens =>
         Stream.emits(tokens)
       }
     }
+  }
 
   private def fastGetByIndex[A](items: IndexedSeq[A], index: Long): A = items(index.toInt)
 

diff --git a/modules/core/src/test/scala/com.snowplowanalytics.snowplow.snowflake/MockEnvironment.scala b/modules/core/src/test/scala/com.snowplowanalytics.snowplow.snowflake/MockEnvironment.scala
@@ -60,9 +60,9 @@ object MockEnvironment {
         metrics      = testMetrics(state),
         appHealth    = testAppHealth(state),
         batching = Config.Batching(
-          maxBytes          = 16000000,
-          maxDelay          = 10.seconds,
-          uploadConcurrency = 1
+          maxBytes                = 16000000,
+          maxDelay                = 10.seconds,
+          uploadParallelismFactor = BigDecimal(1)
         ),
         cpuParallelism = 2,
         schemasToSkip  = List.empty,

diff --git a/modules/kafka/src/test/scala/com/snowplowanalytics/snowplow/snowflake/KafkaConfigSpec.scala b/modules/kafka/src/test/scala/com/snowplowanalytics/snowplow/snowflake/KafkaConfigSpec.scala
@@ -105,9 +105,9 @@ object KafkaConfigSpec {
       )
     ),
     batching = Config.Batching(
-      maxBytes          = 16000000,
-      maxDelay          = 1.second,
-      uploadConcurrency = 3
+      maxBytes                = 16000000,
+      maxDelay                = 1.second,
+      uploadParallelismFactor = BigDecimal(2.5)
     ),
     cpuParallelismFraction = BigDecimal(0.75),
     retries = Config.Retries(
@@ -186,9 +186,9 @@ object KafkaConfigSpec {
       )
     ),
     batching = Config.Batching(
-      maxBytes          = 16000000,
-      maxDelay          = 1.second,
-      uploadConcurrency = 1
+      maxBytes                = 16000000,
+      maxDelay                = 1.second,
+      uploadParallelismFactor = BigDecimal(2.5)
     ),
     cpuParallelismFraction = BigDecimal(0.75),
     retries = Config.Retries(

diff --git a/...s/kinesis/src/test/scala/com/snowplowanalytics/snowplow/snowflake/KinesisConfigSpec.scala b/...s/kinesis/src/test/scala/com/snowplowanalytics/snowplow/snowflake/KinesisConfigSpec.scala
@@ -101,9 +101,9 @@ object KinesisConfigSpec {
       )
     ),
     batching = Config.Batching(
-      maxBytes          = 16000000,
-      maxDelay          = 1.second,
-      uploadConcurrency = 3
+      maxBytes                = 16000000,
+      maxDelay                = 1.second,
+      uploadParallelismFactor = BigDecimal(2.5)
     ),
     cpuParallelismFraction = BigDecimal(0.75),
     retries = Config.Retries(
@@ -177,9 +177,9 @@ object KinesisConfigSpec {
       )
     ),
     batching = Config.Batching(
-      maxBytes          = 16000000,
-      maxDelay          = 1.second,
-      uploadConcurrency = 1
+      maxBytes                = 16000000,
+      maxDelay                = 1.second,
+      uploadParallelismFactor = BigDecimal(2.5)
     ),
     cpuParallelismFraction = BigDecimal(0.75),
     retries = Config.Retries(

diff --git a/...les/pubsub/src/test/scala/com/snowplowanalytics/snowplow/snowflake/PubsubConfigSpec.scala b/...les/pubsub/src/test/scala/com/snowplowanalytics/snowplow/snowflake/PubsubConfigSpec.scala
@@ -101,9 +101,9 @@ object PubsubConfigSpec {
       )
     ),
     batching = Config.Batching(
-      maxBytes          = 16000000,
-      maxDelay          = 1.second,
-      uploadConcurrency = 3
+      maxBytes                = 16000000,
+      maxDelay                = 1.second,
+      uploadParallelismFactor = BigDecimal(2.5)
     ),
     cpuParallelismFraction = BigDecimal(0.75),
     retries = Config.Retries(
@@ -176,9 +176,9 @@ object PubsubConfigSpec {
       )
     ),
     batching = Config.Batching(
-      maxBytes          = 16000000,
-      maxDelay          = 1.second,
-      uploadConcurrency = 1
+      maxBytes                = 16000000,
+      maxDelay                = 1.second,
+      uploadParallelismFactor = BigDecimal(2.5)
     ),
     cpuParallelismFraction = BigDecimal(0.75),
     retries = Config.Retries(