etl

type Action ¶

type Action string

Action tells the pipeline what to do after an error.

const (
	ActionFail Action = "fail" // Stop pipeline and return error
	ActionSkip Action = "skip" // Skip this record and continue
)

type Batcher ¶

type Batcher[T any] interface {
	// Batch groups items into batches for loading.
	Batch(items []T) [][]T
}

Batcher groups transformed records into batches for loading. Implement this interface on your job when the default size-based batching is insufficient.

The pipeline calls Batch after accumulating records from the transform stage. In streaming mode, Batch is called whenever the pending buffer reaches LoadBatchSize, and once more to flush remaining records. In epoch mode, Batch is called once per epoch with all transformed records from that epoch.

The default batcher (used when Batcher is not implemented) is equivalent to SizeBatcher with the resolved LoadBatchSize.

When to implement a custom Batcher:

• Records must be grouped by a key (e.g., tenant ID) so each Load call targets a single partition
• Batch size depends on record weight (e.g., SQL parameter count limits)
• You need stripe-based batching for parallel load workers

Ready-made batchers are available for common patterns:

• SizeBatcher: fixed number of items per batch
• GroupByField: group by a key extracted from each record
• GroupByFieldWithSizeLimit: group by key with a per-group size cap
• WeightedBatcher: batch by cumulative weight (e.g., SQL param limits)
• StripeBatcher: hash-stripe for parallel load workers
• NoBatcher: send everything in a single batch
• CombineBatchers: compose multiple strategies in sequence

Example:

func (j *MyJob) Batch(items []Target) [][]Target {
    return etl.WeightedBatcher(func(t Target) int { return 5 }, 65535).Batch(items)
}

func CombineBatchers[Target any](batchers ...Batcher[Target]) Batcher[Target]

// First group by category, then limit each group to 25 items
batcher := etl.CombineBatchers(
	etl.GroupByField(func(t Target) string { return t.Category }),
	etl.SizeBatcher[Target](25),
)

package main

import (
	"fmt"

	"github.com/bjaus/etl"
)

// Target is a target record type for examples.
type Target struct {
	ID       int
	Name     string
	Category string
	Size     int
}

func main() {
	items := []Target{
		{ID: 1, Category: "x"},
		{ID: 2, Category: "x"},
		{ID: 3, Category: "x"},
		{ID: 4, Category: "y"},
		{ID: 5, Category: "y"},
	}

	// First group by category, then split groups into batches of at most 2
	batcher := etl.CombineBatchers(
		etl.GroupByField(func(t Target) string { return t.Category }),
		etl.SizeBatcher[Target](2),
	)
	batches := batcher.Batch(items)

	fmt.Println("batches:", len(batches))
	for _, batch := range batches {
		fmt.Printf("  category=%s count=%d\n", batch[0].Category, len(batch))
	}

}

Output:

batches: 3
  category=x count=2
  category=x count=1
  category=y count=2

func GroupByField[Target any, K comparable](keyExtractor func(Target) K) Batcher[Target]

// Group all records by category
batcher := etl.GroupByField(func(t Target) string {
	return t.Category
})

package main

import (
	"fmt"

	"github.com/bjaus/etl"
)

// Target is a target record type for examples.
type Target struct {
	ID       int
	Name     string
	Category string
	Size     int
}

func main() {
	items := []Target{
		{ID: 1, Category: "x"},
		{ID: 2, Category: "y"},
		{ID: 3, Category: "x"},
		{ID: 4, Category: "y"},
	}

	batcher := etl.GroupByField(func(t Target) string { return t.Category })
	batches := batcher.Batch(items)

	fmt.Println("batches:", len(batches))
	for _, batch := range batches {
		fmt.Printf("  category=%s count=%d\n", batch[0].Category, len(batch))
	}

}

Output:

batches: 2
  category=x count=2
  category=y count=2

func GroupByFieldWithSizeLimit[Target any, K comparable](
	keyExtractor func(Target) K,
	maxGroupSize int,
) Batcher[Target]

// Group by category, but limit each group to 50 records
batcher := etl.GroupByFieldWithSizeLimit(
	func(t Target) string { return t.Category },
	50,
)

package main

import (
	"fmt"

	"github.com/bjaus/etl"
)

// Target is a target record type for examples.
type Target struct {
	ID       int
	Name     string
	Category string
	Size     int
}

func main() {
	items := []Target{
		{ID: 1, Category: "x"},
		{ID: 2, Category: "x"},
		{ID: 3, Category: "x"},
		{ID: 4, Category: "y"},
	}

	// Group by category, but cap each group at 2 items
	batcher := etl.GroupByFieldWithSizeLimit(
		func(t Target) string { return t.Category },
		2,
	)
	batches := batcher.Batch(items)

	fmt.Println("batches:", len(batches))
	for _, batch := range batches {
		fmt.Printf("  category=%s count=%d\n", batch[0].Category, len(batch))
	}

}

Output:

batches: 3
  category=x count=2
  category=x count=1
  category=y count=1

func NoBatcher[Target any]() Batcher[Target]

package main

import (
	"fmt"

	"github.com/bjaus/etl"
)

func main() {
	batcher := etl.NoBatcher[string]()
	batches := batcher.Batch([]string{"a", "b", "c"})
	fmt.Println(batches)

}

Output:

[[a b c]]

func SizeBatcher[Target any](maxSize int) Batcher[Target]

// Create batches of up to 100 items each
batcher := etl.SizeBatcher[MyRecord](100)

package main

import (
	"fmt"

	"github.com/bjaus/etl"
)

func main() {
	batcher := etl.SizeBatcher[string](2)
	batches := batcher.Batch([]string{"a", "b", "c", "d", "e"})
	fmt.Println(batches)

}

Output:

[[a b] [c d] [e]]

func StripeBatcher[T any, K comparable](keyFn func(T) K, numStripes int, maxBatchSize int) Batcher[T]

// 4 stripes (match load worker count), max 500 items per batch
batcher := etl.StripeBatcher(
    func(r Record) string { return r.TenantID },
    4,   // numStripes
    500, // maxBatchSize
)

package main

import (
	"fmt"

	"github.com/bjaus/etl"
)

// Target is a target record type for examples.
type Target struct {
	ID       int
	Name     string
	Category string
	Size     int
}

func main() {
	items := []Target{
		{ID: 1, Category: "a"},
		{ID: 2, Category: "b"},
		{ID: 3, Category: "a"},
		{ID: 4, Category: "c"},
		{ID: 5, Category: "b"},
		{ID: 6, Category: "a"},
	}

	// 2 stripes, max 10 items per batch
	batcher := etl.StripeBatcher(
		func(t Target) string { return t.Category },
		2,
		10,
	)
	batches := batcher.Batch(items)

	fmt.Println("batches:", len(batches))
	total := 0
	for _, batch := range batches {
		total += len(batch)
	}
	fmt.Println("total items:", total)

}

Output:

batches: 2
total items: 6

func WeightedBatcher[Target any](weigher func(Target) int, maxWeight int) Batcher[Target]

// Batch by SQL parameter count (5 columns per row, 65535 param limit)
batcher := etl.WeightedBatcher(func(t Target) int { return 5 }, 65535)

// Batch by estimated payload size
batcher := etl.WeightedBatcher(func(t Target) int { return len(t.Body) }, 10*1024*1024)

package main

import (
	"fmt"

	"github.com/bjaus/etl"
)

// Target is a target record type for examples.
type Target struct {
	ID       int
	Name     string
	Category string
	Size     int
}

func main() {
	items := []Target{
		{ID: 1, Size: 30},
		{ID: 2, Size: 30},
		{ID: 3, Size: 30},
		{ID: 4, Size: 30},
		{ID: 5, Size: 30},
	}

	// Batch so total size per batch does not exceed 70
	batcher := etl.WeightedBatcher(func(t Target) int { return t.Size }, 70)
	batches := batcher.Batch(items)

	for i, batch := range batches {
		total := 0
		for _, t := range batch {
			total += t.Size
		}
		fmt.Printf("batch %d: %d items, total size %d\n", i, len(batch), total)
	}

}

Output:

batch 0: 2 items, total size 60
batch 1: 2 items, total size 60
batch 2: 1 items, total size 30

type BatcherFunc ¶

type BatcherFunc[T any] func(items []T) [][]T

BatcherFunc adapts a plain function to the Batcher interface.

Example:

batcher := etl.BatcherFunc[Target](func(items []Target) [][]Target {
    // custom batching logic
    return [][]Target{items}
})

func (f BatcherFunc[T]) Batch(items []T) [][]T

type Checkpointer ¶

type Checkpointer[S any, C comparable] interface {
	// CheckpointInterval returns the number of records to process per epoch.
	CheckpointInterval() int

	// Cursor extracts a checkpoint cursor from a source record.
	// The cursor must increase monotonically with source ordering.
	Cursor(src S) C

	// LoadCheckpoint retrieves the last saved cursor and stats.
	// Return (nil, nil, nil) if no checkpoint exists (fresh start).
	LoadCheckpoint(ctx context.Context) (cursor *C, stats *Stats, err error)

	// SaveCheckpoint persists cursor and stats.
	// Called by the pipeline after each epoch completes successfully.
	SaveCheckpoint(ctx context.Context, cursor C, stats *Stats) error

	// ClearCheckpoint removes the saved checkpoint.
	// Called by the pipeline after successful completion.
	ClearCheckpoint(ctx context.Context) error
}

Checkpointer enables checkpoint-based resumability for long-running ETL jobs.

When implemented, the pipeline switches from streaming mode to epoch-based mode. Records are processed in epochs of CheckpointInterval() records:

1. Extract up to CheckpointInterval() records
2. Transform and load them with configured concurrency
3. Save a checkpoint (cursor + stats) after the epoch completes
4. Repeat from step 1 until the source is exhausted
5. Clear the checkpoint on successful completion

On restart, LoadCheckpoint returns the saved cursor and stats so the pipeline resumes from where it left off. Extract receives the cursor to skip already- processed records.

Use Checkpointer when:

• The dataset is large enough that re-processing from scratch is expensive
• The job may be interrupted (deploys, spot instances, timeouts)
• You need progress durability across process restarts

Idempotency requirement: because records near a checkpoint boundary may be re-processed on restart, the Load implementation should be idempotent (use UPSERT rather than INSERT). Cursor values must increase monotonically with source ordering so that resuming from a cursor skips the right records.

Choosing CheckpointInterval:

• Smaller values (500-2000): finer recovery granularity, more checkpoint writes
• Larger values (5000-10000): fewer checkpoint writes, more re-processing on restart
• Match to your source's natural page size when possible

Interaction with graceful shutdown: when the parent context is cancelled, the current epoch is allowed to complete (within DrainTimeout), and the checkpoint is saved before the pipeline exits. This means the next restart will not re-process the completed epoch.

Example:

func (j *MyJob) CheckpointInterval() int { return 5000 }

func (j *MyJob) Cursor(src Source) int64 { return src.ID }

func (j *MyJob) LoadCheckpoint(ctx context.Context) (*int64, *etl.Stats, error) {
    // Return (nil, nil, nil) for a fresh start
    row := j.db.QueryRowContext(ctx, "SELECT cursor, stats FROM checkpoints WHERE job = $1", j.id)
    var cursor int64
    var statsJSON []byte
    if err := row.Scan(&cursor, &statsJSON); err != nil {
        if errors.Is(err, sql.ErrNoRows) { return nil, nil, nil }
        return nil, nil, err
    }
    var stats etl.Stats
    if err := stats.UnmarshalJSON(statsJSON); err != nil { return nil, nil, err }
    return &cursor, &stats, nil
}

func (j *MyJob) SaveCheckpoint(ctx context.Context, cursor int64, stats *etl.Stats) error {
    data, _ := stats.MarshalJSON()
    _, err := j.db.ExecContext(ctx,
        `INSERT INTO checkpoints (job, cursor, stats) VALUES ($1, $2, $3)
         ON CONFLICT (job) DO UPDATE SET cursor = $2, stats = $3`, j.id, cursor, data)
    return err
}

func (j *MyJob) ClearCheckpoint(ctx context.Context) error {
    _, err := j.db.ExecContext(ctx, "DELETE FROM checkpoints WHERE job = $1", j.id)
    return err
}

type DrainTimeout ¶

type DrainTimeout interface {
	// DrainTimeout returns the maximum time to wait for in-flight operations
	// to complete after the parent context is cancelled.
	// A zero value disables graceful shutdown (immediate abort).
	DrainTimeout() time.Duration
}

DrainTimeout controls graceful shutdown behavior. When the parent context is cancelled (e.g., SIGTERM), the pipeline:

1. Stops extracting new records immediately
2. Allows in-flight transform/load operations to complete within the timeout
3. If operations complete within timeout: exits cleanly with nil error
4. If timeout expires: forces abort with error

Implement this interface to set the timeout from the job struct rather than the pipeline builder.

The value can be overridden at runtime via WithDrainTimeout, which takes precedence. If neither is set, DefaultDrainTimeout (5 minutes) is used.

Set to 0 to disable graceful shutdown entirely (immediate abort on context cancellation). Negative values passed to WithDrainTimeout are ignored.

When to customize:

• Jobs with fast loads (< 1s per batch): a short timeout (30s) is sufficient
• Jobs with slow external API calls: increase to match worst-case latency
• Jobs where partial progress is not useful: set to 0 to fail fast

In checkpoint mode, a successful drain saves the checkpoint before exiting, so the next restart picks up exactly where the shutdown left off.

Example:

func (j *MyJob) DrainTimeout() time.Duration { return 30 * time.Second }

type ErrorHandler ¶

type ErrorHandler interface {
	// OnError is called when an error occurs during any stage.
	// Return ActionSkip to continue processing, ActionFail to stop the pipeline.
	OnError(ctx context.Context, stage Stage, err error) Action
}

ErrorHandler customizes error handling per pipeline stage. Without an ErrorHandler, the pipeline stops on the first error in any stage.

Implement this interface when you want to:

• Log errors and continue processing (return ActionSkip)
• Apply different strategies per stage (e.g., skip extract errors, fail on load errors)
• Track error counts for alerting or metrics

Common patterns:

// Log-and-skip for transform, fail for load (data integrity)
func (j *MyJob) OnError(ctx context.Context, stage etl.Stage, err error) etl.Action {
    switch stage {
    case etl.StageExtract:
        slog.Warn("skipping malformed record", "error", err)
        return etl.ActionSkip
    case etl.StageTransform:
        slog.Error("transform error", "error", err)
        return etl.ActionSkip
    case etl.StageLoad:
        return etl.ActionFail
    }
    return etl.ActionFail
}

Skipped errors still increment Stats.Errors. The err parameter passed to Stopper.Stop only contains the fatal error that caused the pipeline to fail (i.e., when OnError returned ActionFail or when no ErrorHandler is present).

type Expander ¶

type Expander[S, T any] interface {
	Expand(ctx context.Context, src S) ([]T, error)
}

Expander converts one input record to multiple output records. Use this when a single source record needs to produce a variable number of target records.

Returning an empty or nil slice effectively filters the record out — no target records are produced and nothing reaches the load stage.

When to use:

• Denormalization: a source row with nested items produces one target per item
• Splitting: a source record contains multiple logical entities
• Conditional expansion: some records produce targets, others don't

Example:

func (j *MyJob) Expand(ctx context.Context, src Source) ([]Target, error) {
    targets := make([]Target, 0, len(src.Items))
    for _, item := range src.Items {
        targets = append(targets, Target{ParentID: src.ID, ItemID: item.ID})
    }
    return targets, nil
}

type Filter ¶

type Filter[S any] interface {
	// Include returns true if the record should be processed.
	// Returning false skips the record before it reaches the transform stage.
	Include(src S) bool
}

Filter excludes records before transformation. Implement this interface when you need to skip records based on their content without incurring the cost of transformation.

Filter runs in the extract stage, before any concurrency. This makes it the most efficient place to drop records because filtered records never enter the transform or load stages and never consume worker capacity.

Use Filter when you have:

• Soft-deleted records that should be ignored
• Records outside a target date range
• Inactive or disabled entities
• Any condition that can be evaluated from the source record alone

Example:

func (j *MyJob) Include(src Source) bool {
    return !src.Deleted && src.UpdatedAt.After(j.since)
}

If you need access to context or want to produce errors, consider filtering inside Transformer or Expander instead. Returning an empty slice from Expander has the same effect, but the record still passes through the transform stage.

Filter interacts with Checkpointer: in epoch mode, filtered records still advance the cursor so that resuming from a checkpoint does not re-process records that were already seen and skipped.

type Job ¶

type Job[S, T any, C comparable] interface {
	// Extract yields records from the source.
	// cursor is nil on fresh start, or the last checkpoint cursor on resume.
	// If Checkpointer[S, C] is not implemented, cursor is always nil.
	Extract(ctx context.Context, cursor *C) iter.Seq2[S, error]

	// Load writes a batch of records to the destination.
	// Should be idempotent (UPSERT) to handle potential re-processing.
	Load(ctx context.Context, batch []T) error
}

Job defines the core ETL operations. This is the only required interface to implement.

The type parameters are:

• S: source record type (extracted from the data source)
• T: target record type (loaded to the destination)
• C: cursor type for checkpoint-based resumability (use any if not needed)

For transformation, implement one of:

• Transformer: 1:1 transform (one input record -> one output record)
• Expander: 1:N transform (one input record -> multiple output records)

If both are implemented, Transformer takes precedence.

type LoadBatchSize ¶

type LoadBatchSize interface {
	// LoadBatchSize returns the number of records to batch before loading.
	LoadBatchSize() int
}

LoadBatchSize controls the number of records batched together before calling Load. Implement this interface to set the batch size from the job struct rather than the pipeline builder.

The value can be overridden at runtime via WithLoadBatchSize, which takes precedence. If neither is set, DefaultLoadBatchSize (100) is used.

This value is used as the default batch size when no custom Batcher is implemented. When a custom Batcher is present, this value is ignored in favor of the Batcher's own logic.

Tuning guidance:

• For SQL INSERTs: balance between fewer round-trips (larger batches) and staying within parameter limits (e.g., PostgreSQL's 65535 param limit)
• For API calls: match the external API's preferred batch size
• For file writes: larger batches reduce I/O overhead

Example:

func (j *MyJob) LoadBatchSize() int { return 500 }

type LoadWorkers ¶

type LoadWorkers interface {
	// LoadWorkers returns the number of concurrent load workers.
	LoadWorkers() int
}

LoadWorkers controls worker parallelism for the load stage. Implement this interface to set the concurrency level from the job struct rather than the pipeline builder.

The value can be overridden at runtime via WithLoadWorkers, which takes precedence. If neither is set, DefaultLoadWorkers (1) is used.

Tuning guidance:

• Match to your database connection pool size to avoid pool exhaustion
• For batch INSERTs/UPSERTs, 2-4 workers often saturates a single database
• Consider using StripeBatcher with matching stripe count to avoid lock contention when multiple workers write to the same table

Example:

func (j *MyJob) LoadWorkers() int { return 4 }

type Pipeline ¶

type Pipeline[S, T any, C comparable] struct {
	// contains filtered or unexported fields
}

Pipeline orchestrates the ETL process.

func New[S, T any, C comparable](job Job[S, T, C]) *Pipeline[S, T, C]

package main

import (
	"context"
	"fmt"
	"iter"

	"github.com/bjaus/etl"
)

// Source is a source record type for examples.
type Source struct {
	ID   int
	Name string
}

// Target is a target record type for examples.
type Target struct {
	ID       int
	Name     string
	Category string
	Size     int
}

type basicJob struct {
	rows []Source
}

func (j *basicJob) Extract(_ context.Context, _ *int) iter.Seq2[Source, error] {
	return func(yield func(Source, error) bool) {
		for _, r := range j.rows {
			if !yield(r, nil) {
				return
			}
		}
	}
}

func (j *basicJob) Transform(_ context.Context, src Source) (Target, error) {
	return Target{ID: src.ID, Name: src.Name + "!"}, nil
}

func (j *basicJob) Load(_ context.Context, batch []Target) error {
	for _, r := range batch {
		fmt.Printf("loaded: %d %s\n", r.ID, r.Name)
	}
	return nil
}

func main() {
	job := &basicJob{
		rows: []Source{
			{ID: 1, Name: "Alice"},
			{ID: 2, Name: "Bob"},
		},
	}

	err := etl.New[Source, Target, int](job).Run(context.Background())
	if err != nil {
		fmt.Println("error:", err)
	}

}

Output:

loaded: 1 Alice!
loaded: 2 Bob!

func (p *Pipeline[S, T, C]) Run(ctx context.Context) error

package main

import (
	"context"
	"fmt"
	"iter"

	"github.com/bjaus/etl"
)

// Source is a source record type for examples.
type Source struct {
	ID   int
	Name string
}

// Target is a target record type for examples.
type Target struct {
	ID       int
	Name     string
	Category string
	Size     int
}

type basicJob struct {
	rows []Source
}

func (j *basicJob) Extract(_ context.Context, _ *int) iter.Seq2[Source, error] {
	return func(yield func(Source, error) bool) {
		for _, r := range j.rows {
			if !yield(r, nil) {
				return
			}
		}
	}
}

func (j *basicJob) Transform(_ context.Context, src Source) (Target, error) {
	return Target{ID: src.ID, Name: src.Name + "!"}, nil
}

func (j *basicJob) Load(_ context.Context, batch []Target) error {
	for _, r := range batch {
		fmt.Printf("loaded: %d %s\n", r.ID, r.Name)
	}
	return nil
}

func main() {
	job := &basicJob{
		rows: []Source{
			{ID: 1, Name: "Alice"},
			{ID: 2, Name: "Bob"},
			{ID: 3, Name: "Charlie"},
		},
	}

	err := etl.New[Source, Target, int](job).
		WithLoadBatchSize(2).
		WithTransformWorkers(2).
		Run(context.Background())
	if err != nil {
		fmt.Println("error:", err)
	}

}

Output:

loaded: 1 Alice!
loaded: 2 Bob!
loaded: 3 Charlie!

func (p *Pipeline[S, T, C]) WithDrainTimeout(d time.Duration) *Pipeline[S, T, C]

func (p *Pipeline[S, T, C]) WithLoadBatchSize(n int) *Pipeline[S, T, C]

func (p *Pipeline[S, T, C]) WithLoadWorkers(n int) *Pipeline[S, T, C]

func (p *Pipeline[S, T, C]) WithReportInterval(n int) *Pipeline[S, T, C]

func (p *Pipeline[S, T, C]) WithTransformWorkers(n int) *Pipeline[S, T, C]

type ProgressReporter ¶

type ProgressReporter interface {
	ReportInterval

	// OnProgress is called periodically during execution.
	OnProgress(ctx context.Context, stats *Stats)
}

ProgressReporter receives periodic progress updates during pipeline execution. Implement this interface when you want to log throughput, emit metrics, or update an external dashboard while the pipeline is running.

OnProgress is called each time the cumulative loaded count crosses a ReportInterval boundary. In streaming mode this happens inside load workers; in epoch mode it happens during batch loading within each epoch.

The Stats snapshot passed to OnProgress is safe to read concurrently. Avoid performing blocking I/O inside OnProgress since it runs on a load worker goroutine.

Use ProgressReporter when:

• You want periodic log lines showing loaded/error counts
• You need to push throughput metrics to Prometheus, Datadog, etc.
• You want a heartbeat to detect stalled pipelines

Example:

func (j *MyJob) ReportInterval() int { return 10000 }

func (j *MyJob) OnProgress(ctx context.Context, stats *etl.Stats) {
    slog.InfoContext(ctx, "progress",
        "extracted", stats.Extracted(),
        "loaded", stats.Loaded(),
        "errors", stats.Errors(),
    )
}

To override the interval at runtime without changing the job struct, use WithReportInterval on the pipeline builder instead.

type ReportInterval ¶

type ReportInterval interface {
	// ReportInterval returns how often to call OnProgress (in records loaded).
	ReportInterval() int
}

ReportInterval controls how often progress is reported, measured in records loaded. This interface can be implemented independently of ProgressReporter when you want to set the interval via the job struct rather than the builder.

The value can be overridden at runtime via WithReportInterval, which takes precedence over this interface. If neither is set, DefaultReportInterval (10,000 records) is used.

This interface is embedded in ProgressReporter, so implementing ProgressReporter automatically satisfies ReportInterval.

Example:

func (j *MyJob) ReportInterval() int { return 5000 }

type Stage ¶

type Stage string

Stage identifies where in the pipeline an event occurred.

const (
	StageExtract   Stage = "extract"
	StageTransform Stage = "transform"
	StageLoad      Stage = "load"
)

type Starter ¶

type Starter interface {
	// Start is called before extraction begins.
	// The returned context is used for the entire pipeline.
	Start(ctx context.Context) context.Context
}

Starter is called before pipeline execution begins. Implement this interface when you need to perform setup work or enrich the context before extraction starts.

Use Starter for:

• Adding values to the context (request IDs, trace spans, logger fields)
• Recording the pipeline start time for elapsed-time metrics
• Acquiring resources that must be held for the pipeline's lifetime
• Logging the start of a pipeline run

The context returned by Start is propagated to all pipeline stages and to Stopper.Stop. This makes it the right place to attach tracing spans or cancellation signals.

Example:

func (j *MyJob) Start(ctx context.Context) context.Context {
    j.startedAt = time.Now()
    slog.InfoContext(ctx, "pipeline starting")
    return ctx
}

Start is called exactly once, before the first call to Extract.

type Stats ¶

type Stats struct {
	// contains filtered or unexported fields
}

Stats provides pipeline statistics with thread-safe access. Counter fields use atomic operations for safe concurrent access from worker goroutines.

func NewStats(extracted, filtered, transformed, loaded, errors int64) *Stats

func (s *Stats) Errors() int64

func (s *Stats) Extracted() int64

func (s *Stats) Filtered() int64

func (s *Stats) Loaded() int64

func (s *Stats) LogValue() slog.Value

func (s *Stats) MarshalJSON() ([]byte, error)

func (s *Stats) Transformed() int64

func (s *Stats) UnmarshalJSON(data []byte) error

type Stopper ¶

type Stopper interface {
	// Stop is called exactly once, after the pipeline Run method returns.
	Stop(ctx context.Context, stats *Stats, err error)
}

Stopper is called after pipeline execution completes, regardless of whether the pipeline succeeded, failed, or was shut down gracefully. Implement this interface for cleanup, final logging, or metrics reporting.

Use Stopper for:

• Logging final stats and elapsed time
• Reporting success/failure metrics to an observability system
• Releasing resources acquired in Starter
• Sending completion notifications

The ctx passed to Stop is derived from the pipeline's drain context, which remains valid even during graceful shutdown. This allows Stop to perform cleanup operations (database writes, API calls) after the parent context has been cancelled.

The err parameter is the same error value returned by Run: the unrecoverable error that caused Run to fail (no ErrorHandler, or ErrorHandler returned ActionFail). Errors handled with ActionSkip do not appear in err, even though they increment stats.Errors while the pipeline continues processing.

Example:

func (j *MyJob) Stop(ctx context.Context, stats *etl.Stats, err error) {
    elapsed := time.Since(j.startedAt)
    if err != nil {
        slog.ErrorContext(ctx, "pipeline failed", "error", err, "stats", stats, "elapsed", elapsed)
    } else {
        slog.InfoContext(ctx, "pipeline complete", "stats", stats, "elapsed", elapsed)
    }
}

Stop is called exactly once, after the pipeline Run method returns.

type TransformWorkers ¶

type TransformWorkers interface {
	// TransformWorkers returns the number of concurrent transform workers.
	TransformWorkers() int
}

TransformWorkers controls worker parallelism for the transform stage. Implement this interface to set the concurrency level from the job struct rather than the pipeline builder.

The value can be overridden at runtime via WithTransformWorkers, which takes precedence. If neither is set, DefaultTransformWorkers (1) is used.

Tuning guidance:

• CPU-bound transforms (parsing, hashing, serialization): set to runtime.NumCPU()
• I/O-bound transforms (HTTP calls, cache lookups): set higher (10-50) depending on the external service's capacity
• Pure field mapping with no I/O: 1 worker is usually sufficient since the overhead of goroutine coordination outweighs the parallelism benefit

Example:

func (j *MyJob) TransformWorkers() int { return runtime.NumCPU() }

type Transformer ¶

type Transformer[S, T any] interface {
	Transform(ctx context.Context, src S) (T, error)
}

Transformer converts one input record to one output record. Use this for simple 1:1 mappings where each source record produces exactly one target record.

When to use Transformer vs Expander:

• Transformer: field mapping, format conversion, enrichment — one in, one out
• Expander: denormalization, splitting — one source record produces a variable number of output records (including zero, which acts as a filter)

Example:

func (j *MyJob) Transform(ctx context.Context, src Source) (Target, error) {
    return Target{
        ID:   src.ID,
        Name: strings.ToUpper(src.Name),
    }, nil
}