Apache Flink

# Download e instalação
wget https://archive.apache.org/dist/flink/flink-1.18.0/flink-1.18.0-bin-scala_2.12.tgz
tar -xzf flink-1.18.0-bin-scala_2.12.tgz
cd flink-1.18.0

# Iniciar cluster local
./bin/start-cluster.sh

# Verificar Web UI
# http://localhost:8081

# Parar cluster
./bin/stop-cluster.sh

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

public class WordCountStreaming {
    public static void main(String[] args) throws Exception {
        // Criar execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // Ler dados do socket
        DataStream text = env.socketTextStream("localhost", 9999);
        
        // Processar dados
        DataStream> wordCounts = text
            .flatMap(new Tokenizer())
            .keyBy(value -> value.f0)
            .timeWindow(Time.seconds(5))
            .sum(1);
        
        // Output
        wordCounts.print();
        
        // Executar
        env.execute("Word Count Streaming");
    }
    
    public static class Tokenizer implements FlatMapFunction> {
        @Override
        public void flatMap(String value, Collector> out) {
            String[] words = value.toLowerCase().split("\\W+");
            for (String word : words) {
                if (word.length() > 0) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        }
    }
}

from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment
from pyflink.table.descriptors import Schema, Kafka, Json

def main():
    # Criar environments
    env = StreamExecutionEnvironment.get_execution_environment()
    t_env = StreamTableEnvironment.create(env)
    
    # Configurar Kafka source
    t_env.connect(Kafka()
                  .version("universal")
                  .topic("input-topic")
                  .start_from_earliest()
                  .property("bootstrap.servers", "localhost:9092")) \
        .with_format(Json()
                     .derive_schema()) \
        .with_schema(Schema()
                     .field("user_id", "STRING")
                     .field("event_type", "STRING")
                     .field("timestamp", "TIMESTAMP(3)")
                     .field("value", "DOUBLE")) \
        .create_temporary_table("source_table")
    
    # Processar dados
    result = t_env.sql_query("""
        SELECT 
            user_id,
            event_type,
            COUNT(*) as event_count,
            AVG(value) as avg_value,
            TUMBLE_START(timestamp, INTERVAL '1' MINUTE) as window_start
        FROM source_table
        GROUP BY 
            user_id, 
            event_type,
            TUMBLE(timestamp, INTERVAL '1' MINUTE)
    """)
    
    # Output para console
    result.execute_insert("print").wait()

if __name__ == '__main__':
    main()

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;

// Definir padrão para detecção de fraude
Pattern fraudPattern = Pattern.begin("first")
    .where(new SimpleCondition() {
        @Override
        public boolean filter(Transaction transaction) {
            return transaction.getAmount() > 1000;
        }
    })
    .next("second")
    .where(new SimpleCondition() {
        @Override
        public boolean filter(Transaction transaction) {
            return transaction.getAmount() > 1000;
        }
    })
    .within(Time.minutes(5));

// Aplicar padrão ao stream
PatternStream patternStream = CEP.pattern(
    transactionStream.keyBy(Transaction::getUserId),
    fraudPattern
);

// Processar matches
DataStream alerts = patternStream.select(
    (Map> pattern) -> {
        List first = pattern.get("first");
        List second = pattern.get("second");
        return new Alert("Possible fraud detected", 
                        first.get(0).getUserId());
    }
);