What breaks most data pipelines is not the algorithm, it’s the glue (compositions). We chain effectful steps-read, transform, write-sprinkled with logging, retries, and configs. Then at 2 AM, one-step returns None/Left/an Exception and the whole thing craters. The code felt composable, but the functions didn’t actually compose.

This is exactly the corner category theory calls the Kleisli category of a monad: wrap the effectful arrow (A => F[B]) so composition becomes legal again. If that refresher helps, skim the Typelevel page on Kleisli (it shows the (>=>) law in two short paragraphs) and jump back.

Promise: by the end of this post you’ll take any A => F[B] and make it compose predictably, testably, and safely-across Cats Effect or ZIO-using Kleisli.

This is the missing trick in day‑to‑day data engineering: turn effectful functions into first‑class, chainable stages the compiler can reason about. Step: give me an A, I work inside box F, and I hand you back a B. That arrow is A => F[B], and Kleisli is the wrapper that lets those arrows click together.

Why this matters

• Effectful edges are everywhere in pipelines: IO, retries, metrics, tracing.
• Plain functions A => B compose. Effectful ones A => F[B] do not-unless you lift them into the right abstraction.
• We want composition that is:
  • type‑safe (compiler checks wiring),
  • effect‑polymorphic (CE or ZIO with the same user code),
  • test‑friendly (swap F or run in memory), ️- observable (trace each stage),
  • clear to read and reason about.

Kleisli gives you exactly this: a value representing an effectful function, with operators that compose like normal functions.

What we’ll build (no prior cats knowledge assumed)

1. A mental model for Kleisli in one screenful.
2. How a production‑style builder wires stages with Kleisli.
3. A small, runnable pipeline sketch.
4. Practical patterns: tracing, retries, testing, and when not to use it.

Prereqs: basic Scala. I’ll explain Cats/Kleisli as we go. We’ll use Cats Effect’s IO for examples (Scala 2.13/3). If you prefer ZIO, the ideas transfer-replace IO with your effect.

If you want a structured video course later, Rock the JVM has excellent material on Cats and Cats Effect you can take at your own pace.

References:

• Cats - Kleisli docs (short and precise): https://typelevel.org/cats/datatypes/kleisli.html
• Rock the JVM - Cats course: https://rockthejvm.com/courses/cats/
• Rock the JVM - Cats Effect course: https://rockthejvm.com/courses/cats-effect/

Part 1 - Kleisli: the mental model

A => B composes. A => F[B] doesn’t (the types don’t line up). Pipelines live at those effectful edges: read: Unit => F[In], transform: In => F[Out], write: Out => F[Unit].

Two pieces to keep in mind:

• F[_] just means “some effect type” (e.g., IO, Future, Task). Read it as: “a computation that may do I/O and eventually produce a B”.
• Kleisli[F, A, B] is a tiny wrapper around a function A => F[B]. With this wrapper, we get andThen, map, flatMap, etc. so effectful functions compose like normal ones.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

// Core idea in one line
// Kleisli[F, A, B] wraps: A => F[B]
import cats.data.Kleisli
import cats.effect.{IO, Resource}

// Types we’ll use in the examples
final case class In(raw: String)
final case class Out(clean: String)

// Lift ordinary functions into Kleislis
val read: Kleisli[IO, Unit, In] = Kleisli { _ =>
  Resource
    .fromAutoCloseable(IO.blocking(scala.io.Source.fromFile("/tmp/in.csv")))
    .use(src => IO.blocking(src.getLines().next()).map(In.apply))
}
val transform: Kleisli[IO, In, Out]   = Kleisli(i => IO.pure(Out(i.raw.trim.toLowerCase)))
val write:     Kleisli[IO, Out, Unit] = Kleisli(o => IO.println(s"write: ${o.clean}"))

// Compose like regular functions
val pipeline: Kleisli[IO, Unit, Unit] = read andThen transform andThen write

You now have one value (pipeline) you can run, test, time, trace, and reuse-no ad‑hoc glue scattered across methods.

What the F[_] ? 😉

The same pipeline should run on different runtimes (Cats Effect IO, ZIO Task, etc.). F[_] is just “some effect”. Keep functions generic and request only the capabilities you need (map/flatMap, timing, concurrency) via typeclasses, then choose the runtime at the boundary. Deep dive coming soon: /effect-polymorphism .

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Generic: works for any effect F with a Monad
def normalizeF[F[_]: cats.Monad]: Kleisli[F, In, Out] =
  Kleisli(i => cats.Monad[F].pure(Out(i.raw.trim.toLowerCase)))

// Choose a runtime
val normalizeIO  : Kleisli[cats.effect.IO,  In, Out] = normalizeF[cats.effect.IO]
// ZIO (requires zio-interop-cats)
import zio.Task
import zio.interop.catz.given
val normalizeZIO : Kleisli[Task, In, Out] = normalizeF[Task]

flowchart LR
  NF["normalizeF[F]: Kleisli[F, In, Out]"] --> IOv["normalizeF[IO]: Kleisli[IO, In, Out]"]
  NF --> ZIOv["normalizeF[Task]: Kleisli[Task, In, Out]"]

sequenceDiagram
  participant Dev as "Developer"
  participant Comp as "Compiler"
  participant Lib as "Generic code: normalizeF[F]"
  participant IO as "Runtime: IO"
  participant Task as "Runtime: Task"
  Dev->>Comp: compile
  Comp->>Lib: typecheck F[_] constraints
  alt choose IO
    Comp->>IO: instantiate normalizeF[IO]
    Dev->>IO: run Kleisli with IO
  else choose Task
    Comp->>Task: instantiate normalizeF[Task]
    Dev->>Task: run Kleisli with Task
  end

Part 2 - Build a small pipeline end‑to‑end

Let’s wire a realistic round‑trip: read → validate → transform → write. No framework, just Kleisli and Cats Effect.

flowchart LR
  A["Source: Unit => IO[Raw]"] --> B["Parse+Validate: Raw => IO[User]"]
  B --> C["Transform: User => IO[User]"]
  C --> D["Sink: User => IO[Unit]"]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

// build.sbt:
// libraryDependencies += "org.typelevel" %% "cats-core"   % "2.10.0"
// libraryDependencies += "org.typelevel" %% "cats-effect" % "3.5.4"

import cats.data.Kleisli
import cats.effect.{IO, IOApp, Resource}
import cats.syntax.all._
import scala.io.Source

final case class Raw(line: String)
final case class User(id: Long, email: String)

// 1) Source: Unit => IO[Raw]
val readRaw: Kleisli[IO, Unit, Raw] = Kleisli { _ =>
  Resource
    .fromAutoCloseable(IO.blocking(Source.fromFile("/tmp/users.csv")))
    .use(src => IO.blocking(src.getLines().next()).map(Raw.apply))
}

// 2) Parse + validate: Raw => IO[User]
val toUser: Kleisli[IO, Raw, User] = Kleisli { r =>
  r.line.split(',') match {
    case Array(idStr, email) if idStr.forall(_.isDigit) && email.contains('@') => IO.pure(User(idStr.toLong, email.trim.toLowerCase))
    case _ => IO.raiseError(new IllegalArgumentException(s"bad row: ${r.line}"))
  }
}

// 3) Transform: User => IO[User]
val normalize: Kleisli[IO, User, User] = Kleisli(u => IO.pure(u.copy(email = u.email.trim)))

// 4) Sink: User => IO[Unit]
val writeUser: Kleisli[IO, User, Unit] = Kleisli(u => IO.println(s"storing: $u"))

// Compose the pipeline
val pipeline: Kleisli[IO, Unit, Unit] = readRaw andThen toUser andThen normalize andThen writeUser

object Main extends IOApp.Simple {
  def run: IO[Unit] = pipeline.run(())
}

Prefer a shorter operator? Cats (and Scalaz years before it) exposes the classic “fish” >=> as syntax sugar for andThen, plus >>>/<<< via import cats.syntax.compose._. Same composition, different paint-use whatever keeps the flow readable.

1
2
3
4

import cats.syntax.compose._ // enables >=>, >>>, <<<

val pipelineFish: Kleisli[IO, Unit, Unit] =
  readRaw >=> toUser >=> normalize >=> writeUser

What’s happening here?

• Each stage is a tiny, named value with a single responsibility.
• The type tells you exactly what goes in and what comes out.
• Errors are typed at the edges (here we used exceptions for brevity; you can prefer Either/Validated).
• Composing andThen keeps the flow obvious.

Tip: if you want multiple failures collected (e.g., data quality), use ValidatedNel inside the stage and only fail the pipeline at the final decision point.

Part 2.5 - A mini builder (type‑safe and small)

Let’s “lift‑and‑shape” a production pattern into a small, self‑contained builder that composes a typed Kleisli as you add stages.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

// File: com/example/pipeline/Builder.scala
package com.example.pipeline

import cats.data.Kleisli
import cats.{Applicative, Monad}

// 1) Contracts and policies (sketch)
sealed trait SchemaPolicy
object SchemaPolicy {
  sealed trait Exact extends SchemaPolicy
  case object Exact extends Exact
}
trait SchemaConforms[C, R, P <: SchemaPolicy]

// 2) Metadata we carry alongside the composed pipeline
final case class PipelineMeta(name: String, stages: Vector[String], transforms: Int)

// 3) The builder keeps a composed, fully-typed Kleisli[F, In, Out]
final case class PipelineBuilder[F[_]: Monad, In, Out](
  name: String,
  k: Kleisli[F, In, Out],
  meta: PipelineMeta,
) {
  def addSource[C, R, P <: SchemaPolicy](
    srcName: String,
  )(read: Unit => F[C])(implicit evIn: In =:= Unit, evOut: Out =:= Unit, ev: SchemaConforms[C, R, P]): PipelineBuilder[F, Unit, C] = {
    val next = k.andThen(Kleisli[F, Out, C](_ => read(()))) // Out is Unit here by constraint
    PipelineBuilder(name, next, meta.copy(stages = meta.stages :+ s"source-$srcName"))
  }

  def addTransform[C](stepName: String)(f: Out => F[C]): PipelineBuilder[F, In, C] = {
    val next = k.andThen(Kleisli(f))
    PipelineBuilder(name, next, meta.copy(stages = meta.stages :+ s"xform-$stepName", transforms = meta.transforms + 1))
  }

  def noTransform(implicit A: Applicative[F]): PipelineBuilder[F, In, Out] = addTransform("identity")(o => A.pure(o))

  def addSink[R, P <: SchemaPolicy](sinkName: String)(write: Out => F[Unit])(implicit ev: SchemaConforms[Out, R, P]): PipelineBuilder[F, In, Unit] = {
    val next = k.andThen(Kleisli(write))
    PipelineBuilder(name, next, meta.copy(stages = meta.stages :+ s"sink-$sinkName"))
  }

  def build(implicit ev: Out =:= Unit): (Kleisli[F, In, Unit], PipelineMeta) = (k, meta)
}

object PipelineBuilder {
  def apply[F[_]: Monad](name: String): PipelineBuilder[F, Unit, Unit] =
    PipelineBuilder(name, Kleisli.ask[F, Unit], PipelineMeta(name, Vector.empty, 0))
}

What this gives you

• Typed endpoints with SchemaConforms evidence at the boundaries.
• A builder that only exposes build when the output type is Unit (after the sink) via an Out =:= Unit constraint.
• A composed Kleisli[F, In, Unit] the whole time - no Any, no unsafe casts in your domain code.

Try it quickly

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import cats.effect.{IO, IOApp}
import com.example.pipeline._

final case class User(id: Long, email: String)

// Provide evidence to allow the pipeline (in real life, derive or macro it)
given SchemaConforms[User, User, SchemaPolicy.Exact] = new SchemaConforms[User, User, SchemaPolicy.Exact] {}

object Demo extends IOApp.Simple {
  val (pipe, meta) = PipelineBuilder[IO]("users")
    .addSource[User, User, SchemaPolicy.Exact]("file")(_ => IO.pure(User(1, "[email protected]")))
    .addTransform("normalize")(u => IO.pure(u.copy(email = u.email.toLowerCase.trim)))
    .addSink[User, SchemaPolicy.Exact]("stdout")(u => IO.println(s"out: $u"))
    .build

  def run: IO[Unit] = IO.println(meta) *> pipe.run(())
}

Compile‑fail demo (comment out the given evidence above) and the source/sink calls that require it will not typecheck..

Part 3 - Add observability and retries without changing business logic

You can wrap a Kleisli to add logging/tracing/retries in one place.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

import cats.data.Kleisli
import cats.effect.IO
import scala.concurrent.duration._

def withRetry[A, B](k: Kleisli[IO, A, B], max: Int): Kleisli[IO, A, B] =
  Kleisli { a =>
    def loop(n: Int): IO[B] =
      k.run(a).handleErrorWith { e =>
        if (n >= max) IO.raiseError(e)
        else IO.sleep(200.millis) *> loop(n + 1)
      }
    loop(0)
  }

def traced[A, B](name: String)(k: Kleisli[IO, A, B]): Kleisli[IO, A, B] =
  Kleisli(a => IO.println(s"start $name") *> k.run(a).guarantee(IO.println(s"end   $name")))

val robust = withRetry(traced("normalize")(normalize), max = 3)
val resilientPipeline = readRaw andThen toUser andThen robust andThen writeUser

Because everything is just values, you can layer concerns without touching the business code.

Part 4 - Practical patterns with Kleisli

• Tracing per stage: wrap with a small decorator (as above). In prod, swap println with your logger or tracing span.
• Typed failures vs aggregated validations: use Either/EitherT for fail‑fast steps and ValidatedNel to collect multiple issues before a sink.
• Retry/backoff at edges: compose a withRetry wrapper around the few stages that hit flaky systems.
• Testing: run stages as plain functions with test inputs; for IO, prefer cats-effect testing utilities (e.g., TestControl) instead of unsafeRunSync.
• Observability: since stages are values, you can add metrics once and reuse the wrapped stage wherever needed.

Small before/after to internalize “effects at edges”:

1
2
3
4
5
6
7
8
9

// ❌ Anti-pattern: effect inside a transform leaks concerns
val transformBad: Kleisli[IO, User, User] = Kleisli { u =>
  IO.println(s"email before: ${u.email}") *> IO.pure(u.copy(email = u.email.toLowerCase))
}

// ✅ Better: keep transform pure; add logs via a wrapper
val transformPure: User => User = u => u.copy(email = u.email.toLowerCase)
val transformGood: Kleisli[IO, User, User] = Kleisli(u => IO.pure(transformPure(u)))
val observed       = instrument("normalize", transformGood)(yourEmitter)

How other teams already do this

Look at the way the Cats docs teach Reader (aka Kleisli) as a pipeline of small Jobs: they model each topic of work as a Kleisli stage and wire the stages with traverse, exactly the way we want to stitch Spark transforms.

Community answers echo the same advice: if you already have A => F[B] slices in prod, wrap them in Kleisli once and use andThen instead of hand-rolled glue.

Flink’s own documentation describes a pipeline as a chain of operators flowing from source to sink. Map each operator to a Kleisli, and you get a mental model that works for both the Table API and your Scala glue code—one stage per arrow, all composed left to right.

A quick note on why not just pick one effect

Most teams standardise on one runtime-Cats Effect IO, ZIO Task, whatever ships with the platform-and that’s fine. The point of Kleisli isn’t to flip runtimes every sprint. It’s to tame everything that sits in F: audit logging, notifications, lineage, retries, metrics. Those are the things that make pipelines noisy. When each stage is an A => F[B], you can wrap that arrow once with withAudit, withNotifications, withRetries, and the business code stays untouched. Composition stays readable because the glue is a value, not scattered side effects.

Part 5 - When not to reach for Kleisli

• Inside a single pure transform: just write A => B.
• When your steps aren’t effectful: function composition is simpler.
• When the abstraction leaks too many types: prefer small helpers and introduce Kleisli only at module boundaries.

Rule of thumb: use Kleisli where IO begins and ends-sources, sinks, boundaries, service calls, and stage glue. Keep inner transforms as plain A => B.

Appendix - Kleisli cheat sheet

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import cats.data.Kleisli
import cats.syntax.all._

// Lift a plain effectful function
def lift[F[_], A, B](f: A => F[B]): Kleisli[F, A, B] = Kleisli(f)

// Compose
val k: Kleisli[F, A, C] = k1 andThen k2

// Map, flatMap
val k2: Kleisli[F, A, C] = k1.map(b => toC(b))
val k3: Kleisli[F, A, C] = k1.flatMap(b => Kleisli(a => next(a, b)))

// Add context (tracing/metrics) without changing types
def traced[F[_], A, B](name: String)(k: Kleisli[F, A, B])(
  inSpan: (String, => F[B]) => F[B]
): Kleisli[F, A, B] = Kleisli(a => inSpan(name, k.run(a)))

References

• Typelevel Cats - Kleisli and composition: https://typelevel.org/cats/datatypes/kleisli.html
• Typelevel Cats - Traverse docs on building pipelines: https://typelevel.org/cats/typeclasses/traverse.html
• Rock the JVM - Scala with Cats course: https://rockthejvm.com/courses/cats/
• Rock the JVM - Cats Effect course: https://rockthejvm.com/courses/cats-effect/
• “Kleisli composition” explainer (theory and (>=>) laws): https://xenon.stanford.edu/~hwatheod/monads.html
• Stack Overflow - When to reach for Kleisli in production pipelines: https://stackoverflow.com/questions/2828588/how-to-compose-functions-that-return-monads
• Apache Flink docs - How operator chains define a pipeline: https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/concepts/programming-model/
• Background post on compile‑time contracts: /compile-time-data-contracts
• FlowForge docs (production builder wiring tracing/lineage with Kleisli): https://github.com/vim89/flowforge/blob/main/docs/start-here.md

TL;DR

• Kleisli turns A => F[B] back into a composable building block.
• A Kleisli is a named step; steps click together like an assembly line.
• Avoid nested flatMaps - read pipelines left → right with andThen.
• Use Kleisli at IO boundaries; keep transforms pure and tiny.

Bonus - Type‑aligned composition

Sometimes you want to accumulate stages of different shapes (Kleisli[F, A, B], then Kleisli[F, B, C], etc.) and end up with one pipeline. The most type‑safe approach is to compose into a single Kleisli[F, In, Out] value as you add stages (exactly what our mini‑builder does) - not a list of erased functions.

If you need an intermediate collection, use a type‑aligned sequence where each element’s output matches the next element’s input. In code, prefer composing immediately:

1
2

val pipeline: Kleisli[IO, Unit, Unit] =
  readRaw andThen toUser andThen normalize andThen writeUser

No Any, no casts - just type‑checked composition.

Minimal typestate builder with typed endpoints

Below is a compact sketch that borrows a proven layout: typed sources/sinks guarded by compile‑time contract evidence and a transform in the middle.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

import cats.data.Kleisli
import cats.Applicative

// Contract policy model
sealed trait SchemaPolicy
object SchemaPolicy {
  sealed trait Exact extends SchemaPolicy
  case object Exact extends Exact
}

// Evidence that producer C conforms to contract R under policy P
// (Provide your own instances or use compile-time macros; see the linked contracts post.)
trait SchemaConforms[C, R, P <: SchemaPolicy]

// Phantom states
sealed trait St
object St { sealed trait Empty extends St; sealed trait WithContract extends St; sealed trait WithTransform extends St; sealed trait Complete extends St }

// Data endpoints
final case class Source[A](id: String)
final case class Sink[A](id: String)

// A tiny builder that composes Kleislis underneath
final case class Builder[F[_]: Monad, In, Out](name: String, k: Kleisli[F, In, Out]) {
  def addTransform[C](f: Out => F[C]): Builder[F, In, C] = Builder(name, k.andThen(Kleisli(f)))
  def build: Kleisli[F, In, Out] = k
}
object Builder { def apply[F[_]: Monad](name: String): Builder[F, Unit, Unit] = Builder(name, Kleisli.ask[F, Unit]) }

This is purposely compact to fit a post. The ideas mirror what you’d do in production: use Kleisli for composition, use phantom types to force a correct sequence, and gate typed endpoints with SchemaConforms evidence so schema drift is a compile‑time error.

classDiagram
  class Builder~F,In,Out~ {
    - k: Kleisli~F,In,Out~
    + addTransform()
    + build(): Kleisli~F,In,Out~
  }
  class Kleisli~F,A,B~
  Builder --> Kleisli

Part 6 - Instrumentation hooks that don’t leak into business code

Per‑stage logs/metrics/lineage shouldn’t be sprinkled into every function. Add them from the outside by wrapping stages with a small decorator that emits Start/Complete/Fail events and timing. Keep it optional so the same pipeline runs with or without it.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

// A generic event emitter (wire to your logging/metrics/lineage system)
trait Emitter[F[_]] {
  def start(stage: String): F[Unit]
  def success(stage: String, tookMs: Long): F[Unit]
  def failure(stage: String, err: Throwable): F[Unit]
}

def instrument[F[_]: cats.MonadThrow, A, B](stage: String, k: Kleisli[F, A, B])(emit: Emitter[F]): Kleisli[F, A, B] =
  Kleisli { a =>
    val now = System.nanoTime()
    emit.start(stage) *> k.run(a).attempt.flatMap {
      case Right(b) =>
        val ms = (System.nanoTime() - now) / 1000000L
        emit.success(stage, ms) *> cats.Monad[F].pure(b)
      case Left(e)  => emit.failure(stage, e) *> cats.MonadThrow[F].raiseError[B](e)
    }
  }

Use it where you need it:

1
2

val observedNormalize = instrument("normalize", normalize)(yourEmitter)
val observedPipeline  = readRaw andThen toUser andThen observedNormalize andThen writeUser

Result: per‑stage observability with no code changes in your transforms.

sequenceDiagram
  actor Dev
  participant Pipe as Pipeline
  participant Emit as Emitter
  participant Stage as "Stage.execute"
  Dev->>Pipe: run()
  Pipe->>Emit: start(stage)
  Pipe->>Stage: Kleisli.run(a)
  alt success
    Stage-->>Pipe: result
    Pipe->>Emit: success(stage, tookMs)
  else failure
    Stage-->>Pipe: error
    Pipe->>Emit: failure(stage, err)
  end

Part 7 - Typed endpoints (very briefly)

If you model sources/sinks with typed contracts, the “are we compatible?” question can be answered at compile time, and the failure shows up before deploy. That topic is bigger than this post; see the dedicated article for a deep dive and examples: /compile-time-data-contracts .

Part 8 - Error model, idempotency, and exactly‑once behavior at the edges

Pipelines fail in the real world; we want actionable errors and safe retries. Use a small error ADT for domain failures; prefer Either/Validated in business logic, and reserve exceptions for hard boundaries. Make sinks idempotent (upsert on a key, or keep a ledger of hashes) so retries don’t duplicate work.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

sealed trait PipelineError
object PipelineError {
  final case class Parse(msg: String) extends PipelineError
  final case class External(msg: String, cause: Throwable) extends PipelineError
}

import cats.data.EitherT

type Stage[F[_], A, B] = Kleisli[EitherT[F, PipelineError, *], A, B]

// Example: parse with good errors, not exceptions
val parse: Stage[IO, Raw, User] = Kleisli { r =>
  EitherT(IO.pure(parseRow(r).toRight(PipelineError.Parse(s"bad row: ${r.line}"))))
}

// Idempotent sink (conceptual): insert if absent else compare‑and‑update
def idempotentWrite(u: User): IO[Unit] = IO.blocking {
  // 1) Upsert on (id) with a unique constraint
  // 2) Or maintain a sink ledger: (id, hash) and skip duplicates
  ()
}

This keeps failures explicit and retries safe.

Part 9 - Parallelism where it actually helps

Some steps fan out naturally (read partitions, transform per partition) and then rejoin. Parallelize over input collections rather than inside a single small stage.

1
2
3
4

import cats.syntax.all._

def processBatch(batch: List[Raw]): IO[Unit] =
  batch.parTraverse_(r => (readRaw andThen toUser andThen normalize andThen writeUser).run(()))

Be selective: parallelism is great for independent items; it’s noise for single‑record transforms.

Part 10 - Resource safety (files, DBs, clients)

Leaking file handles or DB sessions will ruin your day. Use Resource (Cats Effect) to acquire/use/release safely; wrap it in Kleisli.

1
2
3
4
5
6
7
8
9

import cats.effect.{IO, Resource}

def file(path: String): Resource[IO, java.io.BufferedReader] = Resource.fromAutoCloseable(
  IO.blocking(new java.io.BufferedReader(new java.io.FileReader(path)))
)

val readFirst: Kleisli[IO, Unit, String] = Kleisli { _ =>
  file("/tmp/in.csv").use(br => IO.blocking(br.readLine()))
}

You get correctness by construction: even on errors, resources close.

Appendix B - sbt skeleton to copy‑paste

A runnable starter accelerates learning.

1
2
3
4
5
6
7
8
9

// build.sbt
ThisBuild / scalaVersion := "2.13.14"
libraryDependencies ++= Seq(
  "org.typelevel" %% "cats-core"        % "2.10.0",
  "org.typelevel" %% "cats-effect"      % "3.5.4",
  "org.scalatest" %% "scalatest"        % "3.2.19" % Test,
  // If you want to instantiate generic F[_] code with ZIO:
  "dev.zio"       %% "zio-interop-cats" % "23.1.0.5"
)

1
2
3
4

// src/main/scala/Main.scala
object Main extends cats.effect.IOApp.Simple {
  def run = Demo.run // from the examples above
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// src/test/scala/ContractsSpec.scala
import org.scalatest.wordspec.AnyWordSpec
import org.scalatest.matchers.should.Matchers

class ContractsSpec extends AnyWordSpec with Matchers {
  "compile-fail" should {
    "reject mismatched schemas under Exact" in {
      assertTypeError("""
        trait SchemaConforms[C, R, P]
        sealed trait SchemaPolicy; object SchemaPolicy { sealed trait Exact extends SchemaPolicy }
        case class A(id: Long); case class B(id: Long, email: String)
        summon[SchemaConforms[A, B, SchemaPolicy.Exact]]
      """)
    }
  }
}

Take‑home checklist

• Contracts first: write types for producer/consumer and pick a policy.
• Typed edges: require schema evidence at your source/sink boundaries.
• Kleisli everywhere at edges: treat stages as A => F[B] values you can compose.
• Keep transforms pure: IO at the boundaries only; make sinks idempotent.
• Observability from the outside: wrap stages with timing and events; avoid logging inside transforms.
• Fast feedback: add at least one compile‑fail test to prove the policy gate.