faucet-source-csv

A CSV file source that reads delimited text files and yields each row as a serde_json::Value object. Part of the faucet-stream ecosystem.

Built on the streaming RFC-4180 csv-async reader so the file is consumed lazily, line by line, instead of being slurped into memory — a multi-gigabyte export streams through with memory bounded by batch_size, not file size. Reach for it to load CSV/TSV/pipe-delimited exports, spreadsheet dumps, or legacy flat files into any faucet-stream sink.

Feature highlights

• True streaming reader — Source::stream_pages reads from a tokio async reader and emits fixed-size pages; client-side memory is O(batch_size), not O(file size).
• Configurable dialect — set the field delimiter and quote characters (byte values) for CSV, TSV, pipe-delimited, or any single-char-delimited format.
• Header or headerless — with headers, columns are keyed by the header row; without, keys are generated as column_0, column_1, …. Duplicate header names (including two blank-named columns) are rejected with FaucetError::Config at read time — they would otherwise silently overwrite each other last-wins and drop column data for the whole run.
• Strict by default — a row whose field count differs from the header (a ragged/truncated line) aborts the run with FaucetError::Source naming the offending line, rather than silently emitting a record missing columns. Opt in to lenient parsing with flexible: true when ragged rows are expected.
• RFC-4180 correct — quoted fields containing embedded delimiters and embedded newlines parse as a single record, so files written by faucet-sink-csv round-trip losslessly.
• Transparent compression — opt-in compression feature reads .gz / .zst files, with auto-detection from the path suffix.
• No type inference — every field is returned as a JSON string; cast downstream with the cast transform if you need typed values.

Installation

# As a library:
cargo add faucet-source-csv
cargo add tokio --features full

# In the CLI (opt-in connector feature):
cargo install faucet-cli --features source-csv

# Via the umbrella crate:
cargo add faucet-stream --features source-csv

source-csv is not in the CLI default build — enable it explicitly (or use the source / full aggregate features).

Quick start

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

pipeline:
  source:
    type: csv
    config:
      path: ./data/input.csv

  sink:
    type: jsonl
    config:
      path: ./out/records.jsonl

faucet run pipeline.yaml

With a header row of id,name,email, the file produces records like {"id": "1", "name": "Alice", "email": "alice@example.com"}.

Configuration reference

Core

Batching

Format

Byte fields (delimiter, quote) are integers in JSON/YAML. Common values: 44 comma, 9 tab, 124 pipe, 59 semicolon, 34 double-quote, 39 single-quote.

Examples

TSV (tab-separated) file

source:
  type: csv
  config:
    path: /data/export.tsv
    delimiter: 9          # tab

Headerless, pipe-delimited legacy export

source:
  type: csv
  config:
    path: /data/legacy_export.csv
    has_headers: false
    delimiter: 124        # pipe |
    quote: 39             # single quote '

Rows become {"column_0": "...", "column_1": "...", ...}.

Cast string fields to typed values en route to SQLite

version: 1
name: csv_to_sqlite

pipeline:
  source:
    type: csv
    config:
      path: ./data/customers.csv

  transforms:
    - type: cast
      config:
        field: age
        to: integer

  sink:
    type: sqlite
    config:
      connection_url: sqlite://./customers.db
      table: customers
      auto_map: true

Tolerate ragged rows (opt-in lenient parsing)

source:
  type: csv
  config:
    path: /data/messy_export.csv
    flexible: true        # accept rows with uneven field counts

Without flexible: true (the default), a row shorter or longer than the header aborts the run with FaucetError::Source naming the offending line.

Compressed file (gzip), one page per file

source:
  type: csv
  config:
    path: /data/events.csv.gz
    compression: auto      # detects .gz; explicit `gzip` also works
    batch_size: 0          # emit the whole file as one page

Streaming & batching

CsvSource implements Source::stream_pages as a real client-side stream. It opens the file through tokio::fs::File, wraps it in a csv_async::AsyncReaderBuilder configured with the dialect (delimiter / quote / has_headers), and reads records incrementally — buffering up to batch_size rows before yielding a page. Because csv-async is a streaming RFC-4180 reader, it tracks quote state across physical lines, so quoted fields with embedded newlines and delimiters (e.g. "hello, world" or a multi-line cell) parse correctly as a single record.

batch_size = 0 is the no-batching sentinel: the file is fully drained and emitted in one StreamPage. Use it for small lookup tables, or for sinks (SQL COPY, BigQuery load jobs) that prefer one large request over many small ones.

fetch_all / fetch_with_context are convenience methods that collect every page into a single Vec<serde_json::Value> by draining stream_pages.

Every page carries bookmark: None — there is no incremental-replication mode for a flat file (the whole file is read on each run).

Compression

Behind the crate-local compression Cargo feature, the source gains a compression config field (none / gzip / zstd / auto, default auto). Auto-detection consults the path suffix at I/O time — .gz selects gzip, .zst selects zstd, anything else is treated as uncompressed. The streaming reader decompresses on the fly, so memory stays bounded and multi-line quoted records still parse correctly. Enable it with cargo install faucet-cli --features "source-csv compression" or cargo add faucet-source-csv --features compression.

Config loading

use faucet_core::config::{load_json, load_env_file};
use faucet_source_csv::CsvSourceConfig;

# fn example() -> Result<(), Box<dyn std::error::Error>> {
let config: CsvSourceConfig = load_json("config.json")?;
let config: CsvSourceConfig = load_env_file(".env", "CSV_SOURCE")?;
# Ok(()) }

JSON config:

{
  "path": "/data/exports/customers.csv",
  "has_headers": true,
  "delimiter": 44,
  "quote": 34,
  "batch_size": 1000
}

.env file:

CSV_SOURCE_PATH=/data/exports/customers.csv
CSV_SOURCE_HAS_HEADERS=true
CSV_SOURCE_DELIMITER=44
CSV_SOURCE_QUOTE=34

Schema introspection

faucet schema source csv

Prints the JSON Schema for CsvSourceConfig — the authoritative field list, types, and defaults.

Library usage

use faucet_source_csv::{CsvSource, CsvSourceConfig};
use faucet_core::Source;

# async fn example() -> Result<(), Box<dyn std::error::Error>> {
// TSV without a header row
let config = CsvSourceConfig::new("/data/export.tsv")
    .has_headers(false)
    .delimiter(b'\t')
    .with_batch_size(5000);

let source = CsvSource::new(config);
let records = source.fetch_all().await?;

for record in &records {
    println!("{record}");
}
# Ok(()) }

To run it through a pipeline, pair the source with any sink and drive it via faucet_core::Pipeline or faucet_core::run_stream.

How it works

• The source holds only the parsed CsvSourceConfig; there is no client or pool to reuse, since reads are local-filesystem I/O.
• stream_pages builds one csv_async::AsyncReader per run and pulls records through it incrementally, so the whole file is never resident in memory.
• All values are emitted as JSON strings — the reader does no type inference. Use the cast transform to coerce numbers/booleans downstream.
• Empty rows and the trailing newline are handled by the underlying RFC-4180 reader; an empty file yields zero records (not an error).

Lineage dataset URI

file://<path> — e.g. file:///data/input.csv.

Feature flags

The connector itself is enabled in the CLI/umbrella via the source-csv feature.

Troubleshooting / FAQ

See also

• Connector reference · Compression cookbook · Transforms cookbook
• Related crates: faucet-sink-csv · faucet-source-parquet · faucet-source-s3

License

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