Write Kafka

[wernerdaehn]

What would be really cool is the ability to write into Kafka.
Imagine I read from a source, transform all the data in DuckDB and then copy that data into Kafka.

This would entail

1. Read the latest version of a subject provided by the user from the schema registry.
2. If it exists and is compatible, use it. If it does not exist or must be evolved into a new version, create a schema based on the dataset. Only if the user allowed that, else generate an error.
3. Each row is serialized into Avro and published in Kafka.
4. The partition to be used is optionally provided by an integer column of the table.
5. The partition-offsets of each posted record is made available somehow. If the source was a table, we could add another column for the offset and the writer updates it.
6. Alternative would be to the optional feature to write into a commit topic. That is another topic in Avro where the transaction id and the max offset for each partition is stored.

[wernerdaehn]

Hi @rustyconover can we work on that together? I have the use case that data is being transformed inside DuckDB and the last step is to send all rows to Kafka.

I would suggest a staged approach:

Stage 1: Schema id are provided for key and value

COPY (
    SELECT id as "key.id", id, firstname, lastname from table
) TO 'tributary_write_topic' (
    KEY_FORMAT avro,
    VALUE_FORMAT avro,
    KEY_SCHEMA_ID 567,
    VALUE_SCHEMA_ID 568
);

You fetch the key and value schema by id from the schema registry and whatever columns exist in the schema and table, these values are taken. It can be that the schema has more columns, which must have defaults then.
The projection as columns named key_... which are for the key schema, all other columns are for the value schema.
If the projection has columns the key schema or value schema does not know about, an error is raised.

Example: Key schema with the id 567 has the field id, the value schema with the id 568 the fields id, firstname, lastname.
Thus the first row selected with key.id = 1, id = 1, firstname="Peter", lastname="Parker" is converted into the key schema {id: 1} taken from the key.id column and the other values go into the value schema.
If there would be a fifth column in the select, an error is thrown.

Datatypes are checked for compatibility also. Special care has to be taken for date/time data types, decimal(p,s), arrays and structs.

This allows us to create the key and/or value schema in Python, get back the schema id for each and then use these schemas to load the data. Maximum flexibility and yet straight forward for you to implement.

Stage 2: Create schema also by deriving the columns from the select statement.

COPY (
    SELECT id as "key.id", id, firstname, lastname from table
) TO 'tributary_write_topic' (
    KEY_FORMAT avro,
    VALUE_FORMAT avro,
    SUBJECT_KEY customer-key,
    SUBJECT_VALUE customer-value
);

Stage 3: Allow other key serialization also, e.g. key schema is an int with the key.id only and value serialization is Avro.

COPY (
    SELECT id as "key.id", id, firstname, lastname from table
) TO 'tributary_write_topic' (
    KEY_FORMAT int,
    VALUE_FORMAT avro,
    VALUE_SCHEMA_ID 568
);

or

COPY (
    SELECT id as "key.id", id, firstname, lastname from table
) TO 'tributary_write_topic' (
    KEY_FORMAT int,
    VALUE_FORMAT avro,
    SUBJECT_VALUE customer-value
);

to create only a value schema and the key schema is an int primitive.

Is there a better parameter specification? One that avoids specifying conflicting information like KEY_FORMAT int, SUBJECT_KEY customer-key which is illegal, because an int key payload does not use the schema registry.

[rustyconover]

Hi @wernerdaehn,

Good news: I already have Kafka writes implemented in Tributary with Avro, Protobuf, and JSON, including support for Confluent’s Schema Registry. I’m even testing against Confluent Cloud (and paying the bills there), so I’m quite motivated to get this fully polished.

Architecturally, I’ve decided to separate serialization/deserialization concerns from Kafka itself. Kafka just deals in BLOBs, while scalar functions handle serialization and deserialization in coordination with the Schema Registry. This gives us more flexibility and keeps the Kafka side clean.

I’ve also landed a few PRs in DuckDB core to support this work, including the ability for the table in/out functions to control the order in which rows are written to a topic.

I’m preparing for DuckDB Developer Day at the end of the month, and I’m planning a new Tributary release in early February. If you can wait until then, I think you’ll be very happy with where this lands.

Cheers,

Rusty

[wernerdaehn]

I like that approach of splitting Kafka and serialization into two functionalities. It is way more flexible than what I suggested. Looking forward!