Skip to content

streamt

dbt for streaming: develop, test, and deploy streaming applications as code

Preview

Get Started View on GitHub

What is streamt?

streamt puts transformation logic, resource configuration, tests, and dependencies in a versioned project. Write SQL models and use the same workflow to validate, review, and deploy supported changes to Kafka, Flink, and Connect.

Import existing resources as external declarations, or add resources whose lifecycle streamt manages. Describe custom applications and their inputs and outputs alongside the SQL models. Importing or documenting an existing system does not transfer its ownership to streamt.

The product direction prioritizes creating and updating applications, including an execution path for users without Flink. The Kafka Streams starting path creates a bounded projection/filter topology through a fixed local Docker runner. One reviewed Kafka Streams predicate update can preserve its application identity and offsets, with explicit resume after interruption. Broader runner changes, existing Flink updates, and custom-application deployment remain unsupported. External declarations are not live-diffed by ordinary planning; status --include-external explicitly observes them. Managed deployment safety checks still access their providers. See the support matrix for current boundaries.

# streamt:skip
project:
  name: payments-pipeline
  version: "1.0.0"

sources:
  - name: payments_raw
    topic: payments.raw.v1
    description: Raw payment events from checkout service

models:
  - name: payments_validated
    description: Validated payments with fraud scores
    sql: |
      SELECT
        payment_id,
        customer_id,
        amount,
        CASE WHEN amount > 10000 THEN 'HIGH_RISK' ELSE 'NORMAL' END as risk_level
      FROM {{ source("payments_raw") }}
      WHERE status IS NOT NULL

Why streamt?

Declarative First

Define what you want, not how to build it. Write YAML and SQL, let streamt generate Kafka topics, Flink jobs, and Connect configurations.

Built-in Lineage

Track dependencies from SQL references and declared application inputs and outputs. Undeclared application internals remain unknown.

Governance & Quality

Enforce naming conventions, partition requirements, and data classification. Run schema, sample, and continuous tests on your streams.

Plan Before Apply

Review resource changes before deployment. See new topics and jobs, supported connector changes, and blockers for unsafe or unsupported updates.

How It Works

graph LR
    A[YAML + SQL] --> B[Parse & Validate]
    B --> C[Compile Artifacts]
    C --> D[Plan Changes]
    D --> E[Apply to Infrastructure]

    E --> F[Kafka Topics]
    E --> G[Flink Jobs]
    E --> H[Connect Connectors]

    style A fill:#e0e7ff,stroke:#5046e5
    style E fill:#d1fae5,stroke:#059669
  1. Define your sources, models, tests, and exposures in YAML
  2. Validate syntax, references, and governance rules
  3. Compile to infrastructure-specific artifacts
  4. Plan to see what will change
  5. Apply to deploy to Kafka, Flink, and Connect

Quick Example

sources/payments.yml
sources:
  - name: payments_raw
    topic: payments.raw.v1
    description: Raw payment events from checkout
    owner: payments-team
    schema:
      registry: confluent
      subject: payments-raw-value
    columns:
      - name: payment_id
        description: Unique payment identifier
      - name: amount
        description: Payment amount in cents
        classification: internal
models/payments_clean.yml
models:
  - name: payments_clean
    description: Cleaned and validated payments
    sql: |
      SELECT
        payment_id,
        customer_id,
        amount,
        currency,
        created_at
      FROM {{ source("payments_raw") }}
      WHERE payment_id IS NOT NULL
        AND amount > 0
    topic:
      name: payments.clean.v1
      partitions: 12
      config:
        retention.ms: 604800000
tests/payments_tests.yml
tests:
  - name: payments_schema_test
    model: payments_clean
    type: schema
    assertions:
      - not_null:
          columns: [payment_id, customer_id, amount]
      - accepted_values:
          column: currency
          values: [USD, EUR, GBP]

  - name: payments_sample_test
    model: payments_clean
    type: sample
    sample_size: 1000
    assertions:
      - range:
          column: amount
          min: 0
          max: 1000000

# Validate your project
streamt validate

# See what will be deployed
streamt plan

# Deploy to infrastructure
streamt apply

# Run tests
streamt test

# View lineage
streamt lineage

Core Concepts

Concept Description
Source External data entry point (Kafka topic produced by another system)
Model Transformation that creates new data streams
Test Quality assertion (schema, sample, or continuous)
Exposure Documentation of downstream consumers
Materialization How a model is deployed (topic, virtual_topic, flink, sink)

Learn more about concepts →

Materializations

Materializations are automatically inferred from your SQL:

Type Auto-detected When Infrastructure
virtual_topic Stateless SQL + Gateway configured Conduktor Gateway
flink Stateful SQL (GROUP BY, JOIN, windows) Flink SQL job
sink from: without SQL Kafka Connect

Confluent Cloud Flink Statements and its ML functions are not supported deployment paths. Check the support matrix before choosing a runtime.

Learn more about materializations →

Installation

Until the first PyPI alpha is published, install the preview from the repository:

python -m pip install "git+https://github.com/conduktor/streamt.git@main"

The package currently includes the dependencies for every supported runtime integration. Pin main to an immutable commit SHA for reproducible installs.

Full installation guide →

Community

**Ready to build streaming pipelines the declarative way?** [Get Started](getting-started/quickstart.md){ .md-button .md-button--primary }