Bento Documentation

Bento is a declarative data streaming service designed to solve data engineering problems using simple, chained, stateless processing steps.

Key characteristics include:

• Resiliency: Implements transaction-based resiliency with back pressure. When used with at-least-once sources and sinks, it guarantees at-least-once delivery without requiring message persistence during transit.
• Data Agnostic: It does not care about the data format, making it easy to integrate into existing infrastructure.
• Versatility: It can complement or act as a simpler alternative to integration frameworks, log aggregators, and ETL workflow engines.
• Extensibility: Uses a component-based architecture consisting of Inputs, Processors, and Outputs.

The parquet_encode processor encodes a batch of structured messages into Parquet files. It is currently marked as experimental and relies on the parquet-go library.

This processor is typically used within the batching configuration of an output (like aws_s3) to collect messages in memory and convert them into a Parquet file before upload.

parquet_encode:
  schema: [] # Required
  default_compression: uncompressed

The opensnowcat processor processes OpenSnowcat/Snowplow enriched TSV events. It can convert enriched TSV to flattened or nested JSON, filter events based on field values or schema property paths, and transform sensitive fields (PII) for privacy compliance via hashing, redaction, or IP anonymization.

Note: This component is experimental and subject to change.

Introduced in version 1.12.0.

pipeline:
  processors:
    - opensnowcat:
        output_format: json

The sql_raw output component allows you to execute SQL queries against various databases. It supports multiple drivers, connection pooling, batching, and initialization of tables via SQL files or statements.

Key configuration areas include:

• Connection: Define the driver and dsn (Data Source Name).
• Query Execution: Specify the query and use args_mapping (a Bloblang mapping) to provide values for placeholders.
• Initialization: Use init_files (glob patterns for .sql files) or init_statement (a single SQL string) to set up tables on the first connection.
• Batching: Configure how messages are grouped before being flushed using batching policies (count, byte size, or period).

The azure_cosmosdb input component executes a SQL query against an Azure CosmosDB instance and creates a batch of messages from each page of items returned.

Important Limitation: Cross-partition queries are not supported. You must specify the partition_keys_map in your configuration so the PartitionKey value(s) are known in advance for every query.

input:
  azure_cosmosdb:
    endpoint: http://localhost:8080
    account_key: YOUR_ACCOUNT_KEY
    database: blobbase
    container: blobfish
    partition_keys_map: root = "AbyssalPlain"
    query: SELECT * FROM blobfish AS b WHERE b.species = @species
    args_mapping: |
      root = [
          { "Name": "@species", "Value": "smooth-head" },
      ]

The aws_s3_stream output streams data to S3 using multipart uploads. Unlike the standard aws_s3 output, it streams content incrementally rather than buffering entire files in memory, making it ideal for large files (>100MB) or continuous streams in memory-constrained environments.

Warning: Delivery Guarantees This output weakens the delivery guarantees of the pipeline and should not be used where data loss is unacceptable.

Shutdown Behavior This output flushes on the shutdown of the stream. It is intended for use with inputs that have a logical end, such as the file input or an input wrapped with read_until.

The repository provides two different Dockerfile definitions depending on your requirements:

1. Pure Go Image (Dockerfile): Based on busybox. Use this for a lightweight footprint when CGO is not required.
2. CGO Enabled Image (Dockerfile.cgo): Based on debian. Use this if your configuration or plugins require CGO support.

The nlp_zero_shot_classify processor performs zero-shot text classification using a Hugging Face NLP pipeline with an ONNX Runtime model. This allows you to classify text into any set of labels without training on those specific labels by using Natural Language Inference (NLI). It is useful for sentiment analysis, topic classification, intent detection, and content moderation.

Note: This component is currently in BETA. Only models in ONNX format are supported.

pipeline:
  processors:
    - nlp_zero_shot_classify:
        path: "KnightsAnalytics/deberta-v3-base-zeroshot-v1"
        labels: ["fun", "dangerous", "boring"]

A dynamic output is a special broker type that allows outputs to be identified by unique labels. Unlike static outputs, dynamic outputs can be created, updated, or removed during runtime via a REST API.

When using a dynamic output, Bento uses the fan_out broker pattern, meaning every message processed by the dynamic output broker will be delivered to every individual dynamic output registered within it.