Aiven for Apache Flink

-- Calculate metrics per 5-minute tumbling window
CREATE TABLE window_metrics AS
SELECT
    TUMBLE_START(event_time, INTERVAL '5' MINUTE) AS window_start,
    TUMBLE_END(event_time, INTERVAL '5' MINUTE) AS window_end,
    sensor_id,
    COUNT(*) AS reading_count,
    AVG(temperature) AS avg_temperature,
    MAX(temperature) AS max_temperature,
    MIN(temperature) AS min_temperature
FROM kafka_sensor_data
GROUP BY 
    TUMBLE(event_time, INTERVAL '5' MINUTE),
    sensor_id;

-- Join clickstream with purchases (within 1 hour)
CREATE TABLE attributed_purchases AS
SELECT
    p.purchase_id,
    p.user_id,
    p.amount,
    c.page_url,
    c.referrer,
    c.campaign_id
FROM kafka_purchases p
JOIN kafka_clicks c
    ON p.user_id = c.user_id
    AND p.purchase_time BETWEEN c.click_time 
        AND c.click_time + INTERVAL '1' HOUR;

-- Process database changes from Debezium CDC
CREATE TABLE customer_changes (
    customer_id BIGINT,
    name STRING,
    email STRING,
    op_type STRING -- 'c' for create, 'u' for update, 'd' for delete
) WITH (
    'connector' = 'kafka',
    'topic' = 'dbserver1.customers',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'debezium-json'
);

-- Materialize current state
CREATE TABLE current_customers AS
SELECT 
    customer_id,
    name,
    email
FROM customer_changes
WHERE op_type <> 'd';

CREATE TABLE kafka_events (
    event_id STRING,
    event_time TIMESTAMP(3),
    user_id BIGINT
) WITH (
    'connector' = 'kafka',
    'topic' = 'events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'avro',
    'avro.schema' = '{...}'
);

CREATE TABLE kafka_events (
    event_id STRING,
    event_time TIMESTAMP(3),
    user_id BIGINT
) WITH (
    'connector' = 'kafka',
    'topic' = 'events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'avro-confluent',
    'avro-confluent.url' = 'https://schema-registry:8081'
);

CREATE TABLE user_stats (
    user_id BIGINT PRIMARY KEY NOT ENFORCED,
    total_purchases BIGINT,
    total_amount DECIMAL(10,2),
    last_purchase_time TIMESTAMP(3)
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'user-stats',
    'properties.bootstrap.servers' = 'kafka:9092',
    'key.format' = 'json',
    'value.format' = 'json'
);