Tributary DuckDB Extension

The Tributary extension by Query.Farm provides seamless integration between DuckDB and Apache Kafka, enabling real-time querying and analysis of streaming data. With this extension, users can consume messages directly from Kafka topics into DuckDB for immediate processing, as well as write processed data back to Kafka streams.

Features

• Direct Kafka Ingestion: Stream records from Kafka topics directly into DuckDB tables using SQL.
• SQL-Native Interface: Kafka integration is fully accessible via SQL, enabling easy adoption for data engineers and analysts.

Roadmap

• Flexible Topic Consumption: Supports consuming from specific partitions, offsets, or continuously from the latest messages.
• Real-Time Analytics: Perform analytical queries on streaming data as it arrives.
• Kafka Output: Optionally write results or processed data back to Kafka topics.

Tip

Some of this functionality is planned to be implemented. If you’re interested in additional functionality, please contact Query.Farm.

The extension is named Tributary because it is a bridge between the stream of data and the data lake.

Note

DuckDB extensions are modular plugins that enhance the core engine with powerful new capabilities like this.

Getting Started

Tributary is a DuckDB community extension maintained and supported by Query.Farm.

Install Tributary in DuckDB by running:

INSTALL tributary FROM community;

Then load it with:

LOAD tributary;

Functionality

Love ❤️ this DuckDB extension? You’ll Love This.

Get the best from Query.Farm — smart tips, powerful tools, and project updates sent directly to your inbox, but only when we’ve got something great to share.

When Should I Use This?

Use the Tributary extension when you want to interact with Apache Kafka.

Here’s an example of reading the entire contents of a Kafka topic:

SELECT *
FROM tributary_scan_topic('test-topic',
  "bootstrap.servers" := 'localhost:9092'
);

API

The extension adds these functions:

• Metadata
  • tributary_metadata
• Topics
  • tributary_scan_topic
• Version
  • tributary_version

---

tributary_metadata

This function returns the metadata of the Kafka server as a table.

Signature:

tributary_metadata([KAFKA CONNECTION OPTIONS...])

Named Parameters

The named parameters of this function configure the parameters used by librdkafka. All parameters are passed the VARCHAR type. The allowed parameter names are shown as part of tributary_scan_topic.

Return Value

Returns a table with this schema:

Column Name	Type	Description
brokers	STRUCT(id INTEGER, host VARCHAR, port INTEGER)[]	A list of brokers for this Kafka cluster.
topics	STRUCT(name VARCHAR, error VARCHAR, partitions STRUCT(id INTEGER, LEADER integer)[])[]	A list of topics that exist in this Kafka cluster.

Example

SELECT * FROM tributary_metadata(
  "bootstrap.servers" := 'localhost:9092'
);
brokers = [{'id': 1, 'host': localhost, 'port': 9092}]
 topics = [{'name': test-topic2, 'error': NULL, 'partitions': [{'id': 0, 'leader': 1}]}, {'name': test-topic, 'error': NULL, 'partitions': [{'id': 0, 'leader': 1}]}]

---

tributary_scan_topic

This function returns the entire contents of the topic. It uses multiple threads if there are multiple partitions.

Signature:

tributary_scan_topic(VARCHAR,
  [KAFKA OPTIONS...]
)

Positional Arguments

Name	Type	Description
topic_name	VARCHAR	The name of the topic to scan.

Named Parameters

The named parameters of this function configure the parameters used by librdkafka. All parameters are passed the VARCHAR type. The allowed parameter names are:

• allow.auto.create.topics
• api.version.fallback.ms
• api.version.request
• api.version.request.timeout.ms
• auto.commit.interval.ms
• batch.num.messages
• batch.size
• bootstrap.servers
• broker.address.family
• broker.address.ttl
• broker.version.fallback
• builtin.features
• check.crcs
• client.dns.lookup
• client.id
• client.rack
• client.software.name
• compression.codec
• connections.max.idle.ms
• coordinator.query.interval.ms
• debug
• delivery.report.only.error
• enable.auto.commit
• enable.auto.offset.store
• enable.gapless.guarantee
• enable.idempotence
• enable.metrics.push
• enable.partition.eof
• enable.random.seed
• enable.sasl.oauthbearer.unsecure.jwt
• enable.sparse.connections
• enable.ssl.certificate.verification
• enable_sasl_queue
• enabled_events
• fetch.error.backoff.ms
• fetch.max.bytes
• fetch.message.max.bytes
• fetch.min.bytes
• fetch.queue.backoff.ms
• fetch.wait.max.ms
• group.id
• group.protocol
• group.protocol.type
• heartbeat.interval.ms
• internal.termination.signal
• isolation.level
• log.connection.close
• log.queue
• log.thread.name
• log_cb
• log_level
• max.in.flight.requests.per.connection
• max.poll.interval.ms
• message.copy.max.bytes
• message.max.bytes
• message.send.max.retries
• metadata.max.age.ms
• metadata.recovery.strategy
• metadata.request.timeout.ms
• offset.store.method
• open_cb
• partition.assignment.strategy
• queue.buffering.backpressure.threshold
• queue.buffering.max.kbytes
• queue.buffering.max.messages
• queue.buffering.max.ms
• queued.max.messages.kbytes
• queued.min.messages
• receive.message.max.bytes
• reconnect.backoff.jitter.ms
• reconnect.backoff.max.ms
• reconnect.backoff.ms
• retry.backoff.max.ms
• retry.backoff.ms
• sasl.kerberos.kinit.cmd
• sasl.kerberos.min.time.before.relogin
• sasl.kerberos.principal
• sasl.kerberos.service.name
• sasl.mechanism
• sasl.mechanisms
• sasl.oauthbearer.client.id
• sasl.oauthbearer.client.secret
• sasl.oauthbearer.method
• sasl.oauthbearer.token.endpoint.url
• sasl.password
• sasl.username
• schema.registry.basic.auth.user.info
• schema.registry.url
• security.protocol
• session.timeout.ms
• socket.blocking.max.ms
• socket.connection.setup.timeout.ms
• socket.keepalive.enable
• socket.max.fails
• socket.nagle.disable
• socket.receive.buffer.bytes
• socket.send.buffer.bytes
• socket.timeout.ms
• socket_cb
• ssl.ca.certificate.stores
• ssl.ca.location
• ssl.certificate.location
• ssl.endpoint.identification.algorithm
• ssl.engine.id
• ssl.key.location
• ssl.key.password
• statistics.interval.ms
• sticky.partitioning.linger.ms
• test.mock.broker.rtt
• test.mock.num.brokers
• topic.metadata.propagation.max.ms
• topic.metadata.refresh.fast.cnt
• topic.metadata.refresh.fast.interval.ms
• topic.metadata.refresh.interval.ms
• topic.metadata.refresh.sparse
• transaction.timeout.ms
• transactional.id

Return Value

Returns a table with this schema:

Column Name	Type	Description
topic	VARCHAR	The name of the topic being scanned
partition	INTEGER	The parition number where the message exists.
offset	BIGINT	The offset inside of the partition where the message exists.
message	BLOB	The message read from the topic.

Example

SELECT * FROM tributary_scan_topic('test-topic',
  "bootstrap.servers" := 'localhost:9092'
);
┌────────────┬───────────┬────────┬─────────┐
│   topic    │ partition │ offset │ message │
│  varchar   │   int32   │ int64  │  blob   │
├────────────┼───────────┼────────┼─────────┤
│ test-topic │         0 │      0 │ test1   │
│ test-topic │         0 │      1 │ test2   │
│ test-topic │         0 │      2 │ test3   │
└────────────┴───────────┴────────┴─────────┘

---

tributary_version

Return the current version of the tributary extension.

Signature: tributary_version()

Example

SELECT tributary_version();
┌─────────────────────┐
│ tributary_version() │
│       varchar       │
├─────────────────────┤
│ 20250612.01         │
└─────────────────────┘

Longer Examples

NASA / IceCube Gravitational Wave Alerts

NASA operates the General Coordinates Network (GCN), which rapidly distributes alerts among space-based and ground-based observatories, physics experiments, and thousands of astronomers worldwide.

In this example, I’ll demonstrate how to retrieve the latest IceCube alerts related to gravitational waves detected by LIGO, Virgo, and KAGRA (LVK).

These alerts are published to a Kafka topic. Since the Kafka messages are binary blobs, I’ll walk through how to decode them into JSON. There’s also some minimal Kafka configuration required to read the topic, but GCN provides helpful instructions if you choose to sign up for access.

CREATE VIEW icecube_json_alerts AS
SELECT * EXCLUDE message, decode(message)::json as message FROM
tributary_scan_topic('gcn.notices.icecube.lvk_nu_track_search',
  "bootstrap.servers" := "kafka.gcn.nasa.gov",
  "group.id" := "test123",
  "sasl.mechanisms" := "OAUTHBEARER",
  "sasl.oauthbearer.client.id" := "YOUR_CLIENT_ID",
  "sasl.oauthbearer.client.secret" := "YOUR_SECRET",
  "sasl.oauthbearer.method" := "oidc",
  "sasl.oauthbearer.token.endpoint.url" := "https://auth.gcn.nasa.gov/oauth2/token",
  "security.protocol" := "sasl_ssl"
);

-- Parse out the JSON messages into their proper types.
CREATE VIEW icecube_alerts AS
SELECT
 message ->> '$.ref_ID' as ref_id,
(message ->> '$.alert_datetime')::timestamp as alert_datetime,
(message ->> '$.pval_bayesian')::double as pval_bayesian,
(message ->> '$.n_events_coincident')::integer as n_events_coincident,
(message ->> '$.neutrino_flux_sensitivity_range.flux_sensitivity')::double[] as flux_sensitivity,
(message ->> '$.neutrino_flux_sensitivity_range.sensitive_energy_range')::integer[] as sensitive_energy_range
  from icecube_json_alerts;

-- Now show the latest alerts.
SELECT alert_datetime, ref_id, pval_bayesian, flux_sensitivity, sensitive_energy_range
FROM icecube_alerts ORDER BY alert_datetime;
┌─────────────────────────┬───────────┬───────────────┬──────────────────┬────────────────────────┐
│     alert_datetime      │  ref_id   │ pval_bayesian │ flux_sensitivity │ sensitive_energy_range │
│        timestamp        │  varchar  │    double     │     double[]     │        int32[]         │
├─────────────────────────┼───────────┼───────────────┼──────────────────┼────────────────────────┤
│ 2025-06-11 17:20:31.765 │ S250611b  │        0.9898 │ [0.0277, 1.1333] │ [520, 24600000]        │
│ 2025-06-11 17:20:44.449 │ S250611b  │        0.9898 │ [0.0277, 1.1333] │ [520, 24600000]        │
│ 2025-06-11 17:52:40.219 │ S250611d  │         0.872 │ [0.0277, 0.9687] │ [508, 22900000]        │
│ 2025-06-11 18:04:54.18  │ S250611e  │        0.8576 │ [0.0372, 0.4914] │ [728, 22200000]        │
│ 2025-06-11 18:05:07.742 │ S250611f  │        0.9986 │ [0.0376, 0.35]   │ [755, 20900000]        │
│ 2025-06-11 18:05:20.1   │ S250611g  │        0.9991 │ [0.0384, 0.3367] │ [749, 20900000]        │
│ 2025-06-11 18:05:32.466 │ S250611h  │        0.6717 │ [0.0359, 0.4477] │ [613, 21900000]        │
│ 2025-06-11 18:05:44.753 │ S250611d  │        0.8954 │ [0.0277, 0.9687] │ [508, 22900000]        │
│ 2025-06-11 18:05:58.782 │ S250611e  │        0.8576 │ [0.0372, 0.4914] │ [728, 22200000]        │
│ 2025-06-11 18:41:47.742 │ S250611g  │        0.9991 │ [0.0384, 0.3367] │ [749, 20900000]        │
│ 2025-06-11 18:42:00.333 │ S250611h  │        0.6717 │ [0.0359, 0.4477] │ [613, 21900000]        │
│ 2025-06-11 22:50:48.777 │ S250611u  │        0.7604 │ [0.0277, 0.4666] │ [511, 22000000]        │
│ 2025-06-11 22:51:01.427 │ S250611u  │        0.7604 │ [0.0277, 0.4666] │ [511, 22000000]        │
│ 2025-06-12 01:48:27.681 │ S250612k  │         0.057 │ [0.0277, 1.1333] │ [626, 24100000]        │
│ 2025-06-12 01:48:40.172 │ S250612k  │         0.057 │ [0.0277, 1.1333] │ [626, 24100000]        │
│ 2025-06-12 06:30:23.042 │ S250612af │        0.7603 │ [0.0277, 1.0516] │ [553, 23100000]        │
│ 2025-06-12 06:30:35.553 │ S250612af │         0.751 │ [0.0277, 1.0516] │ [559, 23100000]        │
│ 2025-06-12 09:50:20.802 │ S250612aq │        0.3257 │ [0.0277, 0.6479] │ [716, 22900000]        │
│ 2025-06-12 09:50:34.371 │ S250612aq │        0.3257 │ [0.0277, 0.6479] │ [716, 22900000]        │
├─────────────────────────┴───────────┴───────────────┴──────────────────┴────────────────────────┤
│ 19 rows                                                                               5 columns │
└─────────────────────────────────────────────────────────────────────────────────────────────────┘