Flink SQL

Dynamic Tables Concept:

Stream → Dynamic Table → Continuous Query → Dynamic Result Table → Stream

1. Stream to Table:
   Events: (10:01, Alice, click), (10:02, Bob, purchase), (10:03, Alice, click)

   Dynamic Table "events":
   Time  │ User  │ Action
   ──────┼───────┼─────────
   10:01 │ Alice │ click      ← row added
   10:02 │ Bob   │ purchase   ← row added
   10:03 │ Alice │ click      ← row added

2. Continuous Query:
   SELECT user, COUNT(*) AS cnt FROM events GROUP BY user

   Result table (updates over time):
   After event 1: { Alice: 1 }
   After event 2: { Alice: 1, Bob: 1 }
   After event 3: { Alice: 2, Bob: 1 }  ← Alice's count UPDATED

3. Result Table to Stream:
   Append stream: (+Alice,1), (+Bob,1), (+Alice,2)  ← wrong! overwrites
   Retract stream: (+Alice,1), (+Bob,1), (-Alice,1), (+Alice,2)  ← correct!
   Upsert stream: (Alice→1), (Bob→1), (Alice→2)  ← with primary key

-- Processing time: computed column
CREATE TABLE pageviews (
    user_id STRING,
    page STRING,
    proc_time AS PROCTIME()  -- processing time attribute
) WITH ('connector' = 'kafka', ...);

-- Event time: from a column + watermark declaration
CREATE TABLE orders (
    order_id STRING,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    -- Watermark: tolerate 5 seconds of out-of-orderness
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka', ...);

-- Tumbling window aggregation (event time)
SELECT
    user_id,
    TUMBLE_START(order_time, INTERVAL '1' HOUR) AS window_start,
    TUMBLE_END(order_time, INTERVAL '1' HOUR) AS window_end,
    COUNT(*) AS order_count,
    SUM(amount) AS total_amount
FROM orders
GROUP BY
    user_id,
    TUMBLE(order_time, INTERVAL '1' HOUR);

-- Sliding window (HOP)
SELECT
    user_id,
    HOP_START(order_time, INTERVAL '5' MINUTE, INTERVAL '1' HOUR) AS ws,
    COUNT(*) AS cnt
FROM orders
GROUP BY
    user_id,
    HOP(order_time, INTERVAL '5' MINUTE, INTERVAL '1' HOUR);

-- Session window
SELECT
    user_id,
    SESSION_START(order_time, INTERVAL '30' MINUTE) AS session_start,
    COUNT(*) AS events_in_session
FROM orders
GROUP BY
    user_id,
    SESSION(order_time, INTERVAL '30' MINUTE);

-- Source table: Kafka topic with JSON format
CREATE TABLE user_events (
    event_id STRING,
    user_id STRING,
    event_type STRING,
    page STRING,
    event_time TIMESTAMP(3),
    -- Metadata columns from Kafka
    kafka_partition INT METADATA FROM 'partition' VIRTUAL,
    kafka_offset BIGINT METADATA FROM 'offset' VIRTUAL,
    -- Watermark for event-time processing
    WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'user-events',
    'properties.bootstrap.servers' = 'kafka:9092',
    'properties.group.id' = 'flink-sql-analytics',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json',
    'json.fail-on-missing-field' = 'false',
    'json.ignore-parse-errors' = 'true'
);

-- Sink table: Kafka topic for results
CREATE TABLE hourly_stats (
    user_id STRING,
    window_start TIMESTAMP(3),
    event_count BIGINT,
    unique_pages BIGINT,
    PRIMARY KEY (user_id, window_start) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'hourly-user-stats',
    'properties.bootstrap.servers' = 'kafka:9092',
    'format' = 'json',
    'sink.delivery-guarantee' = 'exactly-once',
    'sink.transactional-id-prefix' = 'sql-job-1'
);

-- Continuous query: insert results into sink
INSERT INTO hourly_stats
SELECT
    user_id,
    TUMBLE_START(event_time, INTERVAL '1' HOUR) AS window_start,
    COUNT(*) AS event_count,
    COUNT(DISTINCT page) AS unique_pages
FROM user_events
GROUP BY
    user_id,
    TUMBLE(event_time, INTERVAL '1' HOUR);

Changelog Modes:

1. Append-only (INSERT only):
   Used when results only grow (no updates/deletes)
   Example: SELECT * FROM events (no aggregation)
   Output: +I[Alice, click, 10:01]
           +I[Bob, purchase, 10:02]

2. Retract (INSERT + DELETE pairs):
   Used when results update (old value retracted, new value inserted)
   Example: SELECT user, COUNT(*) FROM events GROUP BY user
   Output: +I[Alice, 1]      ← insert Alice=1
           +I[Bob, 1]        ← insert Bob=1
           -D[Alice, 1]      ← retract old Alice value
           +I[Alice, 2]      ← insert new Alice value

3. Upsert (UPSERT + DELETE):
   Used when results have a primary key (update by key)
   Example: Same query, but sink has PRIMARY KEY (user)
   Output: +U[Alice, 1]      ← upsert Alice=1
           +U[Bob, 1]        ← upsert Bob=1
           +U[Alice, 2]      ← upsert Alice=2 (overwrites)

Sink compatibility:
  Kafka (append): only append-only queries
  Kafka (upsert): queries with PRIMARY KEY → upsert-kafka connector
  JDBC: supports all modes (INSERT, UPDATE, DELETE)
  Elasticsearch: upsert mode with document ID

-- Versioned table: exchange rates that change over time
CREATE TABLE exchange_rates (
    currency STRING,
    rate DECIMAL(10,6),
    update_time TIMESTAMP(3),
    WATERMARK FOR update_time AS update_time - INTERVAL '5' SECOND,
    PRIMARY KEY (currency) NOT ENFORCED
) WITH (
    'connector' = 'kafka',
    'topic' = 'exchange-rates',
    'format' = 'debezium-json'  -- CDC format for versioned table
);

-- Fact table: orders stream
CREATE TABLE orders (
    order_id STRING,
    currency STRING,
    amount DECIMAL(10,2),
    order_time TIMESTAMP(3),
    WATERMARK FOR order_time AS order_time - INTERVAL '5' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'orders',
    'format' = 'json'
);

-- Temporal join: get exchange rate valid at order time
SELECT
    o.order_id,
    o.amount,
    o.currency,
    r.rate,
    o.amount * r.rate AS usd_amount
FROM orders AS o
JOIN exchange_rates FOR SYSTEM_TIME AS OF o.order_time AS r
ON o.currency = r.currency;

-- This gives the rate that was valid WHEN the order was placed
-- Not the current rate (which may have changed since)

Flink SQL Gateway Architecture:

┌─────────────┐     ┌──────────────────┐     ┌──────────────┐
│  SQL Client │────→│  SQL Gateway     │────→│ Flink Cluster│
│  (CLI/JDBC) │     │  (REST API)      │     │              │
└─────────────┘     └──────────────────┘     └──────────────┘
                    │ Session management │
                    │ Query parsing      │
                    │ Result caching     │

Interfaces:
  - REST API: POST /v1/sessions/{id}/statements
  - JDBC: jdbc:flink://gateway-host:8083
  - HiveServer2 compatible: works with existing BI tools

Use cases:
  - Interactive SQL development (Flink SQL CLI)
  - BI tool integration (Tableau, Superset via JDBC)
  - Multi-tenant SQL execution (shared gateway, isolated sessions)
  - Ad-hoc queries on streaming data

Starting the gateway:
  ./bin/sql-gateway.sh start -Dsql-gateway.endpoint.rest.address=0.0.0.0