Apache Flink Provider Guide¶

This guide covers how to construct and run streaming pipelines using the Apache Flink providers.

Available Flink Providers¶

Provider	Class	Capability
Flink Stream Source	`FlinkStreamSourceProvider`	Source Acquisition (streaming files, topics)
Flink Transformation	`FlinkTransformationProvider`	Transformation (SQL + windowed aggregations)
Flink Storage	`FlinkStorageProvider`	Storage (rolling file sinks)

When to Use Flink vs Spark¶

Criterion	Use Spark	Use Flink
Processing model	Batch, micro-batch	True streaming, event-time
Latency requirement	Minutes	Seconds to milliseconds
Windowed aggregations	Limited	Native (tumbling, sliding, session)
State management	Checkpoint-based	Native stateful processing
File formats	CSV, Parquet, JSON, Delta, ORC, Avro	CSV, JSON, Parquet
JDBC support	Full (source + sink)	Limited
Exactly-once	Via Delta/checkpoints	Via Flink checkpoints

File Stream Source Configuration¶

Monitor a directory for new files and process them continuously.

Configuration Keys¶

Key	Required	Default	Description
`path`	✅*	—	Directory to monitor for new files
`topic`	✅*	—	Topic name (Kafka-like sources)
`format`	✅	—	`csv`, `json`, `parquet`
`monitorInterval`	❌	`PT10S`	How often to check for new files

*Either path or topic must be provided.

Acquisition Modes¶

Mode	Behaviour
`streaming`	Continuous file monitoring with configurable interval
`snapshot`	Bounded read of all current data (batch mode)

Example: File Monitoring Source¶

- name: consume-events
  type: source
  capability: source-acquisition
  provider: flink-stream-source
  timeout: PT1H
  config:
    path: /data/streaming/events/
    format: json
    monitorInterval: PT10S

Example: CSV File Source (Batch)¶

- name: read-csv-batch
  type: source
  capability: source-acquisition
  provider: flink-stream-source
  timeout: PT5M
  config:
    path: /data/input/customers/
    format: csv

Transformation Configuration (Flink SQL)¶

Transform data using Flink SQL with support for windowed aggregations.

Transformation Styles¶

Style	Description	Use Case
`sql`	Standard SQL query	Filters, joins, aggregations
`stream`	Streaming SQL with windows	Tumbling, sliding, session windows

The `{input}` Placeholder¶

Same as Spark — {input} is replaced with the registered table name:

SELECT user_id, COUNT(*) AS event_count
FROM {input}
WHERE event_type = 'purchase'
GROUP BY user_id

Windowed Aggregations¶

Flink's Table-Valued Functions enable time-based windowing:

Tumbling Window (Fixed-size, non-overlapping)¶

transformation:
  sql:
    style: stream
    reference: |
      SELECT 
        window_start,
        window_end,
        user_id,
        COUNT(*) AS event_count,
        SUM(amount) AS total_amount
      FROM TABLE(
        TUMBLE(TABLE {input}, DESCRIPTOR(event_time), INTERVAL '5' MINUTE)
      )
      GROUP BY window_start, window_end, user_id

Sliding Window (Fixed-size, overlapping)¶

transformation:
  sql:
    style: stream
    reference: |
      SELECT 
        window_start,
        window_end,
        category,
        AVG(price) AS avg_price
      FROM TABLE(
        HOP(TABLE {input}, DESCRIPTOR(event_time), INTERVAL '1' MINUTE, INTERVAL '5' MINUTE)
      )
      GROUP BY window_start, window_end, category

Session Window (Gap-based)¶

transformation:
  sql:
    style: stream
    reference: |
      SELECT 
        session_start,
        session_end,
        user_id,
        COUNT(*) AS actions
      FROM TABLE(
        SESSION(TABLE {input}, DESCRIPTOR(event_time), INTERVAL '30' MINUTE)
      )
      GROUP BY session_start, session_end, user_id

Example: Simple Filter¶

- name: filter-purchases
  type: transformation
  capability: transformation
  provider: flink-transformation
  timeout: PT1H
  transformation:
    sql:
      style: sql
      reference: "SELECT * FROM {input} WHERE event_type = 'purchase' AND amount > 0"

Storage Configuration (File Sinks)¶

Write streaming results to rolling files.

Write Options¶

Key	Required	Default	Description
`format`	❌	`parquet`	Output format: `parquet`, `json`, `csv`
`mode`	❌	`append`	Write mode: `append`, `overwrite`
`rollingPolicy`	❌	`size`	Rolling policy: `size`, `time`
`rolloverSize`	❌	`128MB`	Max file size before rolling (size policy)

Example: Parquet Sink with Size-Based Rolling¶

- name: write-aggregates
  type: publish
  capability: storage
  provider: flink-storage
  timeout: PT1H
  config:
    format: parquet
    mode: append
    rollingPolicy: size
    rolloverSize: 128MB

Example: JSON Sink with Time-Based Rolling¶

- name: write-events
  type: publish
  capability: storage
  provider: flink-storage
  timeout: PT1H
  config:
    format: json
    mode: append
    rollingPolicy: time

Complete Streaming Pipeline Example¶

name: streaming-event-aggregation
tenancy: example-org
namespace: production
dataset: user-events
executionMode: streaming

stages:
  - name: consume-events
    type: source
    capability: source-acquisition
    provider: flink-stream-source
    timeout: PT1H
    config:
      path: /data/streaming/events/
      format: json
      monitorInterval: PT10S

  - name: window-aggregate
    type: transformation
    capability: transformation
    provider: flink-transformation
    timeout: PT1H
    transformation:
      sql:
        style: stream
        reference: |
          SELECT 
            window_start,
            window_end,
            user_id,
            COUNT(*) AS event_count,
            SUM(amount) AS total_amount
          FROM TABLE(
            TUMBLE(TABLE {input}, DESCRIPTOR(event_time), INTERVAL '5' MINUTE)
          )
          GROUP BY window_start, window_end, user_id

  - name: write-aggregates
    type: publish
    capability: storage
    provider: flink-storage
    timeout: PT1H
    config:
      format: parquet
      mode: append
      rollingPolicy: size
      rolloverSize: 128MB

policies: []
extensions: []

Checkpoint and Exactly-Once Semantics¶

Flink provides exactly-once processing guarantees through checkpointing:

Checkpointing is enabled by default (every 5 seconds in local mode)
Savepoints can be triggered manually for planned maintenance
Resume from checkpoint restores the full processing state

The FlinkStreamSourceProvider supports: - checkpoint() — captures current processing state - resume(checkpoint) — restores from a previous checkpoint

Troubleshooting¶

Issue	Cause	Solution
`ClassNotFoundException: flink-table-planner`	Missing planner dependency	Add `flink-table-planner-loader` to classpath
`No suitable driver found`	JDBC driver not available	Add driver JAR to Flink's lib directory
`Code generation failed`	JDK compiler not on classpath	Ensure JDK (not JRE) is used
`Checkpoint expired`	Processing too slow	Increase checkpoint interval or parallelism
`Schema mismatch`	Source schema changed	Use explicit schema declaration in DDL

Programmatic Usage¶

import io.patternops.providers.flink.FlinkEnvironmentFactory;
import io.patternops.providers.flink.source.FlinkStreamSourceProvider;
import io.patternops.providers.flink.transform.FlinkTransformationProvider;
import io.patternops.providers.flink.storage.FlinkStorageProvider;

// Create Flink environment
var env = FlinkEnvironmentFactory.createLocal();
var tableEnv = FlinkEnvironmentFactory.createTableEnvironment(env);

// Configure source
FlinkStreamSourceProvider source = new FlinkStreamSourceProvider();
source.initialise(Map.of(
    "path", "/data/events/",
    "format", "json",
    "streamExecutionEnvironment", env,
    "tableEnvironment", tableEnv
));

// Acquire data
var dataStream = source.acquire("streaming", Map.of());
String tableName = dataStream.properties().get("tableName").toString();

// Transform
FlinkTransformationProvider transformer = new FlinkTransformationProvider();
transformer.initialise(Map.of(
    "streamExecutionEnvironment", env,
    "tableEnvironment", tableEnv
));

var result = transformer.execute(
    new TransformationDefinition("sql", "SELECT * FROM " + tableName + " WHERE amount > 0", Map.of()),
    tableName,
    "filtered_events"
);