DOCSEARCHbyAlgolia
Submit Docs

Bento Documentation

website·Indexed 29 days ago

https://warpstreamlabs.github.io/bento

Documentation for Bento, a data pipeline tool featuring inputs, outputs, and processors. It includes guides on Bloblang mapping language, streams mode with a REST API, unit testing via the `bento test` command, and integrations with AWS, GCP, and Azure. The platform supports various processors such as HTTP, SQL, cache, and subprocesses, along with batching and rate-limiting configurations.

Tokens
202.7K
Snippets
1.3K
Records
1.7K
Agent score
75%

What's inside Bento

  1. Overview of Bento Streams Mode API

    When Bento is run in streams mode, it exposes an HTTP REST API to manage independent streams of data. Each stream is isolated with its own input, buffer, pipeline, and output sections. Note that stream configurations cannot include resources directly; resources must be managed via the /resources/{type}/{id} endpoint.

  2. Overview of nlp_zero_shot_classify processor

    The nlp_zero_shot_classify processor performs zero-shot text classification, allowing text to be classified into labels without specific training on those labels. It utilizes Natural Language Inference (NLI) models to determine if a text entails a candidate label. This component is powered by the Hugot library, which runs Open Neural Network Exchange (ONNX) models. It is suitable for sentiment analysis, topic classification, intent detection, and content moderation.

  3. Introduction to Bloblang mapping

    Bloblang (or blobl) is a language for mapping data within Bento. It can be used as a processor, within function interpolations, or via the blobl CLI subcommand. A mapping defines how to create a new document by extracting data from an input document using assignments (path segments on the left, query on the right).
    # Execute Bloblang mappings on the command-line
cat data.jsonl | bento blobl 'foo.(bar | baz).buz'

  4. Overview of the Ristretto cache component

    The Ristretto cache stores key/value pairs in a memory-bound map. It is more efficient and appropriate for high-volume use cases than the standard memory cache. Note that the add command is non-atomic, making this cache unsuitable for deduplication tasks.

  5. Overview of Bento data pipeline tool

    Bento is a cloud-native data pipeline tool written in Go and deployed as a static binary. It uses declarative configuration to handle common data engineering tasks including mapping, multiplexing, windowing, and enrichments. It supports at-least-once delivery guarantees using an in-process transaction model without requiring disk-persisted state, making it resilient to crashes and server faults.

  6. Overview of the Elasticsearch processor

    The Elasticsearch processor queries an Elasticsearch index and replaces the message content with the result. It is currently experimental and subject to change. If a query fails, the message remains unchanged and an error is attached to the message, which can be handled using try or catch processors.

  7. Conditionally process messages using the switch processor

    The switch processor allows you to conditionally execute a set of child processors on messages based on their contents. It evaluates a list of cases; for each case, a Bloblang query is checked. If the query returns true (or is empty), the associated child processors are executed. If a check mapping throws an error, the message is flagged as failed and no further cases are tested.
    pipeline:
  processors:
    - switch:
        - check: this.type == "foo"
          processors:
            - mapping: root.processed = true

  8. Decode messages using Confluent Schema Registry

    The schema_registry_decode processor automatically decodes and validates messages by extracting a schema ID from the message and fetching the associated schema from a Confluent Schema Registry service. It supports Avro, Protobuf, and JSON schemas, including those with schema references. If a message fails to match the schema, it remains unchanged.
    pipeline:
  processors:
    - schema_registry_decode:
        url: "https://your-schema-registry:8081"

  9. Group messages by value using the group_by_value processor

    The group_by_value processor splits a single batch of messages into N smaller batches. Each resulting batch contains messages that share the same value determined by an interpolated string evaluated per message. This is useful for grouping messages by specific content or metadata fields to process them individually or route them to unique locations.

  10. Transform and filter messages using the jq processor

    The jq processor transforms and filters messages using jq queries. It uses the gojq library, meaning no external jq installation is required, though some execution differences from the jq CLI may exist.

    Key behaviors:

    • If a query emits no value, the message is filtered out.
    • If a query returns multiple values, the resulting message becomes an array of those values.
    • Message metadata is accessible within the query via the $metadata variable.
    • If a query fails, the message remains unchanged, the error is logged, and the message is flagged as failed for standard error handling.

  11. Insert rows into DuckDB using the Appender API

    The duckdb_append output component writes rows directly into DuckDB's columnar storage via the Appender API. This method is faster than sql_insert for bulk ingestion because it bypasses SQL parsing.

    Key Constraints:

    • The target table must exist before the component connects. Use init_statement or init_files to create the table if necessary.
    • Rows are flushed to disk on every WriteBatch call and upon component shutdown.

    Type Coercions: Values produced by args_mapping are coerced as follows:

    • JSON integers (json.Number) → int64
    • JSON decimals (json.Number) → float64
    • RFC3339 strings → time.Time (for TIMESTAMP columns)
    • Nested types (maps, slices) are recursively coerced for STRUCT and LIST columns
    • All other types are passed through unchanged.
    output:
  duckdb_append:
    dsn: /data/vault.duckdb
    table: vault_deposits
    columns: [ deposit_id, duck, coins, denomination, deposited_at ]
    args_mapping: |
      root = [
        this.deposit_id,
        this.duck,
        this.coins,
        this.denomination,
        this.deposited_at,
      ]
    init_statement: |
      CREATE TABLE IF NOT EXISTS vault_deposits (
        deposit_id   VARCHAR PRIMARY KEY,
        duck         VARCHAR NOT NULL,
        coins        BIGINT  NOT NULL,
        denomination VARCHAR NOT NULL,
        deposited_at TIMESTAMP NOT NULL
      )
    batching:
      count: 10000
      period: 5s

  12. Split message batches using the group_by processor

    The group_by processor splits a batch of messages into N batches. Each resulting batch contains messages that satisfy a specific Bloblang query. Messages that do not satisfy any specified group check are placed into their own separate group. This processor requires messages to be batched to function correctly.