faucet-lineage

OpenLineage event emission for the faucet-stream ecosystem. Every pipeline run emits OpenLineage RunEvents describing the job, the run, its input/output datasets, their inferred schemas, and column-level lineage — to an HTTP endpoint (e.g. Marquez), a local JSON Lines file, or a Kafka topic.

Reach for it when you want your faucet-stream pipelines to show up in a lineage / data-catalog tool automatically — no per-connector instrumentation, no code in your data path. The faucet CLI wires it from a single top-level lineage: block; the types here are also usable directly by library callers via [LineageEmitter].

OpenLineage spec version: 2.0.2. Every event carries the pinned schema URL https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent.

Feature highlights

• Automatic lifecycle events — START / RUNNING / COMPLETE / ABORT / FAIL emitted around every pipeline run, with per-event toggles.
• Three transports — HTTP (with optional bearer auth), local JSON Lines file, or Kafka — all using the project-wide { type, config } shape.
• Schema facets — inferred (field, type) dataset schemas attached to inputs/outputs on terminal events, derived from a bounded record sample.
• Deterministic column-level lineage — per-output-field input edges folded over the declared transform chain, emitted only when the whole chain is mappable (never fabricated).
• Parent-job linkage — tie runs to an orchestrator job (Airflow, Dagster, …) via parent_job.
• Emission never fails a run — a broken lineage backend is logged and counted, then dropped; it can never abort or stall a pipeline.
• Cheap to enable — emission work happens in the CLI layer; the hot run_stream loop is untouched.

Installation

# As a library:
cargo add faucet-lineage

# In the CLI (opt-in feature — enables the top-level `lineage:` block):
cargo install faucet-cli --features lineage

# Add the Kafka transport:
cargo install faucet-cli --features lineage-kafka

lineage is a CLI-only feature and is not in the CLI default build — it is included in full. The lineage CLI feature enables HTTP + file transports; lineage-kafka additionally enables the Kafka transport (pulls in rdkafka).

Quick start

Add a top-level lineage: block to any pipeline config. The pipeline below pulls from PostgreSQL, shapes records, and lands them in BigQuery, emitting OpenLineage events to a Marquez endpoint on the way:

# pipeline.yaml — faucet run pipeline.yaml
version: 1
name: orders_load

lineage:
  namespace: prod.warehouse
  transport:
    type: http
    config:
      url: ${env:MARQUEZ_URL}   # e.g. http://localhost:5000/api/v1/lineage
  include_schema_facet: true
  include_column_lineage: true

pipeline:
  source:
    type: postgres
    config:
      connection_url: postgres://user:pass@localhost/app
      query: SELECT id, created_at, customer_email FROM orders
  transforms:
    - type: rename_field
      config:
        fields:
          customer_email: contact_email
  sink:
    type: jsonl
    config:
      path: ./orders.jsonl

faucet run pipeline.yaml

A full end-to-end example lives at cli/examples/postgres_to_bigquery_with_lineage.yaml.

OpenLineage event types

faucet-lineage emits standard OpenLineage RunEvents. The pipeline's lifecycle maps to event types as follows; each can be toggled independently via emit_on:

"Terminal facets" (schema + column-lineage) are attached only to the terminal events (COMPLETE / ABORT / FAIL), because they depend on the record sample observed during the run.

Configuration reference

The lineage: block deserializes into [LineageConfig]. Unknown top-level fields are rejected.

Transports

transport uses the project-wide { type, config: { … } } shape (adjacently tagged):

# HTTP — POST to an OpenLineage endpoint
transport:
  type: http
  config:
    url: https://marquez/api/v1/lineage
    timeout_secs: 10
    auth:
      type: bearer
      config:
        token: ${env:MARQUEZ_TOKEN}

# File — append JSON Lines locally (great for testing)
transport:
  type: file
  config:
    path: ./out/lineage.jsonl

# Kafka — requires the transport-kafka feature
transport:
  type: kafka
  config:
    brokers: localhost:9092
    topic: openlineage.events

HTTP transport auth

The nested auth uses the same { type, config } shape as connector auth. Only bearer is supported:

Parent job

parent_job links each run to an upstream orchestrator job, populating OpenLineage's parent run facet:

parent_job:
  namespace: airflow
  name: warehouse_dag.load_orders
  run_id: 0190-abc...   # optional

Emit toggles (emit_on)

Per-event booleans. Defaults: start, complete, fail, abort are true; running is false.

emit_on:
  start: true
  running: true        # opt into heartbeats
  complete: true
  fail: true
  abort: true
heartbeat_interval: 15

Schema facets

When include_schema_facet: true, the run wraps its source and sink in sampling wrappers ([SamplingSource] / [SamplingSink]) that clone up to sample_records records, infer an ordered (field, type) schema, and attach it to the input/output datasets on the terminal event.

Inferred OpenLineage types: null, boolean, integer, number, string, array, object.

Column-level lineage

When include_column_lineage: true, the declared transform chain is folded over the input field set ([derive_column_lineage]) to produce per-output-field input edges. Lineage is only emitted when every transform in the chain is mappable; if any transform is opaque, no column-lineage facet is attached for that run — it is never fabricated.

A set field with no upstream column produces a literal output field with no input edge.

Examples

File transport for local development

lineage:
  namespace: dev.local
  transport:
    type: file
    config:
      path: ./out/lineage.jsonl
  include_schema_facet: true

HTTP with bearer auth and an orchestrator parent

lineage:
  namespace: prod.warehouse
  job_name: ${name}::${row_id}
  transport:
    type: http
    config:
      url: ${env:MARQUEZ_URL}
      timeout_secs: 15
      auth:
        type: bearer
        config:
          token: ${env:MARQUEZ_TOKEN}
  parent_job:
    namespace: airflow
    name: warehouse_dag.load_orders
  include_schema_facet: true
  include_column_lineage: true

Heartbeats for long-running pipelines

lineage:
  namespace: prod.warehouse
  transport:
    type: http
    config:
      url: ${env:MARQUEZ_URL}
  emit_on:
    running: true
  heartbeat_interval: 30

Metrics

Emitted via the metrics facade (scraped through the standard faucet-stream observability exporter):

• faucet_lineage_events_total{event_type, outcome} — events sent; outcome ∈ ok | err.
• faucet_lineage_dropped_total{reason} — dropped events; reason ∈ disabled | transport_error.
• faucet_lineage_emit_duration_seconds{event_type} — per-event send-latency histogram.

Emission never fails a run

Lineage is observability, not a data path. [LineageEmitter::emit] returns no error: a transport failure (HTTP non-2xx, unreachable file, Kafka send error), a serialization failure, or a disabled event is logged and counted (faucet_lineage_dropped_total), then dropped. A broken lineage backend can never abort or stall a pipeline.

Library usage

Build a [LineageEmitter] from a [LineageConfig] and emit lifecycle events around your own run loop:

use faucet_lineage::{LineageConfig, LineageEmitter};
use faucet_lineage::event::EventType;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let cfg: LineageConfig = serde_json::from_value(serde_json::json!({
    "namespace": "prod.warehouse",
    "transport": {
        "type": "file",
        "config": { "path": "./out/lineage.jsonl" }
    }
}))?;

let emitter = LineageEmitter::new(cfg)?;   // Arc<LineageEmitter>

// Build a RunLifecycle for the run (job/run ids, input/output datasets, …),
// then emit around the run — emit() never returns an error.
// emitter.emit(EventType::Start, &lifecycle).await;
// emitter.emit(EventType::Complete, &lifecycle).await;
# Ok(())
# }

The CLI bridge (cli/src/lineage_glue.rs) does exactly this: it builds the emitter, wraps the source/sink in the sampling wrappers, maps resolved transform specs onto ColumnOps, and emits around run_stream.

How it works

• The only faucet-core touch-points are dataset_uri() on the Source / Sink traits (so each connector names its dataset) and redact_uri_credentials in faucet_core::util (so credentials never leak into a dataset URI). The hot run_stream loop is untouched.
• event.rs is a serde-faithful subset of the OpenLineage 2.0.2 RunEvent object model; emitter.rs builds events from a RunLifecycle and dispatches them through the configured transport.
• Each transport (transport/{http,file,kafka}.rs) implements a single send(Vec<u8>) method that serializes and ships one event; errors are returned to the emitter, which logs and drops them.
• Column lineage (column.rs) is a pure, deterministic fold over ColumnOps with no I/O — fully unit-tested.

Feature flags

In the CLI these surface as the lineage (HTTP + file) and lineage-kafka (adds Kafka) features.

Troubleshooting / FAQ

See also

• Lineage cookbook — end-to-end walkthrough.
• Config reference — the full lineage: grammar.
• OpenLineage · Marquez — the spec and a reference backend.

License

Licensed under either of Apache License, Version 2.0 or MIT license at your option.