Choreo.Lab.DSL.Dataflow (Choreo v0.12.0)

Copy Markdown View Source

Experimental Livebook-friendly DSL for sketching dataflow pipelines.

This Lab DSL compiles to the stable, pipe-first Choreo.Dataflow builders and returns an ordinary %Choreo.Dataflow{}. It uses node constructors for dataflow stages and typed edge vocabulary for normal, error, retry, and dead-letter paths.

Examples

iex> import Choreo.Lab.DSL.Dataflow
...> pipeline = dataflow do
...>   ingest = source("Kafka Ingest", rate: "10k/s")
...>   parser = transform("JSON Parser", latency_ms: 5)
...>   valid = conditional("Valid?")
...>   postgres = sink("Postgres")
...>   dlq = sink("Dead Letter Queue")
...>
...>   ingest ~> parser |> emits("raw events")
...>   parser ~> valid |> emits("parsed events")
...>   valid ~> postgres |> writes("valid records")
...>   dead_letter valid ~> dlq, "invalid records"
...> end
iex> pipeline.graph.nodes[:ingest].node_type
:source
iex> pipeline.edge_meta[{:valid, :dlq}].path_type
:dead_letter

Edge labels can use generic labels, typed data labels, or explicit path types:

source ~> transform |> on("events")
edge source ~> transform, data_type: "events"
emits source ~> transform, "events"
error transform ~> sink, "invalid input"
retry transform ~> buffer, "retry later"
dead_letter transform ~> dlq, "poison message"

Summary

Functions

Builds a %Choreo.Dataflow{} from a compact Lab DSL block.

Returns the vocabulary supported by the dataflow DSL.

Compatibility alias for taxonomy/0.

Types

cluster_decl()

@type cluster_decl() :: %{id: String.t(), opts: keyword()}

edge_decl()

@type edge_decl() :: %{from: Yog.node_id(), to: Yog.node_id(), opts: keyword()}

node_decl()

@type node_decl() :: %{id: Yog.node_id(), builder: atom(), opts: keyword()}

Functions

buffer(arg1 \\ nil, arg2 \\ nil, opts \\ [])

cluster(arg1 \\ nil, arg2 \\ nil, opts \\ [])

conditional(arg1 \\ nil, arg2 \\ nil, opts \\ [])

consumer(arg1 \\ nil, arg2 \\ nil, opts \\ [])

consumes(arg1 \\ nil, arg2 \\ nil, opts \\ [])

dataflow(list)

(macro)

Builds a %Choreo.Dataflow{} from a compact Lab DSL block.

dataflow(opts, list)

(macro)

dead_letter(arg1 \\ nil, arg2 \\ nil, opts \\ [])

decision(arg1 \\ nil, arg2 \\ nil, opts \\ [])

dlq(arg1 \\ nil, arg2 \\ nil, opts \\ [])

emits(arg1 \\ nil, arg2 \\ nil, opts \\ [])

error(arg1 \\ nil, arg2 \\ nil, opts \\ [])

flow(arg1 \\ nil, arg2 \\ nil, opts \\ [])

flows(arg1 \\ nil, arg2 \\ nil, opts \\ [])

input(arg1 \\ nil, arg2 \\ nil, opts \\ [])

join(arg1 \\ nil, arg2 \\ nil, opts \\ [])

lane(arg1 \\ nil, arg2 \\ nil, opts \\ [])

merge(arg1 \\ nil, arg2 \\ nil, opts \\ [])

normal(arg1 \\ nil, arg2 \\ nil, opts \\ [])

output(arg1 \\ nil, arg2 \\ nil, opts \\ [])

process(arg1 \\ nil, arg2 \\ nil, opts \\ [])

processor(arg1 \\ nil, arg2 \\ nil, opts \\ [])

producer(arg1 \\ nil, arg2 \\ nil, opts \\ [])

publishes(arg1 \\ nil, arg2 \\ nil, opts \\ [])

queue(arg1 \\ nil, arg2 \\ nil, opts \\ [])

reads(arg1 \\ nil, arg2 \\ nil, opts \\ [])

retry(arg1 \\ nil, arg2 \\ nil, opts \\ [])

routes(arg1 \\ nil, arg2 \\ nil, opts \\ [])

sends(arg1 \\ nil, arg2 \\ nil, opts \\ [])

sink(arg1 \\ nil, arg2 \\ nil, opts \\ [])

source(arg1 \\ nil, arg2 \\ nil, opts \\ [])

split(arg1 \\ nil, arg2 \\ nil, opts \\ [])

stage(arg1 \\ nil, arg2 \\ nil, opts \\ [])

taxonomy()

@spec taxonomy() :: %{
  clusters: [atom()],
  nodes: [atom()],
  edges: [atom()],
  modifiers: [atom()],
  options: [atom()]
}

Returns the vocabulary supported by the dataflow DSL.

This is meant as a lightweight Livebook discovery helper when autocomplete is not enough.

iex> taxonomy = Choreo.Lab.DSL.Dataflow.taxonomy()
iex> :source in taxonomy.nodes
true
iex> :emits in taxonomy.edges
true
iex> :dead_letter in taxonomy.edges
true

topic(arg1 \\ nil, arg2 \\ nil, opts \\ [])

transform(arg1 \\ nil, arg2 \\ nil, opts \\ [])

verbs()

@spec verbs() :: map()

Compatibility alias for taxonomy/0.

writes(arg1 \\ nil, arg2 \\ nil, opts \\ [])