Flink Sql

Apache Flink SQL & Table API

Stream processing with SQL semantics. Tables that change over time — think of them as movies, not photographs.

Quick Reference

| Task | Approach | |------|----------| | Simple transformations | Flink SQL SELECT/WHERE/GROUP BY | | Windowed aggregation | Window TVFs (TUMBLE/HOP/SESSION/CUMULATE) | | Pattern detection | MATCHRECOGNIZE | | Custom scalar logic | UDF (ScalarFunction) | | One-to-many expansion | UDTF (TableFunction) | | Stateful processing | Process Table Function (PTF) — see [ptf-guide.md](references/ptf-guide.md) | | Join streams | Interval joins, temporal joins, lookup joins | | Deduplication | ROWNUMBER() with OVER clause | | Top-N queries | ROWNUMBER() OVER (ORDER BY ...) | | Row-to-row comparison | LAG/LEAD window functions | | Late data routing | CURRENTWATERMARK() + Statement Sets | | Multi-level aggregation | Chained windows (fine → coarse) |

Mental Model: Streams ↔ Tables

Traditional DB: Table = snapshot (photograph)
Flink:          Table = changelog (movie)

Every INSERT/UPDATE/DELETE is an event in the changelog.
SQL queries become continuous — results update as data arrives.

Changelog modes:

• Insert-only: Append-only streams (logs, events)
• Upsert: Updates keyed by primary key (CDC, aggregations)
• Retract: Full changelog with retractions

Environment Setup

Confluent Cloud (Recommended for managed)

Java Table API:

  io.confluent.flink
  confluent-flink-table-api-java-plugin
  2.1-8

import io.confluent.flink.plugin.ConfluentSettings;
TableEnvironment env = TableEnvironment.create(
    ConfluentSettings.fromGlobalVariables()
);

Python Table API:

pip install confluent-flink-table-api-python-plugin

from pyflink.table.confluent import ConfluentSettings
from pyflink.table import TableEnvironment
settings = ConfluentSettings.from_global_variables()
env = TableEnvironment.create(settings)

Required environment variables:

export CLOUD_PROVIDER="aws"
export CLOUD_REGION="us-east-1"
export FLINK_API_KEY=""
export FLINK_API_SECRET=""
export ORG_ID=""
export ENV_ID=""
export COMPUTE_POOL_ID=""

OSS Flink (Self-managed)

EnvironmentSettings settings = EnvironmentSettings
    .newInstance()
    .inStreamingMode()
    .build();
TableEnvironment env = TableEnvironment.create(settings);

Flink SQL Patterns

Window Aggregations

Tumbling window (fixed, non-overlapping):

SELECT 
  window_start, window_end,
  COUNT(*) as cnt,
  SUM(amount) as total
FROM TABLE(
  TUMBLE(TABLE orders, DESCRIPTOR(event_time), INTERVAL '1' HOUR)
)
GROUP BY window_start, window_end;

Hopping window (overlapping):

SELECT window_start, window_end, AVG(price)
FROM TABLE(
  HOP(TABLE trades, DESCRIPTOR(ts), INTERVAL '5' MINUTE, INTERVAL '1' HOUR)
)
GROUP BY window_start, window_end;

Session window (gap-based):

SELECT window_start, window_end, user_id, COUNT(*)
FROM TABLE(
  SESSION(TABLE clicks, DESCRIPTOR(click_time), INTERVAL '30' MINUTE)
)
GROUP BY window_start, window_end, user_id;

Joins

Interval join (time-bounded):

SELECT o.*, s.ship_time
FROM orders o, shipments s
WHERE o.order_id = s.order_id
  AND s.ship_time BETWEEN o.order_time AND o.order_time + INTERVAL '4' HOUR;

Temporal join (point-in-time lookup):

SELECT o.*, r.rate
FROM orders o
JOIN currency_rates FOR SYSTEM_TIME AS OF o.order_time AS r
ON o.currency = r.currency;

Lookup join (external table):

SELECT o.*, c.name
FROM orders o
JOIN customers FOR SYSTEM_TIME AS OF o.proc_time AS c
ON o.customer_id = c.id;

Deduplication

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY event_time DESC) AS rn
  FROM events
)
WHERE rn = 1;

Top-N

SELECT *
FROM (
  SELECT *,
    ROW_NUMBER() OVER (PARTITION BY category ORDER BY sales DESC) AS rn
  FROM products
)
WHERE rn = 100
);

Table API Patterns

Basic Operations (Java)

Table orders = env.from("orders");

Table result = orders
    .filter($("status").isEqual("completed"))
    .select($("order_id"), $("amount"), $("customer_id"))
    .groupBy($("customer_id"))
    .select($("customer_id"), $("amount").sum().as("total"));

Mixing SQL and Table API

// SQL → Table
Table fromSql = env.sqlQuery("SELECT * FROM orders WHERE amount > 100");

// Table → SQL
env.createTemporaryView("filtered_orders", fromSql);
Table fromTable = env.sqlQuery("SELECT customer_id, SUM(amount) FROM filtered_orders GROUP BY customer_id");

Confluent-Specific: ConfluentTools

// Print results (limited)
ConfluentTools.printMaterializedLimit(table, 100);

// Collect results
List rows = ConfluentTools.collectMaterializedLimit(table, 100);

// Statement lifecycle
TableResult result = env.executeSql("SELECT * FROM orders");
String statementName = ConfluentTools.getStatementName(result);
ConfluentTools.stopStatement(result);

Confluent-Specific: ConfluentTableDescriptor

TableDescriptor descriptor = ConfluentTableDescriptor.forManaged()
    .schema(Schema.newBuilder()
        .column("id", DataTypes.INT())
        .column("data", DataTypes.STRING())
        .watermark("$rowtime", $("$rowtime").minus(lit(5).seconds()))
        .build())
    .build();
env.createTable("my_table", descriptor);

User-Defined Functions

For UDF development patterns, templates, and deployment: See [udf-guide.md](references/udf-guide.md)

Quick UDF Example (Java)

public class MyUpperCase extends ScalarFunction {
    public String eval(String s) {
        return s == null ? null : s.toUpperCase();
    }
}

// Register and use
env.createTemporaryFunction("my_upper", MyUpperCase.class);
env.sqlQuery("SELECT my_upper(name) FROM users");

Confluent Cloud UDF Deployment

# Build JAR
mvn clean package

# Upload artifact
confluent flink artifact create my-udf \
  --cloud aws --region us-east-1 \
  --artifact-file target/my-udf-1.0.jar

# Register function
CREATE FUNCTION my_upper 
AS 'com.example.MyUpperCase' 
USING JAR 'confluent-artifact://cfa-xxxxx';

Watermarks & Event Time

Declare watermark in DDL:

CREATE TABLE events (
  event_id STRING,
  event_time TIMESTAMP(3),
  payload STRING,
  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (...);

Watermark strategies:

• event_time - INTERVAL 'n' SECOND — bounded out-of-orderness
• event_time — strictly ascending (no late data)
• SOURCE_WATERMARK() — preserve source watermarks

Critical Rules

1. Always declare watermarks for event-time processing
2. State TTL matters — unbounded state = memory explosion
3. Schema must match — SQL expects explicit type declarations
4. Know your changelog mode — insert-only vs updating affects downstream
5. Confluent Cloud limits:

• Max 10 UDFs per statement
• Max 100 artifacts per environment
• JDK 17 max for uploaded JARs
• No aggregate UDFs (only scalar and table functions)
• No MATCH_RECOGNIZE with UDFs

Troubleshooting

| Error | Cause | Fix | |-------|-------|-----| | Cannot resolve watermark | Missing watermark declaration | Add WATERMARK FOR col AS ... | | Schema mismatch | Column types don't align | Check data types with DESCRIBE table | | State too large | Unbounded aggregation | Add state TTL or use windows | | Late data dropped | Watermark too aggressive | Increase watermark delay, or route late data with CURRENT_WATERMARK() (see sql-patterns.md) | | UDF not found | Function not registered | Check catalog/database scope |

For detailed troubleshooting: See [troubleshooting.md](references/troubleshooting.md)

Reference Files

• [udf-guide.md](references/udf-guide.md) — UDF development, templates, deployment
• [sql-patterns.md](references/sql-patterns.md) — Advanced SQL patterns and examples
• [confluent-cloud.md](references/confluent-cloud.md) — Confluent-specific features and CLI
• [troubleshooting.md](references/troubleshooting.md) — Common errors and solutions
• [ptf-guide.md](references/ptf-guide.md) — Process Table Functions (stateful operators)
• [kafka-patterns.md](references/kafka-patterns.md) — Kafka connector patterns, Avro/SR, Flink vs Kafka Streams

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: gAmUssA
• Source: gAmUssA/flink-sql-skill
• License: Apache-2.0

Install and usage instructions live in the source repository linked above.