Connectors / Databases

Elasticsearch

sink

Elasticsearch bulk index sink

Install

cargo install faucet-cli --features sink-elasticsearch
faucet init --sink elasticsearch
# Cargo.toml — use it as a library
cargo add faucet-sink-elasticsearch

Starter config

sink:
  type: elasticsearch
  config:
    base_url: ""    # REQUIRED — Base URL of the Elasticsearch cluster (e.g.
    index: ""    # REQUIRED — Target index name.
    auth:
      type: none    # one of: none, basic, bearer, api_key
      # --- Alternative variants — replace the block above with one of these ---
      # type: basic
      # config:
        # username: ""    # REQUIRED
        # password: ""    # REQUIRED
      #
      # type: bearer
      # config:
        # token: ""    # REQUIRED
      #
      # type: api_key
      # config:
        # key: ""    # REQUIRED
    # batch_size: 1000    # Maximum documents per `_bulk` HTTP request.
    # id_field: ""    # Optional JSON field name to use as the document `_id`.
    # write_mode: null    # Write semantics for a sink.
    # key: []    # Key columns.
    # delete_marker: null    # Optional.
    # rollback: null    # Runtime rollback settings (#706): the run id to journal under, the run-id column, and whether to journal before-images / keep a replaced table.
    # connect_timeout_secs: 10    # Seconds to wait for a TCP/TLS connection to the cluster.
    # request_timeout_secs: 300    # Seconds one HTTP request (a `_bulk`, `_delete_by_query`, mapping call) may take end to end before it fails as retriable.
    # tls: { ... }    # Server verification and client-certificate settings for a connector that dials a TLS server itself (Redis, NATS, Elasticsearch).
    # op_type: null    # The `_bulk` action an append write issues.

Capabilities

Delivery Effectively-once (keyed upsert)
Write modes append, upsert, delete, overwrite
Exactly-once writes No
Schema evolution Yes
Scoped cleanup Yes
Staged bulk load No
Run rollback No
Compression No

Config fields

FieldTypeDefaultDescription
base_url
required
string — Base URL of the Elasticsearch cluster (e.g. "http://localhost:9200").
index
required
string — Target index name.
auth
required
object — Authentication: either inline ({ type, config }) or a { ref: <name> } pointer to a shared provider in the CLI's top-level auth: catalog.
batch_size integer 1000 Maximum documents per _bulk HTTP request. Defaults to DEFAULT_BATCH_SIZE.
id_field string — Optional JSON field name to use as the document _id.
write_mode object append Append (default), upsert, or delete.
key array [] Key columns. Required and non-empty for upsert/delete; ignored for append.
delete_marker object — Optional. Upsert only: rows whose field matches one of values are deletes; all others are upserts. The marker field is stripped from upsert rows before writing.
rollback object — Runtime rollback settings (#706): the run id to journal under, the run-id column, and whether to journal before-images / keep a replaced table. Injected per invocation by the CLI when a config has a top-level rollback: block — not meant to be written by hand.
connect_timeout_secs integer 10 Seconds to wait for a TCP/TLS connection to the cluster. Default 10.
request_timeout_secs integer 300 Seconds one HTTP request (a _bulk, _delete_by_query, mapping call) may take end to end before it fails as retriable. Default 300, so a half-open connection or wedged node cannot hang a run forever.
tls object — TLS settings for an https:// cluster: a private CA, a client certificate for mutual TLS, or insecure_skip_verify.
op_type object index The _bulk action for append writes: index (default; overwrites a document with the same _id) or create (required by data streams, which reject index; a document whose _id already exists is left as it is and counted as written, so a replayed page stays idempotent). upsert / delete / overwrite always use index.
Get started

Your first pipeline runs in five minutes.

Install the CLI, scaffold a config, and move real data — nothing external to stand up.

curl -LsSf https://github.com/faucet-hq/faucet-stream/releases/latest/download/faucet-cli-installer.sh | sh
brew install faucet-hq/faucet-stream/faucet-cli