A thin pipeline abstraction over ExArrow streams.

ExArrow.Pipeline provides a stable, composable API for transforming and sinking Arrow streams without exposing the underlying execution mechanism. Internally it threads an Elixir Stream of ExArrow.RecordBatch values alongside the stream's schema, so transformations stay lazy and batches are never converted to row maps.

Every function accepts and returns {:ok, pipeline} | {:error, reason}, so pipelines compose with |>/2 directly from ExArrow.Stream.from_*/1 constructors (which return {:ok, stream} | {:error, _}). Errors short- circuit through every stage.

Example

ExArrow.Stream.from_flight_sql(client, "SELECT * FROM events")
|> ExArrow.Pipeline.map_batches(fn batch ->
  {:ok, slim} = ExArrow.Batch.select(batch, ["id", "score"])
  slim
end)
|> ExArrow.Pipeline.write_parquet("/data/events.parquet")

Laziness

map_batches/2 and each_batch/2 return a lazy pipeline — no batches are consumed until a sink (write_parquet/2, write_flight/3, or write_dataframe/1) runs. This means a pipeline built but never sunk does no work.

Telemetry

map_batches/2 and each_batch/2 emit [:ex_arrow, :pipeline, :batch] for every batch that flows through the stage, carrying rows, columns, and batch_count measurements and %{source: :pipeline} metadata. The sinks emit their respective [:ex_arrow, :parquet, :write] and [:ex_arrow, :flight, :query] events.