DATTAX Streaming — Syntax

Following data in real time used to require a separate engineering project: Kafka consumers, offset management, windows, state. In DATTAX, streaming is the same language as your batch scripts — you swap FROM for FROM STREAM, add a window and a watermark, and the pipeline starts emitting live results. All primitives are additive: existing batch scripts keep working unchanged.

Source: FROM STREAM ...

FROM STREAM <KIND> "<connectionName>" [options]

Supported kinds:

Options (all optional, in the order they appear)

TOPIC "nome"                 -- obrigatorio para KAFKA/PULSAR/OPENSEARCH_POLL
TABLE "schema.tabela"        -- obrigatorio para CDC
LABEL "Label"                -- obrigatorio para NEO4J_CDC fallback
FORMAT JSON | AVRO | PROTOBUF
STARTING FROM EARLIEST | LATEST | TIMESTAMP "2026-01-01T00:00:00Z"
GROUP ID "meu-consumer"      -- default: datta-dattax-<subscriptionId>

Examples:

EVALUATE FROM STREAM KAFKA "my-kafka" TOPIC "orders" FORMAT JSON STARTING FROM LATEST ;

EVALUATE FROM STREAM CDC "postgres-prd" TABLE "public.orders" ;

EVALUATE FROM STREAM NEO4J_CDC "neo4j-main" LABEL "Order" STARTING FROM TIMESTAMP "2026-01-01T00:00:00Z" ;

Important: connectionName is the logical identifier of a connection already registered on the platform — see the connections guide. Server addresses and credentials never appear in the script: they live in the platform's secure vault.

Stream transforms

PARSE TIMESTAMP <column>

Marks the column as event-time — a requirement for watermark/window.

|> PARSE TIMESTAMP ts

WATERMARK ON <column> DELAY "<duration>"

Tolerates out-of-order events up to delay after the timestamp.

|> WATERMARK ON ts DELAY "5s"

WINDOW ...

Three types:

|> WINDOW TUMBLING "30s"
|> WINDOW SLIDING  "60s" SLIDE "10s"
|> WINDOW SESSION  "5m"

Emission happens when the watermark passes the end of the window (tumbling/sliding) or when no events arrive within the gap (session).

CACHE AS LOOKUP [REFRESH EVERY "<duration>"]

Materializes a batch dataset in memory for enrichment:

DATASET users = FROM JDBC "warehouse" "SELECT id, nome, tier FROM users"
  |> CACHE AS LOOKUP REFRESH EVERY "30m" ;

JOIN ... AS LOOKUP

One-sided enrichment: each stream event looks its key up in the lookup in constant time — without slowing the flow down.

EVALUATE FROM STREAM KAFKA "my-kafka" TOPIC "events"
  |> JOIN users AS LOOKUP ON user_id = id
  |> STREAM OUTPUT ;

JOIN ... AS STREAM

Joining two streams is not supported yet — attempting it shows a message in Portuguese suggesting AS LOOKUP. Planned for a future version.

STREAM OUTPUT

Marks the pipeline as a streaming terminal. The subscription publishes events in real time in the interface until it is canceled.

Worked example — premium orders live

Scenario: you want to see, every 30 seconds, the orders from premium customers arriving through the Kafka topic orders, already enriched with the user registry from the warehouse.

1. Confirm that the orders-stream (Kafka) and warehouse (JDBC) connections are already registered on the platform.
2. In the DATTAX editor, paste and run:

-- Enrichment de pedidos ao vivo com dim de usuarios
DATASET users = FROM JDBC "warehouse" "SELECT id, nome, tier FROM users"
  |> CACHE AS LOOKUP REFRESH EVERY "30m" ;

EVALUATE FROM STREAM KAFKA "orders-stream" TOPIC "orders" STARTING FROM LATEST
  |> PARSE TIMESTAMP event_ts
  |> WATERMARK ON event_ts DELAY "5s"
  |> JOIN users AS LOOKUP ON user_id = id
  |> FILTER tier = "premium"
  |> WINDOW TUMBLING "30s"
  |> STREAM OUTPUT ;

1. Every time a window closes, new rows appear on screen. The subscription stays active — and listed in the background executions panel — until you cancel it.

Stream subscriptions can also be created, followed (including consumption lag) and canceled through an integration: the platform exposes endpoints to create the subscription, receive the events over WebSocket or through the alternative event channel, query the lag metrics and end the subscription — see the API reference.

Permissions

• DATTABI_STREAM_VIEW — subscribe/view (analyst+)
• DATTABI_STREAM_CREATE — create a subscription (advanced user+)
• DATTABI_STREAM_ADMIN — manage CDC connectors (admin)