The fast, config-driven way to move data in Rust.

Move data at Rust speed, govern it in flight, and ship it as a single binary. On a 1M-row CSV→JSONL move, faucet sustains 712k rows/s in 11.8 MiB of RAM — ~96× faster and ~62× less memory than Meltano, output identical row-for-row (see the benchmarks). No Python runtime, no platform to stand up, no daemon to babysit.

faucet-stream is a data-movement platform for Rust — with governance built in: 37 source and 29 sink connectors (66 in total) plus in-flight transforms, including a page-level embedded-DuckDB sql transform — wired by a single faucet binary that runs pipelines declaratively from YAML/JSON (no Rust code required), or embedded in your own service through the typed Source / Sink traits. One platform, whether you want a CLI you can drop on any box or a library you compile in.

📖 Guide · 📊 Benchmarks · 📜 Connector spec (FCP v0)

brew install faucet-hq/faucet-stream/faucet-cli   # the CLI — prebuilt, no Rust needed
# — or —
curl -LsSf https://github.com/faucet-hq/faucet-stream/releases/latest/download/faucet-cli-installer.sh | sh
# — or —
cargo install faucet-cli          # build the CLI from source
# — or —
cargo add faucet-stream           # the library

• 🚀 Built for throughput — native streaming with bounded memory, connection pooling, multi-row inserts, bulk APIs, and parallel I/O. Throughput is a first-class design goal for every connector. On single-machine batch throughput, faucet runs ~1–2 orders of magnitude faster than a Python Singer runtime: a reproducible 1M-row CSV→JSONL move (a best case that maximally exposes Python's per-row overhead) hit 712k rows/s in 11.8 MiB vs Meltano's 7.4k rows/s / 724 MiB (~96× faster, ~62× less memory, exact row parity); sink-bound moves like Postgres→Postgres narrow the gap. See BENCHMARKS.md for the methodology, the sink-bound scenario, and honest caveats.
• 🔌 Adopt incrementally — bring your Singer taps — the singer source runs any existing Singer/Meltano tap unchanged, so you can start with the taps you already have and move to native connectors where throughput matters. Experimental (v0): single-stream today, and a bridged tap still runs its own Python process.
• 🧩 Config-driven or embeddable — run faucet run pipeline.yaml, or call Pipeline::new(&source, &sink).run().await? from Rust. Same orchestration either way.
• ⚙️ A runtime, not just connectors — incremental + resumable replication, change-data-capture, effectively-once delivery (idempotent dedup-on-resume), upsert/delete write modes, dead-letter queues, automatic retries, adaptive batch sizing, secrets-manager interpolation, cron scheduling, and an HTTP control plane with event-driven triggers — plus built-in Prometheus metrics + tracing spans, all with zero per-connector code.
• 🛡️ Governance in the movement path — the guardrails most pipelines bolt on downstream, native and zero-config: data-quality checks, versioned data contracts, PII masking (applied before any sink sees a row), schema-drift detection & policy, column-level lineage (OpenLineage) + a data-movement catalog, and freshness/volume SLA monitoring.
• 📦 Pay only for what you use — every connector is a Cargo feature, so a slim build can be just REST + JSONL, or pull in all 66 connectors with --features full.

Documentation: the faucet-stream guide (getting started, tutorials, cookbook, operations) · API reference on docs.rs · cli/README.md for the full config grammar.

• Quickstart — the CLI
• Quickstart — the library
• What's in the box
• Connectors
• How it compares
• When to use faucet-stream
• Architecture
• Performance
• Observability
• Feature flags
• Using faucet-stream as a Rust library
• Building custom connectors
• Project structure
• Contributing
• License

Just want to poke at it? From a clone, run ./scripts/try-local.sh — it builds a light feature set, runs a no-infrastructure demo (transforms, quality, contracts, masking, lineage, catalog, DLQ replay) against generated data, then leaves the web console running so you can browse Runs, Datasets, and Lineage. No Docker or cloud accounts needed.

Move data without writing any Rust:

cargo install faucet-cli
faucet init my_pipeline --source postgres --sink bigquery   # scaffold pipeline.yaml from schemas
faucet validate pipeline.yaml                               # parse + resolve secrets, no run
faucet doctor pipeline.yaml                                 # preflight: probe auth/network/permissions
faucet test tests/*.yaml                                    # offline fixture tests for pipeline logic
faucet run pipeline.yaml                                    # one-shot run to completion
faucet discover conn.yaml -o pipeline.yaml                  # introspect a database and generate a config
faucet backfill pipeline.yaml --from 2026-06-01 --to 2026-07-01 --window 1d   # resumable historical replay
faucet schedule pipeline.yaml                               # run on a cron schedule (add a schedule: block)
faucet serve --no-auth                                      # HTTP control plane: submit/poll/cancel runs over REST

A minimal config — fetch open GitHub issues and write them to JSON Lines:

# faucet.yaml — `faucet run` auto-discovers this file (and a sibling `.env`) in cwd
version: 1
pipeline:
  source:
    type: rest
    config:
      base_url: https://api.github.com
      path: /repos/faucet-hq/faucet-stream/issues
      method: GET
      auth: { type: api_key, config: { header: Authorization, value: "Bearer ${env:GITHUB_TOKEN}" } }
      query_params: { state: open }
      pagination: { type: LinkHeader }
      max_retries: 3
  transforms:
    - type: keys_case          # re-case every key: snake / camel / pascal / kebab / screaming_snake
      config: { mode: snake }
  sink:
    type: jsonl
    config:
      path: ./out/issues.jsonl

Run many invocations from one config with a matrix: block (independent fan-out, a parent/child DAG, or depends_on: completion ordering between rows), and bound concurrency with execution:. See cli/README.md for the full grammar, cli/examples/rest_to_bigquery_matrix.yaml for matrix fan-out, and cli/examples/rest_users_posts_dag.yaml for the DAG pattern. The cli/examples/ directory has runnable configs for every common source→sink combination.

Embed the same engine in a Rust service:

cargo add faucet-stream --features sink-jsonl   # default already has the REST source
cargo add tokio --features full

use faucet_stream::{Pipeline, RestStream, RestStreamConfig, PaginationStyle};
use faucet_stream::sink::jsonl::{JsonlSink, JsonlSinkConfig};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let source = RestStream::new(
        RestStreamConfig::new("https://api.example.com", "/v1/users")
            .records_path("$.data[*]")
            .pagination(PaginationStyle::Cursor {
                next_token_path: "$.meta.next_cursor".into(),
                param_name: "cursor".into(),
            }),
    )?;
    let sink = JsonlSink::new(JsonlSinkConfig::new("./users.jsonl"));

    let result = Pipeline::new(&source, &sink).run().await?;
    println!("Wrote {} records", result.records_written);
    Ok(())
}

More library recipes — pagination styles, OAuth2, streaming, incremental replication, partitions, transforms, and custom connectors — are in Using faucet-stream as a Rust library below and the library tutorial.

faucet-stream is a full data-movement runtime, not just a bag of connectors. Every capability below works across all connectors with zero per-connector code, and each is a one-block addition to your YAML:

All connector crates depend only on faucet-core, so any source pairs with any sink. See the connector capability matrix (streaming, resumable state, compression, auth per connector) and the choosing-a-connector guide for help picking between overlapping connectors (Postgres query vs CDC, S3 vs Parquet, Redis vs Kafka, …).

Support tiers (the Tier column below). A connector is Tier-1 ✅ when it invokes and passes the faucet-conformance battery in CI against the connector's real backend — valid config schema, bounded-memory streaming, bookmark round-trip, idempotent replay, truthful capabilities, errors-not-panics (see the Faucet Connector Protocol spec). That battery is the tiering mechanism — there is no separate scheme, and the Tier-1 set grows as more connectors wire it in. ᵐ marks a connector whose battery runs against a wiremock HTTP mock in CI rather than a live service (the rest, graphql, xml, elasticsearch, bigquery, snowflake, databricks sources and the http sink): the mock faithfully drives the paging / schema / error paths the checks assert, but it is not an end-to-end test against the real system. ᵉ marks a connector whose battery runs against an official emulator in Docker — a real implementation, not the managed service: the Cloud Spanner pair (Spanner emulator), the Pub/Sub source and sink (Pub/Sub emulator), and the Azure Blob sink (Azurite). Unmarked T1 ✅ connectors run against a real backend with no caveat — a local filesystem (delta, parquet, csv) or a testcontainers-launched real server (postgres, mysql, mongodb, redis, clickhouse, kafka, …). Tier-2 connectors are not conformance-certified in CI — their full battery can't run without a live cloud backend (the BigQuery / Snowflake / Elasticsearch sinks are tested against wiremock, which can't prove real idempotent dedup; the GCS source/sink need a real gRPC backend the emulator doesn't provide), or the connector's shape doesn't fit a check (the webhook source is buffer-shaped; the append-only Iceberg sink's terminal flush doesn't fit the effectively-once replay check). They still have their own extensive integration tests (wiremock / testcontainers) and are used in production — Tier-2 means "not certified," not "low quality." The Singer bridge is additionally experimental (v0, single-stream) ⚠️.

Tier: T1 ✅ = passes the faucet-conformance battery in CI; T2 = not yet wired into the battery (see the support-tiers note above).

CLI — prebuilt binaries (macOS arm64/x86_64, Linux x86_64/aarch64; no Rust toolchain required). Includes every first-party connector plus serve (web console), schedule, and lineage:

# Homebrew
brew install faucet-hq/faucet-stream/faucet-cli

# Shell installer (installs to ~/.cargo/bin by default, checksummed)
curl -LsSf https://github.com/faucet-hq/faucet-stream/releases/latest/download/faucet-cli-installer.sh | sh

# Or download an archive + SHA256 checksum from the latest faucet-cli GitHub Release

CLI — from source (full/custom feature sets, e.g. DuckDB SQL transforms, OTLP export):

cargo install faucet-cli                   # default feature set
cargo install faucet-cli --features full   # everything

Library:

# Everything (default includes the REST source)
cargo add faucet-stream

# All sources / all sinks / all connectors
cargo add faucet-stream --features source
cargo add faucet-stream --features sink
cargo add faucet-stream --features full

# Pick individual connectors
cargo add faucet-stream --features source-rest,sink-postgres,sink-s3

# Or depend on individual connector crates directly
cargo add faucet-source-rest
cargo add faucet-source-mongodb

There are many great data-movement tools. faucet-stream's niche is being a single fast native binary and an embeddable Rust library — config-driven, with no Python runtime, no platform to operate, and a typed library API when you want to compile pipelines into your own service.

¹ Singer CDC depends on the individual tap. ² The original Benthos is Apache-2.0; Redpanda Connect's maintained build is source-available. ³ "Effectively-once" = idempotent at-least-once: per-page commit tokens are committed atomically with the data so a resumed run drops duplicates — not distributed-consensus exactly-once (see delivery guarantees). Comparison reflects the general shape of each tool as of 2026-05 — check each project for current details.

For reference: Singer is a connector spec and Meltano is its most common runtime; both appear above. faucet-stream is a full ETL tool — it transforms data in flight as it moves (11 record transforms plus filter, explode, and CDC-unwrap stages and a page-level embedded-DuckDB sql transform for aggregation, joins, and filtering), not just extract-and-load. dbt is complementary, not a competitor: it models transformations in the warehouse on data already loaded (the "T" of ELT, at warehouse scale) — pair the two when you need heavy in-warehouse modeling on top of what faucet extracts, transforms, and loads.

Deep dives (detailed, and honest about where each tool wins): vs. Meltano · vs. Airbyte · vs. Singer.

Reach for it when:

• You want one fast static binary (or a Rust library) to move data between APIs, databases, object stores, and warehouses — without standing up a platform, scheduler, or Python environment.
• You want version-controlled, config-driven pipelines you can run anywhere: locally, in CI, behind cron, or inside another service.
• You need streaming with bounded memory, incremental/resumable replication, CDC, effectively-once delivery, data-quality assertions, retries, dead-letter queues, and metrics without hand-writing that plumbing.
• You're already in Rust and want typed Source/Sink traits you can embed and extend.

Look elsewhere (for now) when:

• You need a connector faucet-stream doesn't ship yet and can't write — Meltano (600+ Singer taps) and Airbyte (350+) have far broader catalogs today.
• You want a fully-managed, hosted service with a UI and a team operating it — Fivetran or Airbyte Cloud.
• Your job is heavy in-warehouse transformation modeling — a tested, version-controlled SQL DAG built on data already in the warehouse, at warehouse scale. That's dbt's domain; pair it with faucet-stream, which extracts, transforms in-flight, and loads.
• You need a continuous record-by-record streaming processor — Benthos / Redpanda Connect and Vector are purpose-built for that. faucet-stream runs discrete pipelines to completion; even the long-running modes (faucet schedule, faucet serve) orchestrate complete runs rather than a never-ending stream.

A Source streams batches of records, optional Transforms reshape them, and the Pipeline writes each batch to a Sink — bounding memory at one batch on both sides regardless of total volume. The pipeline also drives the cross-cutting runtime (bookmarks, dead-letter routing, quality checks, metrics) so connectors stay simple:

%%{init: {'theme':'base','flowchart':{'curve':'basis','nodeSpacing':50,'rankSpacing':72,'padding':14},'themeVariables':{'fontFamily':'-apple-system,BlinkMacSystemFont,Segoe UI,sans-serif','fontSize':'14px','lineColor':'#a5b4c4','clusterBkg':'#f8fafc','clusterBorder':'#e2e8f0'}}}%%
flowchart LR
    S["<b>Source</b><br/>REST · DB · CDC<br/>Kafka · S3 · Parquet"]
    T["<b>Transforms</b><br/>flatten · rename · keys_case<br/>select · drop · set · cast<br/>redact · hash · json_parse · coalesce<br/>value_case · split · join · spell_symbols<br/>sql (DuckDB, page-level)"]
    P{{"<b>Pipeline</b>"}}
    K["<b>Sink</b><br/>BigQuery · Postgres<br/>Parquet · Kafka · ..."]
    ST[("State store<br/>file · Redis · Postgres")]
    D[("Dead-letter<br/>queue")]
    O(["Prometheus<br/>+ tracing"])

    S -->|StreamPage batches| T --> P -->|write_batch| K
    P -.->|bookmark per page| ST
    ST -.->|resume from bookmark| S
    P -.->|failed rows| D
    P -.->|metrics + spans| O
    classDef src fill:#e0f2f1,stroke:#26a69a,stroke-width:1.5px,color:#00695c
    classDef proc fill:#eceff8,stroke:#7986cb,stroke-width:1.5px,color:#303f9f
    classDef dec fill:#fff3e0,stroke:#ffa726,stroke-width:1.5px,color:#e65100
    classDef bad fill:#fdecec,stroke:#ef9a9a,stroke-width:1.5px,color:#c62828
    classDef store fill:#f3e5f5,stroke:#ab47bc,stroke-width:1.5px,color:#6a1b9a
    classDef sink fill:#e3f2fd,stroke:#42a5f5,stroke-width:1.5px,color:#1565c0
    class S src
    class T,O proc
    class P dec
    class D bad
    class ST store
    class K sink

faucet-stream is a Cargo workspace with 92 crates — 37 sources, 29 sinks, 16 shared connector libraries, the shared auth-provider library, 2 state-store backends, the lineage crate, the SQL transform crate, the conformance test battery, the shared core, the umbrella crate, and the CLI binary. See the Connectors table above and the architecture guide.

Performance and reliability are why this library exists. Every connector is optimised for throughput out of the box — there are no "slow defaults" to tune away.

Every pipeline emits OTel-compatible tracing spans and Prometheus metrics automatically — labelled by pipeline, row (matrix row id), and connector, with zero per-connector code. The CLI exposes a /metrics endpoint via the optional observability: block in faucet.yaml. See the CLI README for the YAML grammar and the OpenTelemetry bridge snippet.

Default features: source-rest, transform-flatten, transform-rename-keys, transform-keys-case. Every connector and capability is an opt-in Cargo feature.

The CLI is just one consumer of the engine. Everything it does is available as a typed API.

use faucet_stream::{RestStream, RestStreamConfig, Auth, PaginationStyle};

// Cursor-based pagination with Bearer auth
let stream = RestStream::new(
    RestStreamConfig::new("https://api.example.com", "/v1/users")
        .auth(Auth::Bearer { token: "my-token".into() })
        .records_path("$.data[*]")
        .pagination(PaginationStyle::Cursor {
            next_token_path: "$.meta.next_cursor".into(),
            param_name: "cursor".into(),
        })
        .max_pages(50),
)?;
let users: Vec<serde_json::Value> = stream.fetch_all().await?;

// Page-number pagination with an API key
let stream = RestStream::new(
    RestStreamConfig::new("https://api.example.com", "/v2/orders")
        .auth(Auth::ApiKey { header: "X-Api-Key".into(), value: "secret".into() })
        .records_path("$.results[*]")
        .pagination(PaginationStyle::PageNumber {
            param_name: "page".into(),
            start_page: 1,
            page_size: Some(100),
            page_size_param: Some("per_page".into()),
        }),
)?;

use faucet_stream::{Auth, fetch_oauth2_token};

// OAuth2 client credentials
let token = fetch_oauth2_token(
    "https://auth.example.com/oauth/token",
    "client-id",
    "client-secret",
    &["read:data".into()],
).await?;
let config = RestStreamConfig::new("https://api.example.com", "/data")
    .auth(Auth::Bearer { token });

use faucet_stream::{RestStream, RestStreamConfig, ReplicationMethod, PaginationStyle};
use futures::StreamExt;
use serde::Deserialize;
use serde_json::json;

// Stream page-by-page (bounded memory)
let mut pages = stream.stream_pages();
while let Some(result) = pages.next().await {
    let records = result?;
    println!("processing page of {} records", records.len());
}

// Incremental replication — only fetch records newer than a stored bookmark
let stream = RestStream::new(
    RestStreamConfig::new("https://api.example.com", "/events")
        .records_path("$.data[*]")
        .replication_method(ReplicationMethod::Incremental)
        .replication_key("updated_at")
        .start_replication_value(json!("2024-06-01T00:00:00Z")),
)?;
let (records, bookmark) = stream.fetch_all_incremental().await?;  // persist `bookmark` for next run

// Typed deserialization straight into your structs
#[derive(Debug, Deserialize)]
struct User { id: u64, name: String, email: String }
let users: Vec<User> = stream.fetch_all_as::<User>().await?;

use faucet_stream::{
    KeyCaseMode, Labels, RecordTransform, RestStream, RestStreamConfig, Source, TransformingSource,
};

let inner = RestStream::new(
    RestStreamConfig::new("https://api.example.com", "/data").records_path("$.results[*]"),
)?;
let stream = TransformingSource::new(
    Box::new(inner) as Box<dyn Source>,
    vec![
        RecordTransform::Flatten { separator: "__".into() },         // {"user":{"id":1}} -> {"user__id":1}
        RecordTransform::KeysCase { mode: KeyCaseMode::Snake },        // re-case every key
        RecordTransform::RenameKeys { pattern: r"^_sdc_".into(), replacement: "".into() },
        RecordTransform::custom(|mut record| {                         // arbitrary closure
            if let serde_json::Value::Object(ref mut map) = record {
                map.insert("_source".to_string(), serde_json::json!("my-api"));
            }
            record
        }),
    ],
    Labels::for_named("rest"),
)?;

use faucet_core::config::{load_json, load_env, load_env_file};
use faucet_source_rest::RestStreamConfig;

let source: RestStreamConfig = load_json("source_config.json")?;       // from a JSON file
let source: RestStreamConfig = load_env("REST")?;                       // from REST_* env vars
let source: RestStreamConfig = load_env_file(".env", "REST")?;          // from a .env file

// Every source/sink can print a full JSON Schema of its config — always in sync with the code
let schema = RestStream::new(source)?.config_schema();
println!("{}", serde_json::to_string_pretty(&schema)?);

faucet-stream is a marketplace ecosystem — third-party developers can publish their own faucet-source-* / faucet-sink-* crates with minimal friction. The only dependency you need is faucet-core; it re-exports everything required (async_trait, serde_json, Value, json!, JsonSchema, schema_for!).

use faucet_core::{async_trait, FaucetError, Source, Sink, Value, json, JsonSchema, schema_for};

#[async_trait]
impl Source for MySource {
    async fn fetch_all(&self) -> Result<Vec<Value>, FaucetError> {
        Ok(vec![json!({"id": 1, "name": "example"})])
    }
    fn config_schema(&self) -> Value {
        serde_json::to_value(schema_for!(MySourceConfig)).expect("schema serialization")
    }
}

#[async_trait]
impl Sink for MySink {
    async fn write_batch(&self, records: &[Value]) -> Result<usize, FaucetError> {
        Ok(records.len())
    }
    fn config_schema(&self) -> Value {
        serde_json::to_value(schema_for!(MySinkConfig)).expect("schema serialization")
    }
}

Any Source works with any Sink via Pipeline::new(&source, &sink).run().await?. Map your own failures to FaucetError variants (Source / Sink / Config, or Custom(boxed_err) to wrap any std::error::Error without losing the chain). Publish under the naming convention faucet-source-<name> / faucet-sink-<name>. Full walkthrough: authoring connectors.

To use a third-party connector from a faucet.yaml config (not just from Rust), build a custom faucet binary that registers it via PluginRegistry and faucet_cli::run_main — the connector then works as type: <name> across every CLI command with zero runtime overhead. See Custom binaries with third-party connectors and the runnable cli/examples/custom-cli/.

Cargo.toml                    — workspace manifest (<!--COUNT:crates-->92<!--/COUNT--> crates)
crates/
  core/                       — faucet-core: shared types, traits, pipeline, transforms, config
  auth/                       — faucet-auth: shared OAuth2 / token-endpoint providers
  source/                     — <!--COUNT:sources-->37<!--/COUNT--> source connectors (rest, graphql, xml, grpc, *-cdc, kafka, s3, azure-blob, redshift, clickhouse, pubsub, delta, databricks, singer, duckdb, sqs, nats, sftp, …)
  sink/                       — <!--COUNT:sinks-->29<!--/COUNT--> sink connectors (bigquery, iceberg, delta, postgres, parquet, kafka, redshift, clickhouse, pubsub, azure-blob, duckdb, sqs, nats, sftp, …)
  common/                     — <!--COUNT:common-->16<!--/COUNT--> shared connector libraries (bigquery, elasticsearch, gcs, kafka, snowflake, mssql, kinesis, spanner, delta, redshift, pubsub, clickhouse, azure, sqs, nats, sftp)
  state/                      — Redis- and Postgres-backed StateStore backends
  lineage/                    — faucet-lineage: OpenLineage event emission
  transform-sql/              — faucet-transform-sql: embedded DuckDB SQL transform
  transform-wasm/             — faucet-transform-wasm: WebAssembly (wasmtime) per-record transform
faucet-stream/                — umbrella crate with feature-gated re-exports
cli/                          — faucet-cli: `faucet` binary, YAML/JSON pipeline runner
  examples/                   — ready-to-run pipeline YAMLs
  tests/                      — assert_cmd + wiremock integration tests
examples/                     — repo-level examples: docker-compose infra stack + run index
  orchestration/              — ELT recipe: faucet (EL) + dbt (T) + Airflow/Dagster
Dockerfile                    — multi-stage image build (name-based connector selection)
deploy/                       — container + Kubernetes assets
  helm/faucet-stream/         — Helm chart (serve Deployment and/or run Job/CronJob)
scripts/                      — helper scripts (try-local.sh, build-image.sh, cleanup-artifacts.sh)
docs/book/                    — mdBook documentation site (source under docs/book/src)
.github/workflows/            — ci.yml, release-plz.yml, docs.yml, docker-images.yml
.github/assets/               — brand assets: logo, wordmark, social-preview banner, favicon

Using faucet-stream in production? Open a PR to add your team here — real adopters are the best signal for the next person deciding whether to bet a pipeline on it.

Contributions — core changes and third-party connectors alike — are welcome. See CONTRIBUTING.md for setup, the checks CI runs, and the add-a-connector checklist, and the authoring guide for building your own faucet-source-* / faucet-sink-* crate. Please review our Code of Conduct. To report a vulnerability, see SECURITY.md.

Licensed under either of

• Apache License, Version 2.0 (LICENSE-APACHE or http://www.apache.org/licenses/LICENSE-2.0)
• MIT license (LICENSE-MIT or http://opensource.org/licenses/MIT)

at your option.

Unless you explicitly state otherwise, any contribution intentionally submitted for inclusion in the work by you, as defined in the Apache-2.0 license, shall be dual licensed as above, without any additional terms or conditions.