Models

Models are the heart of streamt. They define transformations that produce new data streams from sources and other models.

What is a Model?

A model is a named transformation that:

• Takes input from sources or other models
• Applies SQL transformations
• Produces output to a new stream
• Is deployed as infrastructure (topic, Flink job, or connector)

Basic Model

models/orders_clean.yml

models:
  - name: orders_clean
    description: Cleaned and validated orders
    ownership:
      mode: managed  # Default for model output resources
    sql: |
      SELECT order_id, customer_id, amount
      FROM {{ source("orders_raw") }}
      WHERE amount > 0

The materialized property is automatically inferred from your SQL. Simple transforms become topics; windowed aggregations become Flink jobs.

Lifecycle Ownership

Model output topics, Flink jobs, connectors, and Gateway rules default to ownership.mode: managed. Use external for observe-only outputs. Use adopted only to describe a resource already recorded in persisted ownership state; declaring the mode does not adopt a live resource or authorize mutation. The human/team owner field is independent metadata.

Materializations

Materializations are automatically inferred from your SQL:

• Stateless SQL (WHERE, projections) → virtual_topic (if Gateway configured) or flink (fallback)
• Stateful SQL (GROUP BY, JOIN, windows) → flink materialization
• ML_PREDICT, ML_EVALUATE → flink (Confluent Cloud only)
• from: only (no SQL) → sink materialization

You can override auto-inference by explicitly specifying materialized:.

Topic Materialization

Creates a Kafka topic with stateless transformations:

- name: high_value_orders
  sql: |
    SELECT *
    FROM {{ source("orders") }}
    WHERE amount > 10000

  # Optional configuration
  topic:
    name: orders.high-value.v1
    partitions: 12
    config:
      retention.ms: 604800000

Best for: Filtering, field selection, simple transformations

Virtual Topic Materialization

Creates a Gateway rule for read-time filtering (requires Conduktor):

- name: orders_europe
  materialized: virtual_topic  # Must be explicit
  from: orders_clean
  sql: |
    SELECT *
    FROM {{ ref("orders_clean") }}
    WHERE region = 'EU'

Best for: Multi-tenant views, access control, no additional storage

Flink Materialization

Deploys a Flink SQL job for stateful processing:

- name: hourly_revenue
  sql: |
    SELECT
      TUMBLE_START(event_time, INTERVAL '1' HOUR) as window_start,
      SUM(amount) as total_revenue,
      COUNT(*) as order_count
    FROM {{ ref("orders_clean") }}
    GROUP BY TUMBLE(event_time, INTERVAL '1' HOUR)

  # Optional configuration
  flink:
    parallelism: 8
    checkpoint_interval_ms: 60000
  topic:
    name: analytics.revenue.v1
    partitions: 6

The TUMBLE window automatically triggers Flink materialization.

Best for: Windowed aggregations, joins, complex state

Sink Materialization

Creates a Kafka Connect connector to export data:

- name: orders_snowflake
  from: orders_clean  # No SQL = sink materialization
  connector:
    type: snowflake-sink
    config:
      snowflake.url.name: ${SNOWFLAKE_URL}
      snowflake.database.name: ANALYTICS
      snowflake.schema.name: ORDERS

Using from: without sql: automatically triggers sink materialization.

Best for: Exporting to data warehouses, databases, S3

ML Inference (Confluent Flink)

Use ML_PREDICT and ML_EVALUATE for real-time ML inference:

- name: fraud_scores
  sql: |
    SELECT
      transaction_id,
      amount,
      ML_PREDICT('FraudModel', amount, merchant_id) as prediction
    FROM {{ ref("transactions") }}

  # Declare ML output schema for proper type inference
  ml_outputs:
    FraudModel:
      columns:
        - name: fraud_score
          type: DOUBLE
        - name: confidence
          type: DOUBLE

Requirements: - Confluent Cloud Flink cluster - Model registered in Confluent's model registry

Best for: Real-time fraud detection, recommendations, anomaly detection

Complete Model Reference

models:
  - name: customer_metrics
    description: |
      Real-time customer metrics aggregated by hour.
      Used for the customer dashboard and recommendations.

    # Ownership and organization
    owner: analytics-team
    ownership:
      mode: managed
    tags: [analytics, customer, kpi]
    access: protected  # public, protected, private

    # Partitioning key
    key: customer_id

    # SQL transformation
    sql: |
      SELECT
        customer_id,
        email,
        phone,
        TUMBLE_START(event_time, INTERVAL '1' HOUR) as window_start,
        COUNT(*) as event_count,
        SUM(CASE WHEN event_type = 'purchase' THEN amount ELSE 0 END) as revenue
      FROM {{ source("customer_events") }}
      GROUP BY
        customer_id,
        email,
        phone,
        TUMBLE(event_time, INTERVAL '1' HOUR)

    # Infrastructure configuration (optional)

    # Output topic configuration
    topic:
      name: analytics.customer-metrics.v1
      partitions: 12
      replication_factor: 3
      config:
        retention.ms: 2592000000  # 30 days
        cleanup.policy: delete

    # Flink job configuration
    flink:
      parallelism: 8
      checkpoint_interval_ms: 60000
      state_backend: rocksdb
    flink_cluster: production

    # Security policies
    security:
      policies:
        - mask:
            column: email
            method: hash
        - mask:
            column: phone
            method: partial

SQL Syntax

Referencing Sources

Use {{ source("name") }} to reference a source:

SELECT * FROM {{ source("orders_raw") }}

Referencing Other Models

Use {{ ref("name") }} to reference another model:

SELECT *
FROM {{ ref("orders_clean") }} o
JOIN {{ ref("customers") }} c ON o.customer_id = c.id

Jinja Templating

Use Jinja for dynamic SQL:

SELECT
  order_id,
  amount,
  {% if var("include_pii", false) %}
  customer_email,
  {% endif %}
  created_at
FROM {{ source("orders") }}
WHERE created_at > '{{ var("start_date") }}'

Reserved Keywords

When using SQL reserved words as column names, escape them:

SELECT
  `timestamp`,  -- Reserved keyword
  `value`,      -- Reserved keyword
  user_id
FROM {{ source("events") }}

Topic Configuration

Configure the output Kafka topic using the topic: field:

topic:
  name: orders.clean.v1        # Topic name
  partitions: 12               # Number of partitions
  replication_factor: 3        # Replication factor
  config:                      # Topic configs
    retention.ms: 604800000    # 7 days
    cleanup.policy: delete     # or compact
    min.insync.replicas: 2

Flink Configuration

Configure Flink job settings using the flink: field:

flink:
  parallelism: 8                  # Job parallelism
  checkpoint_interval_ms: 60000    # Checkpoint interval (ms)
  checkpoint:
    timeout_ms: 300000            # Checkpoint timeout (ms)
  state_backend: rocksdb          # hashmap or rocksdb
  restart_strategy:               # Restart strategy
    type: fixed-delay

Set flink_cluster: production on the model to select a configured target.

Connector Configuration

Configure Kafka Connect sinks using the sink: field:

sink:
  connector: snowflake-sink      # Connector type
  config:
    tasks.max: 4
    # Connector-specific configuration
    snowflake.url.name: ${SNOWFLAKE_URL}
    snowflake.user.name: ${SNOWFLAKE_USER}
    snowflake.database.name: ANALYTICS

Supported connector types:

Type	Target
snowflake-sink	Snowflake
bigquery-sink	BigQuery
s3-sink	Amazon S3
jdbc-sink	PostgreSQL, MySQL, etc.
elasticsearch-sink	Elasticsearch

Access Control

Control who can read your model's output:

access: protected  # or public, private

# With group-based access
access:
  level: protected
  allowed_groups:
    - analytics-team
    - data-science

Level	Description
public	Anyone can consume
protected	Only specified groups
private	Only within this project

Security & Masking

Apply masking policies to sensitive columns:

security:
  policies:
    - mask:
        column: email
        method: hash      # MD5 hash

    - mask:
        column: phone
        method: partial   # Show last 4 digits

    - mask:
        column: ssn
        method: redact    # Replace with ***

    - mask:
        column: credit_card
        method: tokenize  # Replace with token

Dependencies

Dependencies are automatically extracted from your SQL:

- name: enriched_orders
  sql: |
    SELECT o.*, c.name as customer_name
    FROM {{ ref("orders_clean") }} o        -- Depends on orders_clean
    JOIN {{ source("customers") }} c         -- Depends on customers source
    ON o.customer_id = c.id

This model depends on both orders_clean and the customers source. streamt builds a DAG from these dependencies.

Best Practices

1. One Transformation Per Model

# Good: Single responsibility
- name: orders_validated
  sql: SELECT * FROM {{ source("orders") }} WHERE valid = true

- name: orders_enriched
  sql: |
    SELECT o.*, c.tier
    FROM {{ ref("orders_validated") }} o
    JOIN {{ source("customers") }} c ON o.customer_id = c.id

# Bad: Too much in one model
- name: orders_all_in_one
  sql: |
    WITH validated AS (...),
         enriched AS (...),
         aggregated AS (...)
    SELECT * FROM aggregated

2. Use Meaningful Names

# Good
- name: daily_revenue_by_region
- name: high_risk_transactions
- name: customer_lifetime_value

# Bad
- name: orders_v2
- name: temp_table
- name: test_model

3. Document Everything

- name: fraud_scores
  description: |
    Real-time fraud risk scores for transactions.
    Score ranges from 0 (safe) to 100 (high risk).
    Updated every 5 seconds based on recent patterns.
  owner: fraud-team
  tags: [fraud, ml, real-time]
  sql: 'SELECT * FROM {{ source("transactions") }}'

4. Set Appropriate Partitions

# High-volume: more partitions for parallelism
topic:
  partitions: 24

# Low-volume: fewer partitions
topic:
  partitions: 3

5. Use Keys for Ordering

# Ensures customer events stay ordered
- name: customer_events_clean
  key: customer_id  # Partition by customer

Example: Multi-stage Pipeline

models/pipeline.yml

models:
  # Stage 1: Clean raw data
  - name: events_clean
    sql: |
      SELECT event_id, user_id, event_type, created_at
      FROM {{ source("events_raw") }}
      WHERE event_id IS NOT NULL

  # Stage 2: Enrich with user data
  - name: events_enriched
    sql: |
      SELECT e.*, u.tier, u.country
      FROM {{ ref("events_clean") }} e
      JOIN {{ source("users") }} u ON e.user_id = u.id

  # Stage 3: Aggregate metrics (auto-inferred as Flink due to TUMBLE)
  - name: hourly_metrics
    sql: |
      SELECT
        country,
        TUMBLE_START(created_at, INTERVAL '1' HOUR) as hour,
        COUNT(*) as event_count
      FROM {{ ref("events_enriched") }}
      GROUP BY country, TUMBLE(created_at, INTERVAL '1' HOUR)

  # Stage 4: Export to warehouse (auto-inferred as sink)
  - name: metrics_warehouse
    from: hourly_metrics
    connector:
      type: snowflake-sink

Next Steps

• Materializations Reference — Detailed materialization docs
• Tests — Test your models
• DAG & Lineage — Understand dependencies