CREATE OR REPLACE STREAM "source_stream" (
    "user_id" VARCHAR(64),
    "event_type" VARCHAR(32),
    "timestamp" TIMESTAMP,
    "amount" DECIMAL(10, 2)
);

CREATE OR REPLACE STREAM "aggregated_stream" (
    "user_id" VARCHAR(64),
    "total_amount" DECIMAL(10, 2),
    "event_count" INTEGER,
    "window_start" TIMESTAMP,
    "window_end" TIMESTAMP
);

CREATE OR REPLACE PUMP "aggregation_pump" AS
    INSERT INTO "aggregated_stream"
    SELECT
        "user_id",
        SUM("amount") AS "total_amount",
        COUNT(*) AS "event_count",
        STEP("source_stream". ROWTIME BY INTERVAL '5' MINUTE) AS "window_start",
        STEP("source_stream". ROWTIME BY INTERVAL '5' MINUTE) + INTERVAL '5' MINUTE AS "window_end"
    FROM "source_stream"
    GROUP BY
        "user_id",
        STEP("source_stream". ROWTIME BY INTERVAL '5' MINUTE)
    HAVING SUM("amount") > 100;

-- Tumbling window: 5-minute non-overlapping buckets
SELECT
    "user_id",
    SUM("amount") AS "total",
    WINDOW_START,
    WINDOW_END
FROM TABLE(
    TUMBLE(TABLE "source_stream", DESCRIPTOR("event_time"), INTERVAL '5' MINUTE)
)
GROUP BY "user_id", WINDOW_START, WINDOW_END;

-- Sliding window: 10-minute window, advancing every 2 minutes
SELECT
    "sensor_id",
    AVG("temperature") AS "avg_temp",
    MIN("temperature") AS "min_temp",
    MAX("temperature") AS "max_temp"
FROM TABLE(
    HOP(TABLE "sensor_stream", DESCRIPTOR("reading_time"),
        INTERVAL '2' MINUTE, INTERVAL '10' MINUTE)
)
GROUP BY "sensor_id", WINDOW_START, WINDOW_END;

-- Session window: closes after 30 minutes of inactivity
SELECT
    "user_id",
    COUNT(*) AS "page_views",
    MIN("event_time") AS "session_start",
    MAX("event_time") AS "session_end"
FROM TABLE(
    SESSION(TABLE "click_stream", DESCRIPTOR("event_time"), INTERVAL '30' MINUTE)
)
GROUP BY "user_id";

public class FraudDetectionJob {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env =
            StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(60000);
        env.getCheckpointConfig().setCheckpointingMode(
            CheckpointingMode.EXACTLY_ONCE
        );

        DataStream<Transaction> transactions = env.addSource(
            FlinkKinesisConsumer.<Transaction>builder()
                .setArn("arn:aws:kinesis:us-east-1:123456:stream/transactions")
                .setDeserializationSchema(new TransactionDeserializer())
                .build()
        );

        DataStream<FraudAlert> alerts = transactions
            .keyBy(Transaction::getUserId)
            .process(new FraudDetector())
            .name("fraud-detection")
            .uid("fraud-detector-001");

        alerts.addSink(
            FlinkKinesisProducer.<FraudAlert>builder()
                .setSerializationSchema(new FraudAlertSerializer())
                .setStream("fraud-alerts")
                .build()
        );

        env.execute("Fraud Detection Job");
    }
}

public class FraudDetector
    extends KeyedProcessFunction<String, Transaction, FraudAlert> {

    private ValueState<Double> totalAmount;
    private ValueState<Long> transactionCount;
    private ValueState<Long> lastTransactionTime;

    @Override
    public void open(Configuration parameters) {
        totalAmount = getRuntimeContext().getState(
            new ValueStateDescriptor<>("total-amount", Double.class)
        );
        transactionCount = getRuntimeContext().getState(
            new ValueStateDescriptor<>("tx-count", Long.class)
        );
        lastTransactionTime = getRuntimeContext().getState(
            new ValueStateDescriptor<>("last-tx-time", Long.class)
        );
    }

    @Override
    public void processElement(
            Transaction tx, Context ctx, Collector<FraudAlert> out)
            throws Exception {

        Double currentTotal = totalAmount.value();
        Long currentCount = transactionCount.value();
        Long lastTime = lastTransactionTime.value();

        if (currentTotal == null) currentTotal = 0.0;
        if (currentCount == null) currentCount = 0L;
        if (lastTime == null) lastTime = 0L;

        long now = tx.getTimestamp();
        long timeSinceLastTx = now - lastTime;

        totalAmount.update(currentTotal + tx.getAmount());
        transactionCount.update(currentCount + 1);
        lastTransactionTime.update(now);

        if (currentCount >= 4 &&
            timeSinceLastTx < 60000 &&
            (currentTotal + tx.getAmount()) > 500) {

            out.collect(new FraudAlert(
                tx.getUserId(),
                "RAPID_HIGH_VALUE_TRANSACTIONS",
                currentTotal + tx.getAmount(),
                now
            ));
        }

        ctx.timerService().registerProcessingTimeTimer(now + 60000);
    }

    @Override
    public void onTimer(long timestamp, OnTimerContext ctx,
            Collector<FraudAlert> out) {
        totalAmount.clear();
        transactionCount.clear();
        lastTransactionTime.clear();
    }
}

#!/bin/bash
# Create KDA SQL application

APPLICATION_NAME=$1
INPUT_STREAM=$2
OUTPUT_STREAM=$3

aws kinesisanalyticsv2 create-application \
    --application-name "${APPLICATION_NAME}" \
    --runtime-environment FLINK-1_18 \
    --service-execution-role "arn:aws:iam::123456789012:role/kda-role" \
    --application-configuration "{
        \"ApplicationCodeConfiguration\": {
            \"CodeContent\": {
                \"S3ContentLocation\": {
                    \"BucketARN\": \"arn:aws:s3:::my-kda-code\",
                    \"FileKey\": \"app.jar\"
                }
            },
            \"CodeContentType\": \"ZIPFILE\"
        },
        \"FlinkApplicationConfiguration\": {
            \"ParallelismConfiguration\": {
                \"AutoScalingEnabled\": true,
                \"Parallelism\": 4,
                \"ParallelismPerKPU\": 1,
                \"ConfigurationType\": \"CUSTOM\"
            },
            \"CheckpointConfiguration\": {
                \"CheckpointingEnabled\": true,
                \"CheckpointIntervalInMillis\": 60000,
                \"ConfigurationType\": \"CUSTOM\"
            }
        }
    }"

#!/bin/bash
# Deploy Flink application to KDA

APPLICATION_NAME=$1
JAR_PATH=$2

# Update application code
aws kinesisanalyticsv2 update-application \
    --application-name "${APPLICATION_NAME}" \
    --current-application-version-id $(aws kinesisanalyticsv2 describe-application \
        --application-name "${APPLICATION_NAME}" \
        --query 'ApplicationDetail.ApplicationVersionId' \
        --output text) \
    --application-configuration-update "{
        \"ApplicationCodeConfigurationUpdate\": {
            \"CodeContentUpdate\": {
                \"S3ContentLocationUpdate\": {
                    \"BucketARN\": \"arn:aws:s3:::my-kda-code\",
                    \"FileKey\": \"${JAR_PATH}\"
                }
            }
        }
    }"

#!/bin/bash
# Setup CloudWatch alarms for KDA

APPLICATION_NAME=$1

# CPU utilization alarm
aws cloudwatch put-metric-alarm \
    --alarm-name "${APPLICATION_NAME}-HighCPU" \
    --alarm-description "KDA application CPU high" \
    --metric-name "kinesis_analytics_cpu_utilization" \
    --namespace "AWS/KinesisAnalytics" \
    --statistic Average \
    --period 300 \
    --evaluation-periods 2 \
    --threshold 80 \
    --comparison-operator GreaterThanOrEqualToThreshold \
    --dimensions Name=Application,Value="${APPLICATION_NAME}" \
    --alarm-actions arn:aws:sns:us-east-1:123456789012:ops-alerts

# Heap usage alarm
aws cloudwatch put-metric-alarm \
    --alarm-name "${APPLICATION_NAME}-HighHeap" \
    --alarm-description "KDA application heap usage high" \
    --metric-name "kinesis_analytics_heap_used" \
    --namespace "AWS/KinesisAnalytics" \
    --statistic Average \
    --period 300 \
    --evaluation-periods 2 \
    --threshold 80 \
    --comparison-operator GreaterThanOrEqualToThreshold \
    --dimensions Name=Application,Value="${APPLICATION_NAME}" \
    --alarm-actions arn:aws:sns:us-east-1:123456789012:ops-alerts

-- Complex tumbling window with multiple aggregations
SELECT
    "category",
    COUNT(*) AS "total_events",
    SUM("amount") AS "total_amount",
    AVG("amount") AS "avg_amount",
    MIN("amount") AS "min_amount",
    MAX("amount") AS "max_amount",
    COUNT(DISTINCT "user_id") AS "unique_users",
    WINDOW_START,
    WINDOW_END
FROM TABLE(
    TUMBLE(TABLE "source_stream", DESCRIPTOR("event_time"), INTERVAL '1' HOUR)
)
GROUP BY "category", WINDOW_START, WINDOW_END;

-- Sliding window for moving averages
SELECT
    "sensor_id",
    AVG("reading") AS "moving_avg",
    STDDEV("reading") AS "moving_stddev",
    COUNT(*) AS "sample_count",
    WINDOW_START,
    WINDOW_END
FROM TABLE(
    HOP(TABLE "sensor_stream", DESCRIPTOR("reading_time"),
        INTERVAL '5' MINUTE, INTERVAL '1' HOUR)
)
GROUP BY "sensor_id", WINDOW_START, WINDOW_END;

-- Session window for user activity
SELECT
    "user_id",
    COUNT(*) AS "page_views",
    COUNT(DISTINCT "page") AS "unique_pages",
    MIN("event_time") AS "session_start",
    MAX("event_time") AS "session_end",
    TIMESTAMPDIFF(MINUTE, MIN("event_time"), MAX("event_time")) AS "session_duration"
FROM TABLE(
    SESSION(TABLE "click_stream", DESCRIPTOR("event_time"), INTERVAL '30' MINUTE)
)
GROUP BY "user_id";

public class ErrorHandler implements ExceptionHandler<Record> {
    @Override
    public void handleException(Record record, Throwable exception) {
        // Log the error
        LOG.error("Failed to process record: {}", record, exception);

        // Send to dead letter queue
        sendToDLQ(record, exception);

        // Update error metrics
        errorCounter.inc();
    }

    private void sendToDLQ(Record record, Throwable exception) {
        // Implementation for sending to DLQ
    }
}