What is the difference between Kinesis Data Analytics (managed Flink) and Amazon Kinesis Analytics (SQL-only), and when do you use each

package com.bank.flink.aws;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.CheckpointingMode;

public class ManagedFlinkFullFeature {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(60000, CheckpointingMode.EXACTLY_ONCE);

        // Full Flink feature set available:
        // - Event-time windows with watermarks
        // - Keyed state with RocksDB
        // - Timers and process functions
        // - CEP and MATCH_RECOGNIZE
        // - Flink SQL and Table API
        // - Rich connector ecosystem
        // - Exactly-once end-to-end

        env.fromElements("fraud-check-001")
            .map(check -> "Scored: " + check)
            .print();
        env.execute("NBI-Full-Feature-Managed");
    }
}

Managed Flink capabilities:
  ✓ DataStream API (Java, Scala, Python)
  ✓ Table API and Flink SQL
  ✓ Event-time processing with watermarks
  ✓ Keyed state (HashMap, RocksDB)
  ✓ Event-time and processing-time timers
  ✓ CEP library and MATCH_RECOGNIZE
  ✓ Temporal joins, windowed joins
  ✓ Async I/O
  ✓ Rich connectors (Kinesis, MSK, S3, JDBC, DynamoDB)
  ✓ Exactly-once with compatible sinks
  ✓ Elastic scaling with KPU autoscaling

-- SQL-only Kinesis Analytics example:
-- Simple tumbling window aggregate on a Kinesis stream
CREATE OR REPLACE STREAM output_stream (
    branch_id VARCHAR(32),
    total_amount DOUBLE,
    window_end TIMESTAMP
);

CREATE OR REPLACE PUMP output_pump AS
INSERT INTO output_stream
SELECT STREAM
    branch_id,
    SUM(amount) AS total_amount,
    STEP(transactions.ROWTIME BY INTERVAL '1' MINUTE) AS window_end
FROM transactions
GROUP BY branch_id, STEP(transactions.ROWTIME BY INTERVAL '1' MINUTE);

SQL-only limitations:
  ✗ No keyed state or RocksDB
  ✗ No event-time watermarks (processing time only)
  ✗ No complex joins or temporal joins
  ✗ No CEP or MATCH_RECOGNIZE
  ✗ No custom logic or UDFs
  ✗ Limited windowing (tumbling only, basic)
  ✗ No exactly-once guarantees
  ✗ Legacy path (AWS discourages new use)

package com.bank.flink.aws;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class ManagedFlinkSqlDemo {
    public static void main(String[] args) {
        // Even SQL-style logic runs on managed Flink
        // with the full engine underneath
        TableEnvironment tEnv = TableEnvironment.create(
            EnvironmentSettings.inStreamingMode());

        tEnv.executeSql(
            "SELECT account_id, SUM(amount) " +
            "FROM transactions " +
            "GROUP BY account_id"
        );
        // Full Flink SQL: temporal joins, MATCH_RECOGNIZE,
        // CDC, watermarks, state TTL all available
    }
}

Acceptable SQL-only use cases:
  - Simple filter: route high-value transactions to an alert stream
  - Basic count: count events per minute for a dashboard
  - Quick prototype: test an idea before building a real pipeline

Not acceptable:
  - Fraud detection (needs state, patterns, exactly-once)
  - CDC synchronization (needs changelog semantics)
  - Financial aggregation (needs event-time correctness)
  - Anything production-critical and long-lived

Migration steps:
  1. Identify SQL-only queries and their input/output streams
  2. Rewrite as Flink SQL (often 1:1 translation for simple queries)
  3. Add watermarks, state TTL, and proper sink configuration
  4. Deploy on managed Flink
  5. Validate output matches the old SQL-only results
  6. Decommission the SQL-only application

// Senior tip: use Flink SQL on managed Flink
// Gets the SQL simplicity the team wants
// Plus the full engine: temporal joins, CDC, CEP, exactly-once
// No ceiling, no rewrite risk