faucet-sink-gcs

Google Cloud Storage sink for the faucet-stream ecosystem. Serializes batches of serde_json::Value records as JSON Lines (NDJSON) and uploads them concurrently to a GCS bucket with time-sortable UUIDv7 object keys.

Reach for it when you want to land data from any faucet-stream source — a database, an API, a queue, a CDC stream — into GCS as newline-delimited JSON, ready for BigQuery external tables, Dataflow, or any downstream reader. It is built on the official google-cloud-storage SDK and uses buffer_unordered to fan uploads out across the wire.

Feature highlights

• Concurrent uploads — each batch is chunked and the chunks upload in parallel via buffer_unordered(concurrency) (default 10), so throughput scales with your network rather than serializing one PUT at a time.
• Time-sortable keys — every object is named {prefix}{uuidv7}{file_extension}. UUIDv7 embeds a timestamp, so a bucket listing returns objects in write order without any extra index.
• Explicit NDJSON content type — uploads are tagged application/x-ndjson so consumers and tooling recognize the format.
• Flexible object sizing — combine batch_size (pipeline-level chunking) and max_records_per_file (a hard per-object cap); the sink uploads at whichever limit is smaller.
• Optional compression — gzip or zstd per object behind the compression feature, with auto-detection from the file extension.
• Apache Parquet (Arrow columnar) — behind the arrow feature, format: parquet writes each object as a complete, self-contained ZSTD-compressed Parquet file and enables the columnar fast path, so a Parquet/Delta source can stream Arrow RecordBatches straight through with no serde_json::Value in between. See Arrow columnar (Parquet) mode.
• Four credential modes — Application Default Credentials, a service-account key file, inline service-account JSON, or anonymous (emulators). The shared GcsCredentials enum is re-exported from faucet-common-gcs, so it matches the GCS source byte-for-byte.
• Client built once — the authenticated Storage client is constructed in new() and reused for every upload.

Installation

# As a library:
cargo add faucet-sink-gcs

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

# With compression support:
cargo install faucet-cli --features "sink-gcs,compression"

sink-gcs is opt-in — it is not part of the CLI's default feature set.

Quick start

# pipeline.yaml — faucet run pipeline.yaml
version: 1
pipeline:
  source:
    type: rest
    config:
      url: https://api.example.com/events
  sink:
    type: gcs
    config:
      bucket: my-bucket
      prefix: events/
      auth:
        type: application_default

faucet run pipeline.yaml

This uploads each batch of records as one or more events/{uuidv7}.jsonl objects in gs://my-bucket.

Configuration reference

Core

Batching & throughput

Compression (feature compression)

Advanced

Authentication

auth uses the shared GcsCredentials enum (the project-wide { type, config } shape, snake_case discriminators). It defaults to application_default when omitted.

# Application Default Credentials (workload identity, gcloud)
auth:
  type: application_default

# Service-account key file
auth:
  type: service_account_json_file
  config:
    path: /run/secrets/sa.json

# Inline service-account JSON via env indirection
auth:
  type: service_account_json_inline
  config:
    json: ${env:GCP_SA_JSON}

Examples

Large export with no re-chunking (recommended)

Let the source size its own pages and write one object per page — the standard, cost-efficient layout for GCS.

sink:
  type: gcs
  config:
    bucket: my-bucket
    prefix: warehouse/daily/
    auth: { type: application_default }
    batch_size: 0   # accept upstream pages as-is, one object per page

Hard cap per object with high concurrency

Bound each object at 50k records and fan 16 uploads out at once.

sink:
  type: gcs
  config:
    bucket: my-bucket
    prefix: events/2026/
    auth:
      type: service_account_json_file
      config: { path: /run/secrets/sa.json }
    max_records_per_file: 50000
    concurrency: 16

Date-partitioned, gzip-compressed output

Uses a ${now.*} token for a dated prefix and gzip via the file extension (requires the compression feature).

sink:
  type: gcs
  config:
    bucket: my-bucket
    prefix: events/dt=${now.date}/
    auth: { type: application_default }
    file_extension: .jsonl.gz
    compression: auto   # resolves to gzip from the .gz suffix
    batch_size: 0

Streaming & batching

The sink implements Sink::write_batch: the pipeline streams pages from the source, and each page is handed to the sink as it arrives, so peak memory is bounded by the page size rather than the total record volume.

Within a write_batch call the records are chunked by the effective chunk size — min(batch_size, max_records_per_file), where batch_size = 0 and max_records_per_file = None each mean "no limit" (usize::MAX). Each chunk becomes a single GCS object with a fresh UUIDv7 key. Chunks upload concurrently via buffer_unordered(concurrency) and try_collect, so the first upload error aborts the batch.

Recommended: batch_size = 0. Writing many small objects is a well-known cloud-storage anti-pattern — per-request overhead, slower downstream reads, and inflated LIST/GET costs. Let the source's batch_size drive object sizing, or set an explicit max_records_per_file if you need a hard cap.

Memory ceiling. Each chunk is buffered fully in memory (and, with compression enabled, briefly held as both the raw and compressed body) before a single-shot upload. Because up to concurrency chunks upload at once, peak memory is roughly concurrency × (chunk size) × ~2. A batch_size = 0 (or max_records_per_file = None) fed by a fetch_all-style source produces one very large chunk per write_batch call — pair batch_size = 0 with a streaming source that already sizes its pages, or set max_records_per_file / lower concurrency to cap peak memory.

Partial-failure caveat: because uploads abort on the first error, a batch that fails mid-flight may leave already-uploaded chunks in the bucket. The pipeline bookmark only advances after the whole batch confirms, so a resumed run re-uploads the batch (yielding new UUIDv7 keys for the chunks that already landed). This sink does not use resumable uploads.

Arrow columnar (Parquet) mode

Behind the crate-local arrow Cargo feature, format: parquet writes each object as a complete, self-contained Apache Parquet file (ZSTD-compressed) instead of JSON Lines. The sink implements the columnar write_batch_columnar fast path (RFC 0002 / #375): when the source is also Arrow-native — the Parquet or Delta Lake source, or the S3 / GCS source in file_format: parquet mode — and no Value-shaped transform is configured, records move end-to-end as Arrow RecordBatches with no serde_json::Value materialization.

If either end of the pipeline isn't Arrow-native — or a Value-shaped transform sits in between — the run transparently falls back to the JSON row path.

# gcs(parquet) → gcs(parquet) — runs Arrow end-to-end
pipeline:
  sink:
    type: gcs
    config:
      bucket: my-bucket
      prefix: events/parquet/
      auth: { type: application_default }
      format: parquet   # requires the `arrow` feature

Enable it with cargo add faucet-sink-gcs --features arrow (library) or cargo install faucet-cli --features "sink-gcs,arrow" (CLI).

Compression

Gated behind the crate-local compression Cargo feature. It adds the compression config field (none / gzip / zstd / auto, default auto). With auto, the codec is resolved from file_extension per upload, so .jsonl.gz triggers gzip and .jsonl.zst triggers zstd; an explicit codec that disagrees with the suffix logs a one-time warning.

sink:
  type: gcs
  config:
    bucket: my-bucket
    prefix: events/
    auth: { type: application_default }
    file_extension: .jsonl.zst
    compression: zstd

Content-Encoding is NOT set. Unlike a gzip object served with Content-Encoding: gzip, this sink uploads the compressed bytes with the application/x-ndjson content type and no encoding metadata. Downstream consumers must decompress the object explicitly (the .gz / .zst suffix is the signal).

Enable it with cargo add faucet-sink-gcs --features compression (library) or cargo install faucet-cli --features "sink-gcs,compression" (CLI).

Config loading & schema

Load config from YAML/JSON or environment. Inspect the full JSON Schema with:

faucet schema sink gcs

Library usage

use faucet_core::Sink;
use faucet_sink_gcs::{GcsCredentials, GcsSink, GcsSinkConfig};
use serde_json::json;

# async fn run() -> Result<(), Box<dyn std::error::Error>> {
let cfg = GcsSinkConfig::new("my-bucket")
    .prefix("events/")
    .auth(GcsCredentials::ApplicationDefault)
    .with_batch_size(0)
    .concurrency(16);

let sink = GcsSink::new(cfg).await?;
let written = sink
    .write_batch(&[json!({ "id": 1 }), json!({ "id": 2 })])
    .await?;
println!("uploaded {written} records");
# Ok(())
# }

Most users drive the sink through a Pipeline rather than calling write_batch directly — see the library tutorial.

How it works

1. new() validates batch_size, resolves GcsCredentials, and builds an authenticated data-plane Storage client once.
2. write_batch chunks the page by the effective chunk size and serializes each chunk to a JSON Lines byte buffer (one serde_json line + \n per record).
3. With the compression feature, each buffer is compressed in-memory before upload.
4. Each chunk gets a fresh UUIDv7 key ({prefix}{uuidv7}{file_extension}) and is uploaded single-shot via write_object(...).set_content_type("application/x-ndjson").send_unbuffered().
5. Chunk uploads run concurrently through buffer_unordered(concurrency); the first error aborts the batch.

The faucet doctor preflight probe (Sink::check) builds a control-plane StorageControl client and issues a non-mutating list_objects call capped at one result, so it verifies bucket reachability and credentials without writing anything.

Lineage dataset URI

gs://<bucket>/<prefix> — e.g. gs://my-bucket/events/.

Feature flags

Enable the connector itself in the CLI/umbrella via the sink-gcs feature.

Troubleshooting / FAQ

See also

• Connector reference & capability matrix
• Compression cookbook
• faucet-source-gcs — the matching GCS source.
• faucet-common-gcs — shared GcsCredentials enum and client builders.
• faucet-sink-s3 — the structurally equivalent AWS S3 sink.

License

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