Tansu is an Apache Kafka API compatible broker written in async 🚀 Rust 🦀 with PostgreSQL, S3 or memory storage engines.

Topics validated by JSON Schema, Apache Avro or Protocol buffers can be written as Apache Iceberg or Delta Lake tables

This repository showcases examples of structured data published to schema-backed topics, instantly accessible as Delta Lake tables.

Prerequisites:

• docker, using compose.yaml which runs tansu and MinIO
• duckdb, a fast open source database system that has native support for Delta Lake
• just, a handy way to save and run project-specific commands

The justfile contains recipes to create topics, produce and query data.

Once you have the prerequisites installed, clone this repository and start everything up with:

git clone git@github.com:tansu-io/example-delta-duckdb.git
cd example-pyiceberg

Start everything up with:

just up

Should result in:

✔ Network example-delta-duckdb_default    Created
✔ Volume "example-delta-duckdb_minio"     Created
✔ Container example-delta-duckdb-minio-1  Healthy
docker compose exec minio mc ready local
mc: Configuration written to `/tmp/.mc/config.json`. Please update your access credentials.
mc: Successfully created `/tmp/.mc/share`.
mc: Initialized share uploads `/tmp/.mc/share/uploads.json` file.
mc: Initialized share downloads `/tmp/.mc/share/downloads.json` file.
The cluster 'local' is ready
docker compose exec minio mc alias set local http://localhost:9000 minioadmin minioadmin
Added `local` successfully.
docker compose exec minio mc mb local/tansu
Bucket created successfully `local/tansu`.
docker compose exec minio mc mb local/lake
Bucket created successfully `local/lake`.
docker compose up --detach --wait tansu
✔ tansu Pulled
✔ Container example-delta-duckdb-tansu-1  Healthy

The above does the following:

• starts the MinIO S3 compatible service
• creates a s3://lake bucket in MinIO, used to store the Delta Lake tables
• creates a s3://tansu bucket in MinIO, used to store Kafka related data used by tansu
• runs the tansu broker configured to use MinIO as the storage engine with Delta Lake

Done! You can now run the examples.

Employee is a protocol buffer backed topic, with the following schema employee.proto:

syntax = 'proto3';

message Key {
  int32 id = 1;
}

message Value {
  string name = 1;
  string email = 2;
}

Sample employee data is in employees.json:

[
  {
    "key": { "id": 12321 },
    "value": { "name": "Bob", "email": "bob@example.com" }
  },
  {
    "key": { "id": 32123 },
    "value": { "name": "Alice", "email": "alice@example.com" }
  }
]

Create the employee topic:

just employee-topic-create

The above command will create an tansu.employee Delta Lake table, that is normalized and partitioned on the meta.year, meta.month, meta.day from the Kafka message:

Publish the sample data onto the employee topic:

just employee-produce

We can view the Delta Lake table created in s3://lake/tansu.employee with:

just minio-mc ls -r local/lake/tansu.employee

Note that the tansu.employee table, is partitioned on the meta.year, meta.month, meta.day:

docker compose exec minio mc ls -r local/lake/tansu.employee
[2025-05-30 06:56:41 UTC] 2.5KiB STANDARD _delta_log/00000000000000000000.json
[2025-05-30 06:56:41 UTC]   998B STANDARD _delta_log/00000000000000000001.json
[2025-05-30 06:56:41 UTC] 1.8KiB STANDARD meta.year=2025/meta.month=5/meta.day=30/part-00000-47182dcc-6071-4836-8233-1ae50678194e-c000.parquet

To view the Delta Lake table in DuckDB:

just employee-duckdb-delta

Giving the following output:

Grade is a JSON schema backed topic, with the following schema grade.json:

{
  "type": "record",
  "name": "Grade",

  "fields": [
    { "name": "key", "type": "string", "pattern": "^\\d{3}-\\d{2}-\\d{4}$" },
    {
      "name": "value",
      "type": {
        "type": "record",
        "fields": [
          { "name": "first", "type": "string" },
          { "name": "last", "type": "string" },
          { "name": "test1", "type": "double" },
          { "name": "test2", "type": "double" },
          { "name": "test3", "type": "double" },
          { "name": "test4", "type": "double" },
          { "name": "final", "type": "double" },
          { "name": "grade", "type": "string" }
        ]
      }
    }
  ]
}

Sample grade data is in: grades.json:

[
  {
    "key": "123-45-6789",
    "value": {
      "lastName": "Alfalfa",
      "firstName": "Aloysius",
      "test1": 40.0,
      "test2": 90.0,
      "test3": 100.0,
      "test4": 83.0,
      "final": 49.0,
      "grade": "D-"
    }
  },
  ...
]

Create the grade topic:

just grade-topic-create

The above command will create an tansu.grade Delta Lake table, that is normalized, partitioned on meta.year from the Kafka message, the Z Order of the data is value.grade :

Tansu will automatically maintain this table compacting small files and applying Z Ordering every 10 minutes or so.

Publish the sample data onto the grade topic:

just grade-produce

We can view the files created by Tansu in s3://lake/tansu.grade with:

just minio-mc ls -r local/lake/tansu.grade

Note that the tansu.grade table, is partitioned on the meta.year:

[2025-05-31 06:57:19 UTC] 3.4KiB STANDARD _delta_log/00000000000000000000.json
[2025-05-31 06:57:19 UTC] 1.4KiB STANDARD _delta_log/00000000000000000001.json
[2025-05-31 06:57:19 UTC] 5.4KiB STANDARD meta.year=2025/part-00000-bd6bce1a-0288-4ab2-9a40-01dc0bff2199-c000.parquet

View the data in DuckDB:

just grade-duckdb-delta

Giving the following output, note that the grade is unordered:

After maintenance has run the table is ordered by grade:

Observation is an Avro backed topic, with the following schema observation.avsc:

{
  "type": "record",
  "name": "observation",
  "fields": [
    { "name": "key", "type": "string", "logicalType": "uuid" },
    {
      "name": "value",
      "type": "record",
      "fields": [
        { "name": "amount", "type": "double" },
        { "name": "unit", "type": "enum", "symbols": ["CELSIUS", "MILLIBAR"] }
      ]
    }
  ]
}

Sample observation data, is in: observations.json:

[
  {
    "key": "1E44D9C2-5E7A-443B-BF10-2B1E5FD72F15",
    "value": { "amount": 23.2, "unit": "CELSIUS" }
  },
  ...
]

Create the observation topic:

just observation-topic-create

Publish the sample data onto the observation topic:

just observation-produce

View the data in DuckDB:

just observation-duckdb-delta

Giving the following output:

Person is a JSON schema backed topic, with the following schema person.json:

{
  "title": "Person",
  "type": "object",
  "properties": {
    "key": {
      "type": "string",
      "pattern": "^\\d{3}-\\d{2}-\\d{4}$"
    },
    "value": {
      "type": "object",
      "properties": {
        "firstName": {
          "type": "string",
          "description": "The person's first name."
        },
        "lastName": {
          "type": "string",
          "description": "The person's last name."
        },
        "age": {
          "description": "Age in years which must be equal to or greater than zero.",
          "type": "integer",
          "minimum": 0
        }
      }
    }
  }
}

Sample person data, is in persons.json:

[
  {
    "key": "123-45-6789",
    "value": { "lastName": "Alfalfa", "firstName": "Aloysius", "age": 21 }
  },
  ...
]

Create the person topic:

just person-topic-create

Publish the sample data onto the person topic:

just person-produce

View the data in DuckDB:

just person-duckdb-delta

Giving the following output:

Search is a protocol buffer backedd topic, with the following schema search.proto:

syntax = 'proto3';

enum Corpus {
  CORPUS_UNSPECIFIED = 0;
  CORPUS_UNIVERSAL = 1;
  CORPUS_WEB = 2;
  CORPUS_IMAGES = 3;
  CORPUS_LOCAL = 4;
  CORPUS_NEWS = 5;
  CORPUS_PRODUCTS = 6;
  CORPUS_VIDEO = 7;
}

message Value {
  string query = 1;
  int32 page_number = 2;
  int32 results_per_page = 3;
  Corpus corpus = 4;
}

Sample search data, is in searches.json:

[
  {
    "value": {
      "query": "abc/def",
      "page_number": 6,
      "results_per_page": 13,
      "corpus": "CORPUS_WEB"
    }
  }
]

Create the search topic:

just search-topic-create

Publish the sample data onto the search topic:

just search-produce

View the data in DuckDB:

just search-duckdb-delta

Giving the following output:

Taxi is a protocol buffer backed topic, with the following schema taxi.proto:

syntax = 'proto3';

enum Flag {
    N = 0;
    Y = 1;
}

message Value {
  int64 vendor_id = 1;
  int64 trip_id = 2;
  float trip_distance = 3;
  double fare_amount = 4;
  Flag store_and_fwd = 5;
}

Sample trip data, is in trips.json:

[
  {
    "value": {
      "vendor_id": 1,
      "trip_id": 1000371,
      "trip_distance": 1.8,
      "fare_amount": 15.32,
      "store_and_fwd": "N"
    }
  },
  ...
]

Create the taxi topic:

just taxi-topic-create

Publish the sample data onto the taxi topic:

just taxi-produce

View the data in DuckDB:

just taxi-duckdb-delta

Giving the following output: