Choreo Dataflow: Comprehensive Walkthrough

Section

Mix.install([
  {:choreo, "~> 0.14.1"},
  # {:choreo, path: Path.expand("../..", __DIR__), force: true},
  {:kino_vizjs, "~> 0.9.0"}
])

Rendering diagrams: This livebook uses Kino.VizJS to render DOT diagrams inline and Choreo.Lab.Siren / Choreo.Lab.Sketch for Mermaid previews.

What is Choreo.Dataflow?

Choreo.Dataflow models stream-processing and ETL pipelines as directed graphs. Unlike static diagramming tools, you define your pipeline as code and then ask it questions:

• "Where are the bottlenecks?"
• "What's the slowest path end-to-end?"
• "Will this topology create backpressure?"
• "Are there orphan nodes that never emit or receive data?"

Nodes are typed: sources, transforms, buffers, conditionals, merges, and sinks. Edges carry data-type annotations and can represent normal, error, retry, or dead-letter paths.

Node Types & Shapes

API Approaches: Programmatic Pipe API vs Lab DSL

Choreo provides two ways to author dataflow pipelines:

1. Programmatic Pipe API (Choreo.Dataflow): Explicit, pipe-first builder functions (add_source/3, add_sink/3, add_transform/3, add_buffer/3, add_conditional/3, add_merge/3, add_cluster/3, connect/4). Best for production services, dynamic DAG generation, and pipeline compilers.
2. Lab DSL (Choreo.Lab.DSL.Dataflow): Concise, sketch-oriented syntax (dataflow do ... end, variable binding, ~>, and pipe modifiers like |> emits("event"), |> retry(), |> dead_letter()). Best for Livebooks, architecture reviews, pipeline sketches, and rapid data movement modeling.

In this guide, the introductory legend demonstrates the programmatic pipe API, while subsequent examples showcase the expressive Lab DSL.

alias Choreo.Dataflow
alias Choreo.Dataflow.Analysis
import Choreo.Lab.DSL.Dataflow

legend =
  Dataflow.new()
  |> Dataflow.add_source(:source, label: "Source")
  |> Dataflow.add_transform(:transform, label: "Transform")
  |> Dataflow.add_buffer(:buffer, label: "Buffer")
  |> Dataflow.add_conditional(:conditional, label: "Conditional")
  |> Dataflow.add_merge(:merge, label: "Merge")
  |> Dataflow.add_sink(:sink, label: "Sink")

siren =
  Choreo.Lab.Siren.new(
    Dataflow.to_mermaid(legend,
      theme: Choreo.Dataflow.theme(:default)
    )
  )

graphviz =
  Kino.VizJS.render(
    Dataflow.to_dot(legend,
      theme: Choreo.Dataflow.theme(:default, graph_rankdir: :tb)
    )
  )

sketch =
  Choreo.Lab.Sketch.new(
    Dataflow.to_mermaid(legend,
      theme: Choreo.Dataflow.theme(:default)
    )
  )

Kino.Layout.tabs(
  Siren: siren,
  Graphviz: graphviz,
  Sketch: sketch
)

Example 1: Simple ETL Pipeline

A classic extract-transform-load flow: ingest logs, parse JSON, buffer through Kafka, and write to TimescaleDB.

etl =
  dataflow do
    log_shipper = source("Log Shipper\n10_000 evt/s", rate: 10_000)
    json_parser = transform("JSON Parser", latency_ms: 25)
    kafka = buffer("Kafka\n(cap: 50_000)", capacity: 50_000)
    enricher = transform("Enricher", latency_ms: 80)
    timescaledb = sink("TimescaleDB")

    log_shipper ~> json_parser |> emits("raw logs")
    json_parser ~> kafka |> emits("parsed event")
    kafka ~> enricher |> emits("event")
    enricher ~> timescaledb |> writes("enriched row")
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(etl)),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(etl, rankdir: :tb), height: "500px"),
  Sketch: Choreo.Lab.Sketch.new(Dataflow.to_mermaid(etl))
)

Basic Analysis

IO.inspect(Analysis.cyclic?(etl), label: "Cyclic?")
IO.inspect(Analysis.orphan_nodes(etl), label: "Orphan nodes")
IO.inspect(Analysis.topological_sort(etl), label: "Topological order")

No cycles, no orphans, clean linear pipeline. But is it fast enough?

Example 2: IoT Pipeline with Backpressure

Real-world streams have mismatched speeds. A sensor bursts data faster than the parser can handle. Let's model that and find the bottleneck.

iot =
  dataflow do
    sensor_a = source("Sensor A\n2_000 evt/s", rate: 2_000)
    sensor_b = source("Sensor B\n3_000 evt/s", rate: 3_000)
    sensor_c = source("Sensor C\n5_000 evt/s", rate: 5_000)
    mux = merge("Merge")
    parser = transform("Parser\n50ms", latency_ms: 50)
    rabbitmq = buffer("RabbitMQ\n(cap: 8_000)", capacity: 8_000)
    aggregator = transform("Window Aggregator\n200ms", latency_ms: 200)
    influxdb = sink("InfluxDB")

    sensor_a ~> mux
    sensor_b ~> mux
    sensor_c ~> mux
    mux ~> parser
    parser ~> rabbitmq
    rabbitmq ~> aggregator
    aggregator ~> influxdb
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(iot)),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(iot), height: "500px"),
  Sketch: Choreo.Lab.Sketch.new(Dataflow.to_mermaid(iot))
)

Bottleneck Detection

Analysis.bottlenecks(iot)
|> IO.inspect(label: "Bottlenecks")

The buffer with the smallest headroom relative to upstream throughput surfaces first. If RabbitMQ appears here, you know the parser or aggregator is slower than the combined sensor rate.

Throughput Simulation

Analysis.simulate(iot)
|> Enum.each(fn {id, stats} ->
  IO.puts("#{id}: Inbound #{stats.in_rate} | Outbound #{stats.out_rate} evt/s")
end)

simulate/1 propagates rates forward and computes effective throughput at every stage. If a transform has lower capacity than its input, the downstream rate drops — that's backpressure in action.

Critical Path

{:ok, path, latency_ms} = Analysis.longest_path(iot)

IO.puts("Slowest path: #{Enum.join(path, " → ")}")
IO.puts("End-to-end latency: #{latency_ms}ms")

The longest path tells you the minimum latency any single event will experience from source to sink. If this exceeds your SLA, you need to parallelise or optimise the slowest transform.

Example 3: Error Handling & Dead-Letter Queues

Production pipelines don't just have happy paths. Failed parses need retries; poison pills need quarantine.

resilient =
  dataflow do
    webhook = source("Stripe Webhook\n500 req/s", rate: 500)
    validator = transform("Signature Validator\n10ms", latency_ms: 10)
    router = conditional("Valid?")
    processor = transform("Event Processor\n100ms", latency_ms: 100)
    retry_queue = buffer("Retry Queue\n(max: 3)", capacity: 1_000)
    retry_handler = transform("Retry Handler\n150ms", latency_ms: 150)
    dlq = buffer("Dead Letter Queue", capacity: 10_000)
    postgres = sink("Postgres")
    alerting = sink("PagerDuty")

    # Happy path
    webhook ~> validator |> emits("raw payload")
    validator ~> router |> emits("validated payload")
    router ~> processor |> emits("yes")
    processor ~> postgres |> writes("processed event")

    # Retry loop
    router ~> retry_queue |> retry("no")
    retry_queue ~> retry_handler
    retry_handler ~> router |> retry("reattempt")

    # Dead letter after retries exhausted
    retry_handler ~> dlq |> dead_letter("failed")
    dlq ~> alerting |> error("alert")
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(resilient), height: "800px"),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(resilient, rankdir: :tb), height: "800px")
)

Notice the edge styling:

• Solid = normal flow
• Dashed red = error path
• Dashed orange = retry loop
• Dotted purple = dead-letter

Detecting Cycles

Retry loops are intentional cycles. Let's verify the pipeline isn't broken elsewhere:

Analysis.cyclic?(resilient)
|> IO.inspect(label: "Has cycles?")

cyclic?/1 returns true here — but that's expected. The retry loop is a designed cycle, not a bug. Use the rendered graph and edge labels to distinguish intentional retry loops from accidental feedback.

Example 4: Multi-Stage Microservices with Sub-Pipelines

Complex systems have bounded contexts. Sub-pipeline clusters let you group related nodes visually while keeping the full graph connected for analysis.

microservices =
  dataflow do
    # Ingestion cluster
    mobile_app = source("Mobile App\n1_200 req/s", rate: 1_200)
    web_app = source("Web App\n800 req/s", rate: 800)
    api_gateway = merge("API Gateway")
    auth = transform("Auth Middleware\n15ms", latency_ms: 15)

    # Order service cluster
    cluster "order_service", label: "Order Service", fillcolor: "#eff6ff" do
      order_handler = transform("Order Handler\n80ms", latency_ms: 80)
      order_kafka = buffer("Order Events", capacity: 20_000)
      inventory_check = transform("Inventory Check\n60ms", latency_ms: 60)
      in_stock = conditional("In Stock?")
      payment = transform("Payment Service\n250ms", latency_ms: 250)
      order_db = sink("Order DB")
    end

    # Notification cluster
    cluster "notifications", label: "Notifications", fillcolor: "#fdf2f8" do
      email_worker = transform("Email Worker\n120ms", latency_ms: 120)
      push_worker = transform("Push Worker\n40ms", latency_ms: 40)
      notification_router = merge("Notify Router")
      sendgrid = sink("SendGrid")
      firebase = sink("Firebase FCM")
    end

    # Cross-cluster flows
    mobile_app ~> api_gateway |> on("HTTP")
    web_app ~> api_gateway |> on("HTTP")
    api_gateway ~> auth |> on("request")
    auth ~> order_handler |> on("authed request")
    order_handler ~> order_kafka
    order_kafka ~> inventory_check
    inventory_check ~> in_stock
    in_stock ~> payment |> on("yes")
    in_stock ~> order_db |> on("no")
    payment ~> order_db
    order_db ~> email_worker |> on("order confirmed")
    order_db ~> push_worker |> on("order confirmed")
    email_worker ~> notification_router
    push_worker ~> notification_router
    notification_router ~> sendgrid
    notification_router ~> firebase
  end

Kino.Layout.tabs(
  Siren:
    Choreo.Lab.Siren.new(Dataflow.to_mermaid(microservices, theme: Dataflow.theme(:default))),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(microservices, theme: Dataflow.theme(:default))),
  Sketch:
    Choreo.Lab.Sketch.new(Dataflow.to_mermaid(microservices, theme: Dataflow.theme(:default)))
)

Choreo.Lab.Siren.new(Dataflow.to_mermaid(microservices, direction: :lr))

Cluster-Aware Analysis

Clusters are visual only — the underlying graph is still one connected structure, so analysis works across boundaries:

IO.inspect(Analysis.bottlenecks(microservices), label: "Bottlenecks")
IO.inspect(Analysis.orphan_nodes(microservices), label: "Orphans")

{:ok, path, latency} = Analysis.longest_path(microservices)
IO.puts("Critical path (#{latency}ms): #{Enum.join(path, " → ")}")

Per-Cluster Bottlenecks

You can also inspect simulation results for specific clusters by filtering on node IDs:

order_nodes = [:order_handler, :order_kafka, :inventory_check, :in_stock, :payment, :order_db]

Analysis.simulate(microservices)
|> Enum.filter(fn {id, _} -> id in order_nodes end)
|> Enum.each(fn {id, stats} ->
  IO.puts("  #{id}: Inbound #{stats.in_rate} | Outbound #{stats.out_rate} evt/s")
end)

Example 5: Fixing a Broken Pipeline

Start with a pipeline that has structural issues and use analysis as a diagnostic checklist.

broken =
  dataflow do
    ingress = source("Ingress")
    worker = transform("Worker")
    orphan_processor = transform("Orphan")
    db = sink("DB")
    never_reached = sink("Unreachable Sink")

    ingress ~> worker
    # worker -> db is forgotten!
    # orphan_processor has no inputs
    # never_reached has no inputs
  end

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(broken)),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(broken, rankdir: :tb), height: "500px"),
  Sketch: Choreo.Lab.Sketch.new(Dataflow.to_mermaid(broken))
)

Diagnosis

IO.inspect(Analysis.orphan_nodes(broken), label: "Orphans (unreachable from sources)")
IO.inspect(Analysis.dead_ends(broken), label: "Dead ends (cannot reach sinks)")
IO.inspect(Analysis.topological_sort(broken), label: "Topological sort")

The orphan processor and unreachable sink are immediately flagged. The missing worker → db connection means the database sink is also orphaned.

The Fix

fixed =
  dataflow do
    ingress = source("Ingress")
    worker = transform("Worker")
    db = sink("DB")

    ingress ~> worker
    worker ~> db
  end

IO.inspect(Analysis.orphan_nodes(fixed), label: "Orphans after fix")
IO.inspect(Analysis.dead_ends(fixed), label: "Dead ends after fix")

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(fixed)),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(fixed)),
  Sketch: Choreo.Lab.Sketch.new(Dataflow.to_mermaid(fixed))
)

Advanced: Custom Theming

Build a theme that matches your organisation's brand colours.

brand_theme =
  Choreo.Theme.custom(
    colors: %{
      source: "#10b981",
      transform: "#3b82f6",
      buffer: "#f59e0b",
      conditional: "#8b5cf6",
      merge: "#ec4899",
      sink: "#ef4444"
    },
    node_fontcolor: "white",
    edge_color: "#94a3b8"
  )

Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(microservices, theme: brand_theme, direction: :lr)),
  Graphviz:
    Kino.VizJS.render(Dataflow.to_dot(microservices, theme: brand_theme),
      engine: "twopi"
    )
)

Example 8: Zooming and Architectural Views

When dealing with large dataflow pipelines, displaying every single transformation can be overwhelming. Choreo.Viewable allows you to "zoom out" to see the big picture, or "zoom in" for detail.

For Dataflow graphs, zoom levels are semantic:

• Level 0: Sources and Sinks only (High-level data movement)
• Level 1: Sources, Sinks, and primary Transforms (Mid-level flow)
• Level 2+: Everything (including Buffers, Conditionals, and Merges)

alias Choreo.View

zoom_flow =
  dataflow do
    ingest = source("Event Stream")
    kafka = buffer("Kafka Topic")
    parse = transform("Parser")
    filter = conditional("Is Valid?")
    enrich = transform("Enricher")
    db = sink("Data Warehouse")
    dlq = sink("Dead Letter Queue")

    ingest ~> kafka
    kafka ~> parse
    parse ~> filter
    filter ~> enrich |> on("yes")
    filter ~> dlq |> error("no")
    enrich ~> db
  end

# Zoom out to Level 0 (Sources and Sinks only)
high_level_view = View.zoom(zoom_flow, level: 0)

IO.puts("Zoom Level 0 (Sources and Sinks Only)")
Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(high_level_view)),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(high_level_view)),
  Sketch: Choreo.Lab.Sketch.new(Dataflow.to_mermaid(high_level_view))
)

When you zoom out, intermediate nodes are hidden, but their transitive dependencies are preserved as virtual edges (dashed light-grey lines) so you don't lose the structural relationship between the visible components!

# Zoom in to Level 1 (Sources, Sinks, and Transforms)
mid_level_view = View.zoom(zoom_flow, level: 1)

IO.puts("Zoom Level 1 (Sources, Sinks, and Transforms)")
Kino.Layout.tabs(
  Siren: Choreo.Lab.Siren.new(Dataflow.to_mermaid(mid_level_view)),
  Graphviz: Kino.VizJS.render(Dataflow.to_dot(mid_level_view)),
  Sketch: Choreo.Lab.Sketch.new(Dataflow.to_mermaid(mid_level_view))
)

Cheat Sheet

Lab DSL Syntax (Choreo.Lab.DSL.Dataflow)

Programmatic Pipe API & Analysis (Choreo.Dataflow)

Summary

Dataflow diagrams as code mean your pipeline architecture is version-controlled, reviewable, and analyzable. Every time you add a transform or buffer, you can immediately see the impact on latency, throughput, and bottlenecks — before you deploy.