YAML Schema Reference¶

Complete reference for all YAML configuration in streamt projects.

Project Configuration¶

The main project configuration file (stream_project.yml) defines project metadata, runtime connections, defaults, and governance rules.

Project Metadata¶

project:
  name: my-pipeline              # Required: Project identifier
  version: "1.0.0"               # Optional: Semantic version
  description: "Pipeline desc"   # Optional: Human-readable description

Field	Type	Required	Description
`name`	string	Yes	Unique project identifier
`version`	string	No	Semantic version string
`description`	string	No	Project description

Runtime Configuration¶

Kafka¶

runtime:
  kafka:
    bootstrap_servers: kafka:9092           # Required
    security_protocol: SASL_SSL             # Optional
    sasl_mechanism: PLAIN                   # Optional
    sasl_username: ${KAFKA_USER}            # Optional
    sasl_password: ${KAFKA_PASSWORD}        # Optional

Field	Type	Required	Description
`bootstrap_servers`	string	Yes	Kafka broker address(es)
`security_protocol`	string	No	`PLAINTEXT`, `SSL`, `SASL_PLAINTEXT`, `SASL_SSL`
`sasl_mechanism`	string	No	`PLAIN`, `SCRAM-SHA-256`, `SCRAM-SHA-512`
`sasl_username`	string	No	SASL username
`sasl_password`	string	No	SASL password
`bootstrap_servers_internal`	string	No	Internal broker address for cross-network routing

Schema Registry¶

runtime:
  schema_registry:
    url: http://schema-registry:8081  # Required
    username: ${SR_USER}               # Optional
    password: ${SR_PASSWORD}           # Optional

Field	Type	Required	Description
`url`	string	Yes	Schema Registry URL
`username`	string	No	Basic auth username
`password`	string	No	Basic auth password

Flink¶

runtime:
  flink:
    default: local                    # Optional: Default cluster name
    clusters:
      local:
        type: rest                    # Cluster type
        rest_url: http://localhost:8082
        sql_gateway_url: http://localhost:8084
        version: "1.18"               # Optional
        environment: dev              # Optional
        api_key: ${FLINK_API_KEY}     # Optional

Field	Type	Required	Description
`default`	string	No	Name of default cluster
`clusters`	map	No	Named cluster configurations

Cluster Configuration:

Field	Type	Required	Description
`type`	string	No	`rest`, `docker`, `kubernetes`, `confluent`
`rest_url`	string	No	Flink REST API URL
`sql_gateway_url`	string	No	Flink SQL Gateway URL
`version`	string	No	Flink version
`environment`	string	No	Environment identifier
`api_key`	string	No	API key for managed services
`retries`	int	No	Number of retries for failed API calls
`statement_timeout`	int	No	Timeout in ms for SQL statement execution

Connect¶

runtime:
  connect:
    default: local
    clusters:
      local:
        rest_url: http://localhost:8083

Field	Type	Required	Description
`default`	string	No	Default cluster name
`clusters.<name>.rest_url`	string	Yes	Connect REST API URL

Conduktor Gateway (Optional)¶

Required for virtual topic models (auto-inferred from gateway: configuration). See Gateway Guide.

runtime:
  conduktor:
    gateway:
      admin_url: http://localhost:8888       # Required: Gateway Admin API
      proxy_bootstrap: localhost:6969         # Required: Gateway proxy for clients
      username: ${GATEWAY_USER}               # Optional (default: admin)
      password: ${GATEWAY_PASSWORD}           # Optional (default: conduktor)
      virtual_cluster: default                # Optional: For multi-tenant setups

Field	Type	Required	Description
`admin_url`	string	Yes	Gateway Admin API URL
`proxy_bootstrap`	string	Yes	Gateway proxy for Kafka clients
`username`	string	No	Admin API username (default: admin)
`password`	string	No	Admin API password (default: conduktor)
`virtual_cluster`	string	No	Virtual cluster for multi-tenancy

Defaults¶

Set project-wide default values:

defaults:
  topic:
    partitions: 1
    replication_factor: 1

  models:
    cluster: production
    topic:
      partitions: 6
      replication_factor: 3

  tests:
    flink_cluster: production

Section	Field	Type	Description
`topic`	`partitions`	int	Default partition count
`topic`	`replication_factor`	int	Default replication factor
`models`	`cluster`	string	Default Flink cluster
`models.topic`	`partitions`	int	Default for model output topics
`models.topic`	`replication_factor`	int	Default for model output topics
`tests`	`flink_cluster`	string	Default cluster for tests

Connections¶

Define shared credentials/connections once, reference in sinks:

connections:
  snowflake_prod:
    type: snowflake
    config:
      snowflake.url.name: acme.snowflakecomputing.com
      snowflake.user.name: ${SF_USER}
      snowflake.private.key: ${SF_KEY}
      snowflake.database.name: ANALYTICS

Reference in sink models via connection:

models:
  - name: users_sink
    from:
      - ref: enriched_users
    sink:
      connector: snowflake
      connection: snowflake_prod    # merges config from connection
      config:
        snowflake.schema.name: PUBLIC  # sink-specific overrides

Connection config is merged as base, sink config overrides.

Governance Rules¶

Topic Rules¶

rules:
  topics:
    min_partitions: 3
    max_partitions: 128
    min_replication_factor: 2
    required_config:
      - retention.ms
    naming_pattern: "^[a-z]+\\.[a-z]+\\.v[0-9]+$"
    forbidden_prefixes:
      - "_"
      - "test"
    forbidden_suffixes:
      - "_test"
      - "_temp"
    max_replication_factor: 3

Field	Type	Description
`min_partitions`	int	Minimum allowed partitions
`max_partitions`	int	Maximum allowed partitions
`min_replication_factor`	int	Minimum replication factor
`max_replication_factor`	int	Maximum replication factor
`required_config`	list	Required topic config keys
`naming_pattern`	string	Regex pattern for topic names
`forbidden_prefixes`	list	Disallowed topic name prefixes
`forbidden_suffixes`	list	Disallowed topic name suffixes

Model Rules¶

rules:
  models:
    require_description: true
    require_owner: true
    require_tests: true
    max_dependencies: 10

Field	Type	Description
`require_description`	bool	Models must have description
`require_owner`	bool	Models must have owner
`require_tests`	bool	Models must have associated tests
`max_dependencies`	int	Max upstream dependencies

Source Rules¶

rules:
  sources:
    require_schema: true
    require_freshness: true

Field	Type	Description
`require_schema`	bool	Sources must define schema
`require_freshness`	bool	Sources must define freshness SLA

Security Rules¶

rules:
  security:
    require_classification: true
    sensitive_columns_require_masking: true

Field	Type	Description
`require_classification`	bool	Columns must have classification
`sensitive_columns_require_masking`	bool	Sensitive columns require masking policy

Data Residency Rules¶

rules:
  data_residency:
    allowed_regions:
      - EU
      - US

Field	Type	Description
`allowed_regions`	list	Regions that models/sources may declare

Explicit Gateway Rule Removals¶

Gateway removal intent is declared at project level with the prior compiler-level rule fields from generated/manifest.json. Copy only name, virtualTopic, physicalTopic, and interceptors; omit ownership because the compiler reconstructs it from the current project:

apiVersion: streamt.dev/v1alpha1
project:
  name: payments
runtime:
  kafka:
    bootstrap_servers: broker.example:9092
  conduktor:
    gateway:
      admin_url: https://gateway.example:8888
      username: ${GATEWAY_USER}
      password: ${GATEWAY_PASSWORD}
lifecycle:
  gateway_rule_removals:
    - logical_owner: orders_view
      prior_artifact:
        name: orders_rule
        virtualTopic: orders.public
        physicalTopic: raw.orders
        interceptors:
          - type: filter
            config:
              where: "region = 'us'"
          - type: mask
            config:
              field: customer.email
              method: MASK_ALL
              forRoles: [support]

An alias-only rule uses an explicit empty interceptor list. The declaration is a tombstone for one exact prior artifact; removing a model or omitting a rule does not imply deletion.

Field	Type	Required	Description
`lifecycle.gateway_rule_removals`	list	No	Explicit Gateway rule tombstones; defaults to an empty list
`logical_owner`	string	Yes	Prior logical model owner; must be non-empty and contain no `/`
`prior_artifact.name`	string	Yes	Exact prior provider rule name
`prior_artifact.virtualTopic`	string	Yes	Exact prior AliasTopic name
`prior_artifact.physicalTopic`	string	Yes	Exact prior physical input topic
`prior_artifact.interceptors`	list	Yes	Exact ordered prior `filter` and `mask` compiler configurations

Gateway resource names accept letters, digits, ., _, and -. Filter configuration contains only where. Mask configuration contains field, method, and optional forRoles; an omitted forRoles is canonically compiled as forRoles: [], matching an ordinary desired Gateway artifact. All lifecycle objects are strict: unknown fields, null objects or collections, singleton interceptor objects, provider endpoints, backend identities, credentials, and author-supplied ownership are rejected.

The compiler injects managed ownership from the current project and logical_owner, validates the reconstructed value with the compiled Gateway artifact parser, and emits it under the separate artifacts.gateway_rule_removals manifest collection when at least one removal is declared. A non-empty tombstone collection is included in the manifest checksum; omitting the empty collection preserves existing-project checksums. Tombstones never appear in gateway_rules, the desired model list, or the DAG.

Explicit Kafka Connect Connector Removals¶

Connector removal intent is declared at project level. The tombstone names one logical owner and one exact Connector on the configured default Connect cluster; it deliberately contains no connector class, topics, configuration, endpoint, backend identity, or ownership object:

apiVersion: streamt.dev/v1alpha1
project:
  name: payments
  version: 1.0.0
runtime:
  kafka:
    bootstrap_servers: broker.example:9092
  connect:
    default: primary-connect
    clusters:
      primary-connect:
        rest_url: https://connect.example:8443
deployment_state:
  backend: postgres
  namespace: platform
  postgres:
    dsn_env: STREAMT_STATE_ADMIN_DSN
    writer_dsn_env: STREAMT_STATE_WRITER_DSN
    schema: streamt
lifecycle:
  connector_removals:
    - logical_owner: archive_orders
      name: archive-orders-sink
      cluster: primary-connect

Field	Type	Required	Description
`lifecycle.connector_removals`	list	No	Explicit Connector tombstones; defaults to an empty list and accepts at most 256 entries
`logical_owner`	string	Yes	Exact prior logical model owner; 1–128 control-free characters and no `/`
`name`	string	Yes	Exact Connect connector name; 1–256 control-free characters
`cluster`	string	Yes	Explicit Connect cluster alias; 1–128 control-free characters and, in this release, exactly `runtime.connect.default`

All lifecycle objects are strict. Null collections, null fields, empty or whitespace-only values, duplicate logical owners, duplicate (cluster, name) pairs, and unknown fields are rejected. A desired Connector cannot claim the same logical resource or provider locator as a tombstone. In particular, do not add config, connector_class, topics, credentials, an endpoint, a fingerprint, a backend, or ownership to the declaration.

A tombstone is affirmative intent, not proof by itself. It can authorize a delete only through a fresh online reviewed plan, exact PostgreSQL deployment-state schema version 2 writer authority, a matching prior managed ownership record, and a complete live Connector observation whose reconstructed artifact checksum matches that record. Local state, PostgreSQL v1, adopted or legacy records, non-default Connect aliases, and partial or offline workflows fail closed.

Removing a sink model, omitting a Connector artifact, or observing an absent Connector never creates deletion authority. An already-absent tombstone with no prior record is a visible no-op; a missing Connector with a prior managed record is blocking state/provider drift. This lifecycle contract deletes only the named Connector object—not topics, records, offsets, schemas, external systems, credentials, or plugins.

Sources¶

Sources declare external Kafka topics that your project consumes.

sources:
  - name: orders_raw                    # Required: Unique identifier
    topic: orders.raw.v1                # Required: Kafka topic name
    description: "Raw order events"     # Optional
    cluster: production                 # Optional: Kafka cluster
    owner: orders-team                  # Optional
    ownership:                         # Optional; defaults to external
      mode: external                    # external, managed, adopted
    tags:                               # Optional
      - orders
      - raw

    schema:                             # Optional: Schema definition
      registry: confluent               # Schema registry type
      subject: orders-raw-value         # Subject name
      version: latest                   # latest or a positive integer (default: latest)
      format: avro                      # avro, json, protobuf
      definition: |                     # Inline schema (alternative to registry)
        {
          "type": "record",
          "name": "Order",
          "fields": [...]
        }

    columns:                            # Optional: Column metadata
      - name: order_id
        description: "Unique order ID"
        classification: internal

      - name: customer_email
        description: "Customer email"
        classification: sensitive

    freshness:                          # Optional: Freshness SLA
      max_lag_seconds: 300
      warn_after_seconds: 120

Source Fields¶

Field	Type	Required	Description
`name`	string	Yes	Unique source identifier
`topic`	string	Yes	Kafka topic name
`description`	string	No	Human-readable description
`cluster`	string	No	Kafka cluster name
`owner`	string	No	Team or person responsible
`ownership.mode`	string	No	Lifecycle mode; defaults to `external`
`tags`	list	No	Categorization tags
`schema`	object	No	Schema definition
`columns`	list	No	Column metadata
`freshness`	object	No	Freshness SLA configuration
`event_time`	object	No	Event time column and watermark strategy
`region`	string	No	Data residency region (validated against governance rules)

Schema Definition¶

Field	Type	Description
`registry`	string	`confluent` (the current Confluent-compatible Registry client)
`subject`	string	Schema Registry subject name
`version`	integer or `latest`	Subject version; defaults to `latest`
`format`	string	`avro`, `json`, `protobuf`
`definition`	string	Inline schema definition

streamt validate remains fully offline by default. With --check-schemas, external references (a subject/registry without definition or fields) are resolved using read-only requests. The check verifies subject and version existence, the declared format, and every version-pinned reference. Avro and JSON Schema documents are decoded as JSON. Protobuf schema text is retained as-is; streamt currently validates only its Registry metadata/reference graph and does not infer columns or validate Protobuf message semantics.

Column Definition¶

Field	Type	Description
`name`	string	Column name
`type`	string	Flink SQL type (e.g., `STRING`, `BIGINT`, `TIMESTAMP(3)`)
`description`	string	Column description
`classification`	string	`public`, `internal`, `confidential`, `sensitive`, `highly_sensitive`
`proctime`	bool	Mark as processing time attribute (for Flink)

Processing Time Example:

columns:
  - name: order_id
    type: STRING
  - name: proc_time
    type: TIMESTAMP(3)
    proctime: true         # This column will use PROCTIME() in Flink SQL

This generates:

CREATE TABLE ... (
  `order_id` STRING,
  `proc_time` AS PROCTIME()  -- Processing time attribute
)

Freshness Configuration¶

Field	Type	Description
`max_lag_seconds`	int	Maximum acceptable lag (error)
`warn_after_seconds`	int	Lag threshold for warning

Event Time Configuration¶

Configure event time semantics for streaming processing. This enables proper watermark generation for windowed aggregations.

Simple case (just specify the column):

sources:
  - name: events
    topic: events.raw.v1
    event_time:
      column: event_timestamp           # Simple: just the column name

Advanced case (custom watermark settings):

sources:
  - name: events
    topic: events.raw.v1
    columns:
      - name: event_id
      - name: user_id
      - name: event_timestamp  # Will be TIMESTAMP(3) in Flink

    # Event time with watermark configuration
    event_time:
      column: event_timestamp
      watermark:
        strategy: bounded_out_of_orderness
        max_out_of_orderness_ms: 5000   # Allow 5 seconds of late data
      allowed_lateness_ms: 60000        # Accept data up to 1 minute late

This generates proper Flink SQL watermark declarations:

CREATE TABLE events (
  ...
  `event_timestamp` TIMESTAMP(3),
  WATERMARK FOR `event_timestamp` AS `event_timestamp` - INTERVAL '5' SECOND
) WITH (...)

Event Time Fields:

Field	Location	Type	Description	Required
`column`	Top-level	string	Column name containing event time	Yes
`watermark.strategy`	Advanced	string	`bounded_out_of_orderness`, `monotonously_increasing`, or `custom`	No (default: bounded)
`watermark.max_out_of_orderness_ms`	Advanced	int	Max delay in ms (bounded strategy)	No (default: 5000)
`watermark.expression`	Advanced	string	Custom SQL expression (custom strategy)	No
`allowed_lateness_ms`	Advanced	int	Accept late events within window	No

Watermark Strategies:

Strategy	Use Case
`bounded_out_of_orderness`	Events can arrive out of order (most common)
`monotonously_increasing`	Events always arrive in order (rare)
`custom`	Arbitrary SQL expression for watermark generation

Custom watermark example:

event_time:
  column: event_ts
  watermark:
    strategy: custom
    expression: "CASE WHEN `event_ts` > CURRENT_TIMESTAMP THEN CURRENT_TIMESTAMP ELSE `event_ts` END"

See Streaming Fundamentals for more on watermarks.

Models¶

Models define transformations that create new data streams.

Auto-Inferred Materialization¶

IMPORTANT: The materialized field is now auto-inferred from your SQL and configuration. You no longer need to specify it explicitly.

Inference Rules:

SQL Pattern	Inferred Materialization
Contains `TUMBLE`, `HOP`, `SESSION`, or `CUMULATE`	`flink`
Contains `GROUP BY` without window functions	`flink`
Contains `JOIN`	`flink`
Has `gateway:` configuration	`virtual_topic`
Has `from:` without `sql:`	`sink`
Simple `SELECT`, `WHERE`, projection	`topic`

Basic Example¶

models:
  - name: orders_validated
    description: "Validated orders"
    owner: orders-team
    tags: [orders, validated]

    sql: |
      SELECT
        order_id,
        customer_id,
        amount,
        CASE WHEN amount > 1000 THEN 'high' ELSE 'normal' END as tier
      FROM {{ source("orders_raw") }}
      WHERE order_id IS NOT NULL
        AND amount > 0

    # materialized: topic (auto-inferred from simple SELECT)

Example With Overrides¶

models:
  - name: hourly_revenue
    description: "Hourly revenue aggregation"
    owner: analytics-team
    ownership:
      mode: managed                    # Default for model output resources

    sql: |
      SELECT
        TUMBLE_START(order_time, INTERVAL '1' HOUR) as window_start,
        SUM(amount) as revenue
      FROM {{ ref("orders_validated") }}
      GROUP BY TUMBLE(order_time, INTERVAL '1' HOUR)

    # materialized: flink (auto-inferred from TUMBLE)

    # Only specify when overriding defaults:
    flink:
      parallelism: 8
      checkpoint_interval_ms: 60000
      state_backend: rocksdb
      state_ttl_ms: 86400000

    flink_cluster: production

    topic:
      name: analytics.revenue.v1
      partitions: 12
      replication_factor: 3
      config:
        retention.ms: 604800000
        cleanup.policy: delete

    columns:
      - name: window_start
        description: "Start of the hourly window"
      - name: revenue
        description: "Total revenue in the window"
        classification: confidential

Full Example (All Options)¶

models:
  - name: orders_validated
    description: "Validated orders"     # Optional but recommended
    owner: orders-team                  # Optional
    tags: [orders, validated]           # Optional
    access: protected                   # Optional: private, protected, public
    group: orders                       # Optional: Logical grouping
    version: 1                          # Optional: Model version
    key: order_id                       # Optional: Partition key

    sql: |                              # Required (except for sink models)
      SELECT
        order_id,
        customer_id,
        amount,
        CASE WHEN amount > 1000 THEN 'high' ELSE 'normal' END as tier
      FROM {{ source("orders_raw") }}
      WHERE order_id IS NOT NULL
        AND amount > 0

    # Infrastructure configuration (optional - only when overriding defaults)

    # Flink job settings (for flink materialization)
    flink:
      parallelism: 4
      checkpoint_interval_ms: 60000
      state_backend: rocksdb
      state_ttl_ms: 86400000

    # Target Flink cluster
    flink_cluster: production

    # Output topic configuration
    topic:
      name: orders.validated.v1
      partitions: 12
      replication_factor: 3
      config:
        retention.ms: 604800000
        cleanup.policy: delete

    # Connect cluster (for sink materialization)
    connect_cluster: production

    # Column metadata (stays at top level)
    columns:
      - name: customer_id
        description: "Customer identifier"
        classification: internal
      - name: amount
        description: "Order amount"
        classification: confidential

    security:                           # Optional
      classification:
        customer_id: internal
        amount: confidential
      policies:
        - mask:
            column: customer_email
            method: hash
            for_roles: [analyst]

Model Fields (Top Level)¶

Top-level fields (always visible, business-focused):

Field	Type	Required	Description
`name`	string	Yes	Unique model identifier
`description`	string	No	Human-readable description
`owner`	string	No	Team or person responsible
`ownership.mode`	string	No	`external`, `managed`, or `adopted`; defaults to `managed`
`tags`	list	No	Categorization tags
`access`	string	No	`private`, `protected`, `public`
`group`	string	No	Logical grouping
`version`	int	No	Model version (duplicate/gap detection in validator)
`region`	string	No	Data residency region
`key`	string	No	Partition key column
`sql`	string	Conditional	SQL transformation (required except for sink)
`from`	string	Conditional	Source model (required for sink)
`columns`	list	No	Column metadata and documentation
`primary_key`	list	No	Primary key columns (composite key support)
`contract`	object	No	Schema contract enforcement (see ModelContract below)
`macro`	string	No	SQL macro template name for parameterized models
`params`	dict	No	Parameters for macro-based models
`security`	object	No	Security policies and classification
`ml_outputs`	object	No	ML model output schemas for `ML_PREDICT` (see below)

Note: The materialized field is no longer specified - it's auto-inferred from your SQL and configuration.

Lifecycle ownership is distinct from the human owner. external artifacts are observe-only. managed artifacts may be created, but live resources without matching prior state still require adoption. Declaring adopted is not an adoption operation; the planner requires matching persisted state before any mutation.

ML Output Schemas¶

When using ML_PREDICT in your SQL, declare the expected output schema so streamt can properly type-check the results:

models:
  - name: fraud_scored_orders
    description: "Orders with fraud predictions"

    sql: |
      SELECT
        order_id,
        amount,
        ML_PREDICT('fraud_detector', order_id, amount, customer_id) as prediction
      FROM {{ ref("orders_validated") }}

    # Declare ML model output schemas for type inference
    ml_outputs:
      fraud_detector:
        description: "Fraud detection ML model"
        columns:
          - name: is_fraud
            type: BOOLEAN
          - name: confidence
            type: DOUBLE
          - name: risk_score
            type: DOUBLE

Field	Type	Description
`ml_outputs.<model_name>.columns`	list	Output columns from the ML model
`ml_outputs.<model_name>.description`	string	Description of the ML model

Confluent Cloud Only

ML_PREDICT and ML_EVALUATE are only available on Confluent Cloud Flink clusters.

Model Contract¶

Schema contract enforcement for models. When a contract is defined, the model's output columns must match the declared schema.

models:
  - name: orders_validated
    contract:
      enforced: true
      columns:
        - name: order_id
          type: STRING
          nullable: false
        - name: amount
          type: DOUBLE
          nullable: false

Field	Type	Default	Description
`enforced`	bool	`true`	Whether the contract is actively enforced
`columns`	list	—	Expected column definitions

Contract Column fields:

Field	Type	Required	Description
`name`	string	Yes	Column name
`type`	string	No	Expected Flink SQL type
`nullable`	bool	No	Whether the column allows nulls
`description`	string	No	Column description

Model Fields (Infrastructure)¶

Infrastructure/tuning fields (optional, top-level on models):

Field	Type	Description
`flink`	object	Flink job configuration (see below)
`flink_cluster`	string	Target Flink cluster name
`topic`	object	Output topic configuration (see below)
`connect_cluster`	string	Target Kafka Connect cluster
`event_time`	object	Event time watermark configuration
`gateway`	object	Gateway-specific settings

Topic Configuration¶

Top-level topic: field on models:

Field	Type	Description
`name`	string	Explicit topic name (default: model name)
`partitions`	int	Partition count
`replication_factor`	int	Replication factor
`config`	map	Kafka topic configuration

Gateway Configuration¶

Top-level gateway: field on models (for virtual topic models):

Field	Type	Description
`virtual_topic.name`	string	Authoritative custom virtual topic name
`virtual_topic.compression`	string	Compression type (`gzip`, `snappy`, `lz4`, `zstd`)

models:
  - name: filtered_orders
    description: "Orders filtered by region"

    sql: |
      SELECT * FROM {{ source("orders") }}
      WHERE region = 'US'

    # Gateway virtual topic with compression
    gateway:
      virtual_topic:
        name: orders.us.filtered
        compression: lz4

For virtual-topic models, gateway.virtual_topic.name determines the compiled Gateway alias. The older topic.name spelling remains supported as a fallback when gateway.virtual_topic.name is omitted. If both fields are present, they must contain the same name.

Virtual Topics

Virtual topics are automatically inferred when Gateway is configured. They provide schema transformation and data masking without creating physical Kafka topics. See the Gateway Guide.

Flink Configuration¶

Top-level flink: field on models:

Field	Type	Description
`parallelism`	int	Job parallelism
`checkpoint_interval_ms`	int	Checkpoint interval in ms
`state_backend`	string	`hashmap` or `rocksdb`
`state_ttl_ms`	int	State TTL in ms
`checkpoint`	object	Advanced checkpoint config (timeout, mode, etc.)
`restart_strategy`	object	Restart strategy (fixed-delay, failure-rate, exponential-delay)
`changelog_mode`	string	`append` (default) or `upsert` (switches to upsert-kafka)
`rocksdb`	object	RocksDB tuning (block_cache_size_mb, write_buffer_size_mb, predefined_options)
`resources`	object	TM/JM resources (taskmanager_memory_mb, taskmanager_slots, jobmanager_memory_mb)

See Flink Options Reference for complete Flink configuration.

Security Policies¶

Field	Type	Description
`classification`	map	Column name → classification level
`policies`	list	Masking and access policies

Masking Methods¶

Method	Description
`hash`	SHA-256 hash of value
`redact`	Replace with `***`
`partial`	Show partial value (e.g., `***-1234`)
`tokenize`	Replace with reversible token
`null`	Replace with null

Sink Models¶

Sink models export data to external systems via Kafka Connect.

models:
  - name: orders_snowflake
    description: "Export orders to Snowflake"

    # materialized: sink (auto-inferred from having 'from' without 'sql')
    from: orders_validated              # Source model or topic

    sink:
      connector: snowflake-sink
      config:
        snowflake.url.name: ${SNOWFLAKE_URL}
        snowflake.user.name: ${SNOWFLAKE_USER}
        snowflake.private.key: ${SNOWFLAKE_KEY}
        snowflake.database.name: ANALYTICS
        snowflake.schema.name: RAW
        snowflake.table.name: ORDERS
        tasks.max: 4

    # Only when overriding defaults:
    connect_cluster: production

Sink Configuration¶

Field	Type	Required	Description
`connector`	string	Yes	Connector type/class
`connection`	string	No	Reference to global connection (see Connections)
`config`	map	Yes	Connector configuration (overrides connection config)

Common Connector Types¶

Type	Use Case
`snowflake-sink`	Snowflake data warehouse
`bigquery-sink`	Google BigQuery
`s3-sink`	Amazon S3
`gcs-sink`	Google Cloud Storage
`jdbc-sink`	PostgreSQL, MySQL, etc.
`elasticsearch-sink`	Elasticsearch/OpenSearch
`http-sink`	HTTP/REST APIs

Tests¶

Tests validate data quality in your streaming pipelines.

tests:
  - name: orders_schema_test            # Required
    model: orders_validated             # Required: Model to test
    type: schema                        # Required: schema, sample, continuous
    flink_cluster: production           # Optional

    assertions:                         # Required
      - not_null:
          columns: [order_id, customer_id, amount]

      - accepted_values:
          column: status
          values: [pending, confirmed, shipped, delivered]

      - accepted_types:
          types:
            order_id: STRING
            amount: DOUBLE
            created_at: TIMESTAMP

      - range:
          column: amount
          min: 0
          max: 1000000

      - unique_key:
          key: order_id
          window: "1 HOUR"
          tolerance: 0.01

    on_failure:                         # Optional
      severity: error                   # error, warning
      actions:
        - dlq:
            topic: test_failures

Test Fields¶

Field	Type	Required	Description
`name`	string	Yes	Unique test identifier
`model`	string	Yes	Model to test
`type`	string	Yes	`schema`, `sample`, `continuous`
`assertions`	list	Yes	Test assertions
`sample_size`	int	No	Rows to sample (for sample tests)
`flink_cluster`	string	No	Flink cluster for execution
`on_failure`	object	No	Failure handling

Test Types¶

Type	Description
`schema`	Validate schema structure and types
`sample`	Test assertions on sampled data
`continuous`	Ongoing streaming validation

Assertion Types¶

Assertion	Description	Fields
`not_null`	Columns must not be null	`columns`
`accepted_values`	Column values must be in set	`column`, `values`
`accepted_types`	Column types must match	`types` (map)
`range`	Numeric column within range	`column`, `min`, `max`
`unique_key`	Key uniqueness within window	`key`, `window`, `tolerance`
`max_lag`	Max lag from event time	`column`, `max_seconds`
`throughput`	Messages per second bounds	`min_per_second`, `max_per_second`
`distribution`	Value distribution buckets	`column`, `buckets`
`foreign_key`	Referential integrity	`column`, `ref_model`, `ref_key`
`custom_sql`	Custom SQL predicate	`name`, `where`, `detail_column`

Exposures¶

Exposures document downstream consumers of your data.

exposures:
  - name: checkout-service              # Required
    type: application                   # Required
    role: consumer                      # Optional: producer, consumer, both
    description: "Checkout microservice"
    owner: checkout-team
    url: https://github.com/org/checkout
    repo: org/checkout
    language: java

    consumes:                           # Topics/models consumed
      - ref: orders_validated
      - source: inventory_updates

    produces:                           # Topics/models produced
      - ref: checkout_events

    consumer_group: checkout-service-cg

    sla:
      availability: "99.9%"
      max_lag_messages: 1000
      max_lag_minutes: 5
      max_end_to_end_latency_ms: 500

    contracts:
      schema: orders-validated-value
      compatibility: BACKWARD

    access:
      roles: [checkout-team, platform-team]
      purpose: "Process customer checkouts"

Exposure Types¶

Type	Description
`application`	Microservice or application
`dashboard`	BI dashboard or reporting
`ml_training`	ML training pipeline
`ml_inference`	ML inference service
`api`	External API

Exposure Fields¶

Field	Type	Required	Description
`name`	string	Yes	Unique exposure identifier
`type`	string	Yes	Exposure type
`role`	string	No	`producer`, `consumer`, `both`
`description`	string	No	Description
`owner`	string	No	Owner team/person
`url`	string	No	Documentation URL
`repo`	string	No	Repository
`language`	string	No	Programming language
`tool`	string	No	BI tool (for dashboards)
`consumes`	list	No	Consumed sources/refs
`produces`	list	No	Produced sources/refs
`consumer_group`	string	No	Kafka consumer group
`sla`	object	No	SLA configuration
`contracts`	object	No	Schema contracts
`access`	object	No	Access configuration
`schedule`	string	No	Cron schedule (for batch)
`owners`	list	No	Additional owners with role metadata
`depends_on`	list	No	Explicit dependency references
`data_requirements`	dict	No	Data quality requirements

SLA Configuration¶

Field	Type	Description
`availability`	string	Uptime target (e.g., "99.9%")
`max_lag_messages`	int	Max message lag
`max_lag_minutes`	int	Max time lag
`max_produce_latency_ms`	int	Max producer latency
`max_end_to_end_latency_ms`	int	Max E2E latency
`max_error_rate`	float	Max error rate (0.0-1.0, validated)
`freshness_minutes`	int	Max data freshness in minutes

SQL Macros¶

Use Jinja-style macros in SQL to reference sources and models.

source()¶

Reference a source topic:

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

ref()¶

Reference another model:

SELECT * FROM {{ ref("orders_validated") }}

Example with Multiple References¶

models:
  - name: order_enriched
    description: "Orders enriched with customer data"

    # materialized: flink (auto-inferred from JOIN)
    sql: |
      SELECT
        o.order_id,
        o.amount,
        c.name as customer_name,
        c.tier as customer_tier
      FROM {{ ref("orders_validated") }} o
      JOIN {{ source("customers") }} c
        ON o.customer_id = c.id
        AND o.order_time BETWEEN c.update_time - INTERVAL '1' HOUR
                             AND c.update_time + INTERVAL '1' HOUR

    # Only when overriding defaults:
    flink:
      parallelism: 4

Environment Variables¶

Use ${VAR_NAME} syntax to reference environment variables:

runtime:
  kafka:
    bootstrap_servers: ${KAFKA_BOOTSTRAP_SERVERS}
    sasl_password: ${KAFKA_PASSWORD}

  schema_registry:
    url: https://sr.example.com:8081

Variables can be set via:

System environment: export KAFKA_PASSWORD=secret
.env file in project root
CI/CD secrets

.env

KAFKA_BOOTSTRAP_SERVERS=localhost:9092
KAFKA_PASSWORD=secret
SCHEMA_REGISTRY_URL=http://localhost:8081

File Organization¶

Single File (Small Projects)¶

stream_project.yml

project:
  name: simple-pipeline

runtime:
  kafka:
    bootstrap_servers: localhost:9092

sources:
  - name: events
    topic: events.raw

models:
  - name: events_clean
    description: "Cleaned event stream"
    # materialized: topic (auto-inferred from simple SELECT)
    sql: |
      SELECT * FROM {{ source("events") }}
      WHERE event_id IS NOT NULL

Multi-File (Large Projects)¶

my-pipeline/
├── stream_project.yml          # Project config + runtime
├── sources/
│   ├── orders.yml
│   └── customers.yml
├── models/
│   ├── staging/
│   │   ├── stg_orders.yml
│   │   └── stg_customers.yml
│   └── marts/
│       ├── order_metrics.yml
│       └── customer_360.yml
├── tests/
│   ├── orders_tests.yml
│   └── customers_tests.yml
└── exposures/
    └── applications.yml

Each YAML file can contain its respective type:

sources/orders.yml

sources:
  - name: orders_raw
    topic: orders.raw.v1
    # ...

models/staging/stg_orders.yml

models:
  - name: stg_orders
    description: "Staging orders with basic validation"
    # materialized: topic (auto-inferred)
    sql: |
      SELECT * FROM {{ source("orders_raw") }}
      WHERE order_id IS NOT NULL