Kafka Topics from Flink

-- booknest.sql: Flink SQL over booknest.orders: revenue per channel and day, back into Kafka
-- (the SQL client splits statements at a semicolon that ends a line, so no comments after one)
SET 'table.dml-sync' = 'true';
CREATE TABLE orders (
  order_id BIGINT, order_ts STRING, channel STRING, status STRING,
  items ARRAY<ROW<book_id INT, qty INT, unit_price DOUBLE>>,
  ts AS TO_TIMESTAMP(order_ts, 'yyyy-MM-dd''T''HH:mm:ss''Z'''),   -- event time
  WATERMARK FOR ts AS ts - INTERVAL '1' HOUR
) WITH ('connector' = 'kafka', 'topic' = 'booknest.orders', 'format' = 'json',
  'properties.bootstrap.servers' = 'l3-kafka:9092',
  'scan.startup.mode' = 'earliest-offset', 'scan.bounded.mode' = 'latest-offset');
CREATE TABLE daily_revenue (day_start TIMESTAMP(3), channel STRING, lines BIGINT,
  revenue DOUBLE) WITH ('connector' = 'kafka', 'topic' = 'booknest.daily-revenue',
  'properties.bootstrap.servers' = 'l3-kafka:9092', 'format' = 'json');
CREATE VIEW order_lines AS
  SELECT o.ts, o.channel, i.qty * i.unit_price AS revenue
  FROM orders AS o CROSS JOIN UNNEST(o.items) AS i (book_id, qty, unit_price)
  WHERE o.status <> 'cancelled';
INSERT INTO daily_revenue
  SELECT window_start, channel, COUNT(*), ROUND(SUM(revenue), 2)
  FROM TABLE(TUMBLE(TABLE order_lines, DESCRIPTOR(ts), INTERVAL '1' DAY))
  GROUP BY window_start, window_end, channel;