gamussa/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 | MATCH_RECOGNIZE | | Custom scalar logic | UDF (ScalarFunction) | | One-to-many expansion | UDTF (TableFunction) | | Stateful processing | Process Table Function (PTF) — see ptf-guide.md | | Join streams | Interval joins, temporal joins, lookup joins | | Deduplication | ROW_NUMBER() with OVER clause | | Top-N queries | ROW_NUMBER() OVER (ORDER BY ...) | | Row-to-row comparison | LAG/LEAD window functions | | Late data routing | CURRENT_WATERMARK() + 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:

<dependency>
  <groupId>io.confluent.flink</groupId>
  <artifactId>confluent-flink-table-api-java-plugin</artifactId>
  <version>2.1-8</version>
</dependency>

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="<key>"
export FLINK_API_SECRET="<secret>"
export ORG_ID="<org-id>"
export ENV_ID="<env-id>"
export COMPUTE_POOL_ID="<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 <= 10;

Pattern Detection (MATCH_RECOGNIZE)

SELECT *
FROM orders
MATCH_RECOGNIZE (
  PARTITION BY user_id
  ORDER BY event_time
  MEASURES
    FIRST(A.event_time) AS start_time,
    LAST(B.event_time) AS end_time,
    COUNT(A.amount) AS cnt
  ONE ROW PER MATCH
  PATTERN (A+ B)
  DEFINE
    A AS A.amount < 100,
    B AS B.amount >= 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<Row> 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

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

Reference Files

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