Skip to main content

sql

Executes an arbitrary SQL query for each message.

For each message this output runs the configured SQL query against the target database, using the selected driver and data_source_name. Query arguments can be built from message contents with an args_mapping.

# Config fields, showing default values
output:
  label: ""
  sql:
    driver: "" # No default (required)
    data_source_name: "" # No default (required)
    query: "" # No default (required)
    args_mapping: ""
    max_in_flight: 64
    batching:
      count: 0
      byte_size: 0
      period: ""
      check: ""
      processors: []

Examples​

Run an insert query for each message, populating the placeholder arguments from message fields and metadata.

output:
  sql:
    driver: mysql
    data_source_name: foouser:foopassword@tcp(localhost:3306)/foodb
    query: "INSERT INTO footable (foo, bar, baz) VALUES (?, ?, ?)"
    args_mapping: |
      root = [
        this.document.foo,
        this.document.bar,
        meta("kafka_topic"),
      ]

Fields​

driver​

A database driver to use, such as mysql, postgres, clickhouse, mssql, sqlite, oracle, snowflake, trino, gocosmos, or spanner.

Type: string

data_source_name​

A Data Source Name to identify the target database.

Type: string

query​

The SQL query to execute for each message. The placeholder style used for query arguments depends on the driver; for example mysql uses ? while postgres uses $1, $2, and so on.

Type: string

args_mapping​

An optional Bloblang mapping which should evaluate to an array of values matching in size to the number of placeholder arguments in the query.

Type: string

max_in_flight​

The maximum number of inserts to run in parallel.

Type: int
Default: 64

batching​

Allows you to configure a batching policy.

Type: object

batching.byte_size​

An amount of bytes at which the batch should be flushed. If 0 disables size based batching.

Type: int
Default: 0

batching.check​

A Bloblang query that should return a boolean value indicating whether a message should end a batch.

Type: string
Default: ""

batching.count​

A number of messages at which the batch should be flushed. If 0 disables count based batching.

Type: int
Default: 0

batching.period​

A period in which an incomplete batch should be flushed regardless of its size.

Type: string
Default: ""

batching.processors​

A list of processors to apply to a batch as it is flushed. This allows you to aggregate and archive the batch however you see fit. Please note that all resulting messages are flushed as a single batch, therefore splitting the batch into smaller batches using these processors is a no-op.

Type: array