Realtime data, one command away.

from basalt import Basalt

db = Basalt(project="acme-prod")

db.source("payments", kafka="payments.v3")

db.view("revenue_by_minute", """
  SELECT window(ts, '1m') AS minute, sum(amount) AS revenue
  FROM payments GROUP BY 1""")

rows = db.table("revenue_by_minute").last(3)

package main

import "github.com/basalt-dev/basalt-go"

func main() {
  db := basalt.New("acme-prod")
  db.Source("payments", basalt.Kafka("payments.v3"))
  db.View("revenue_by_minute", `
    SELECT window(ts, '1m') AS minute, sum(amount) AS revenue
    FROM payments GROUP BY 1`)
  rows, _ := db.Table("revenue_by_minute").Last(3)
}

-- views/revenue.sql · deploy with: basalt view apply
CREATE SOURCE payments FROM KAFKA 'payments.v3';

CREATE VIEW revenue_by_minute AS
SELECT window(ts, '1m') AS minute,
       sum(amount)        AS revenue
FROM payments
GROUP BY 1;

SELECT * FROM revenue_by_minute ORDER BY minute DESC LIMIT 3;