text: CREATE STREAM TEMPERATURE_READINGS (ID VARCHAR KEY, TIMESTAMP VARCHAR, READING BIGINT)
      WITH (KAFKA_TOPIC = 'TEMPERATURE_READINGS',
            VALUE_FORMAT = 'JSON',
            TIMESTAMP = 'TIMESTAMP',
            TIMESTAMP_FORMAT = 'yyyy-MM-dd HH:mm:ss',
            PARTITIONS = 1);
  
  CREATE TABLE TRIGGERED_ALERTS AS
      SELECT
          ID AS KEY,
          AS_VALUE(ID) AS ID,
          TIMESTAMPTOSTRING(WINDOWSTART, 'HH:mm:ss', 'UTC') AS START_PERIOD,
          TIMESTAMPTOSTRING(WINDOWEND, 'HH:mm:ss', 'UTC') AS END_PERIOD,
          SUM(READING)/COUNT(READING) AS AVG_READING
      FROM TEMPERATURE_READINGS
      WINDOW HOPPING (SIZE 10 MINUTES, ADVANCE BY 5 MINUTES)
      GROUP BY ID HAVING SUM(READING)/COUNT(READING) < 45;
  
  CREATE STREAM RAW_ALERTS (ID VARCHAR, START_PERIOD VARCHAR, END_PERIOD VARCHAR, AVG_READING BIGINT)
      WITH (KAFKA_TOPIC = 'TRIGGERED_ALERTS',
            VALUE_FORMAT = 'JSON');
  
  CREATE STREAM ALERTS AS
      SELECT
          ID,
          START_PERIOD,
          END_PERIOD,
          AVG_READING
      FROM RAW_ALERTS
      WHERE ID IS NOT NULL
      PARTITION BY ID;
  