pubsub: sink without order

Currently all enrich apps sink chunks in order but we don't need order for PubSub. We need order for Kinesis and Kafka.

enrich/modules/common-fs2/src/main/scala/com/snowplowanalytics/snowplow/enrich/common/fs2/Enrich.scala

Lines 95 to 104 in 8f91230

    
           val sinkAndCheckpoint: Pipe[F, List[(A, Result)], Unit] = 
        
             _.parEvalMap(env.streamsSettings.concurrency.sink)(chunk => 
        
               for { 
        
                 begin <- Clock[F].realTime 
        
                 result <- sinkChunk(chunk.map(_._2), env).as(chunk.map(_._1)) 
        
                 end <- Clock[F].realTime 
        
                 _ <- Logger[F].debug(s"Chunk of size ${chunk.size} sunk in ${end - begin}") 
        
               } yield result 
        
             ) 
        
               .evalMap(env.checkpoint)

We should conditionally use parEvalMapUnordered for PubSub so that resource usage (CPU and mem) and throughput can improve. It is a matter of benchmarking to see how much improvement it'll bring.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	val sinkAndCheckpoint: Pipe[F, List[(A, Result)], Unit] =
	_.parEvalMap(env.streamsSettings.concurrency.sink)(chunk =>
	for {
	begin <- Clock[F].realTime
	result <- sinkChunk(chunk.map(_._2), env).as(chunk.map(_._1))
	end <- Clock[F].realTime
	_ <- Logger[F].debug(s"Chunk of size ${chunk.size} sunk in ${end - begin}")
	} yield result
	)
	.evalMap(env.checkpoint)

Description

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions