# Flink Sql

> Apache Flink SQL, Table API, and UDF development for both OSS Flink and Confluent Cloud. Use when: (1) Writing Flink SQL queries (windows, joins, aggregations, MATCH_RECOGNIZE), (2) Building Table API pipelines in Java or Python, (3) Creating UDFs (scalar, table functions) for Flink, (4) Deploying Flink jobs to Confluent Cloud, (5) Converting between DataStream and Table API, (6) Troubleshooting…

- **Type:** Skill
- **Install:** `agentstack add skill-gamussa-flink-sql-skill-flink-sql`
- **Verified:** Pending review
- **Seller:** [gAmUssA](https://agentstack.voostack.com/s/gamussa)
- **Installs:** 0
- **Category:** [Databases](https://agentstack.voostack.com/c/databases)
- **Latest version:** 0.1.0
- **License:** Apache-2.0
- **Upstream author:** [gAmUssA](https://github.com/gAmUssA)
- **Source:** https://github.com/gAmUssA/flink-sql-skill/tree/main/tiles/flink-sql/skills/flink-sql

## Install

```sh
agentstack add skill-gamussa-flink-sql-skill-flink-sql
```

Requires the [AgentStack CLI](https://agentstack.voostack.com/docs/cli). Works with Claude Code, Cursor, and any MCP-compatible agent.

## About

# 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](references/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:**
```xml

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

```

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

**Python Table API:**
```bash
pip install confluent-flink-table-api-python-plugin
```

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

**Required environment variables:**
```bash
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)

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

## Flink SQL Patterns

### Window Aggregations

**Tumbling window** (fixed, non-overlapping):
```sql
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):
```sql
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):
```sql
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):
```sql
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):
```sql
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):
```sql
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

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

### Top-N

```sql
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)

```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

```java
// 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

```java
// 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

```java
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)

```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

```bash
# 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:**
```sql
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](https://github.com/gAmUssA)
- **Source:** [gAmUssA/flink-sql-skill](https://github.com/gAmUssA/flink-sql-skill)
- **License:** Apache-2.0

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

## Pricing

- **Free** — Free

## Security capabilities

Automated source analysis of v0.1.0 — what this tool can access:

- **Network access:** no
- **Filesystem access:** no
- **Shell / process execution:** no
- **Environment & secrets:** no
- **Dynamic code execution:** yes

*"Yes" means the capability is present in the source — more access means more to trust, not that it is unsafe.*


## Versions

- **0.1.0** — security scan: flagged — Imported from the upstream source.

## Links

- Listing page: https://agentstack.voostack.com/l/skill-gamussa-flink-sql-skill-flink-sql
- Seller: https://agentstack.voostack.com/s/gamussa
- Browse the marketplace: https://agentstack.voostack.com/browse

---
Listed on AgentStack — the marketplace for AI agent skills and MCP servers. Every listing is security-reviewed. Creators keep 70%.