Evolução das APIs de Processamento de Fluxo

Existem diferentes abordagens para o processamento de streams em sistemas distribuídos. O Storm utiliza o TopologyBuilder para construir grafos, onde o desenvolvedor adiciona nós e especifica arestas direcionadas entre eles. Após a construção, o grafo pode ser submetido a um cluster remoto ou local para execução.

O Flink adota uma abordagem diferente, orientada aos dados. O DataStream é tratado como uma coleção local, permitindo programação orientada a fluxos de dados. Enquanto o Storm oferece maior flexibilidade e controle em nível mais baixo, o Flink proporciona desenvolvimento mais simples e eficiente. Embora ambas as abordagens convergam para abstrações internas similares, a API do Flink é considerada mais alto nível por ser orientada a dados.

Fluxo Básico de Processamento

Como outras engines de processamento distribuído, aplicações Flink seguem um padrão de programação específico. Seja usando a API DataStream ou DataSet, a estrutura permanece consistente. O exemplo abaixo demonstra a contagem de palavras em um arquivo de texto utilizando processamento de fluxo.

public class ContadorPalavras {
    public static void main(String[] args) throws Exception {
        // 1 - Obter ambiente de execução
        Environment execEnv = Environment.getExecutionEnvironment();
        
        // 2 - Criar fonte de dados a partir de socket
        DataSource<String> entrada = execEnv.connectSocket("192.168.52.12", 9000);
        
        // 3 - Processar dados: flattening e contagem
        DataFlow<Par<String, Integer>> fluxoProcessado = entrada
            .flatMap(new SeparadorPalavras())
            .groupBy(0)
            .sum(1);
        
        // 4 - Exibir resultado no console
        fluxoProcessado.output();
        
        // 5 - Executar o job
        execEnv.run("contagem de palavras em stream");
    }
}

class SeparadorPalavras implements FlatMapFunction<String, Par<String, Integer>> {
    public void flatMap(String texto, Collector<Par<String, Integer>> out) {
        String[] tokens = texto.split(" ");
        for (String palavra : tokens) {
            out.collect(new Par<>(palavra, 1));
        }
    }
}

Um programa Flink tipico contém cinco etapas fundamentais: configuração do ambiente de execução, criação e carregamento de dados, definição de transformações, especificação do destino dos resultados, e调用 do método de execução.

Visão Geral das Operações

Ao processar um fluxo de dados, diversas operações podem ser aplicadas:

• Operações por registro: Filtros e mapeamentos aplicados individualmente
• Operações por janela: Cálculos baseados em períodos temporais ou contagem
• Mesclagem de fluxos: União, junção e conexão de múltiplos streams
• Divisão de fluxos: Separação de um stream em múltiplos baseados em condições

Transformações Básicas do DataStream

O DataStream oferece diversas transformações:

1. Transformações 1:1: Operações como map que mantêm o tipo DataStream
2. Divisão: Uso de split para separar um stream em múltiplos com etiquetas
3. Seleção: Extração de streams específicos via método select
4. Conexão: Junção de dois streams diferentes via connect formando ConnectedStreams
5. CoMap: Operações de mapeamento em ConnectedStreams
6. Janelas: Segmentação temporal ou por contagem usando windowAll
7. Aplicação em janelas: Método apply em AllWindowedStream
8. Group by key: Transformação para KeyedStream via keyBy
9. Reduce em KeyedStream: Agregações otimizadas por chave
10. Janelas em KeyedStream: Formação de WindowedStream

Environment de Execução

O ambiente de execução determina onde o programa será executado:

// Ambiente para processamento de stream
StreamEnvironment ambienteStream = StreamEnvironment.getEnvironment();

// Ambiente para processamento em lote
BatchEnvironment ambienteBatch = BatchEnvironment.getEnvironment();

O método getExecutionEnvironment detecta automaticamente o contexto de execução. A paralelização pode ser configurada:

// Definir paralelismo (sobrescreve configuração padrão)
ambienteStream.setParallelism(4);

// Criar ambiente local com paralelismo específico
StreamEnvironment localEnv = StreamEnvironment.createLocalEnvironment(2);

// Criar ambiente remoto para submitão em cluster
BatchEnvironment remoteEnv = BatchEnvironment.createRemoteEnvironment(
    "endereco-jobmanager", 
    6123, 
    "/caminho/app.jar"
);

Fonte de Dados (Source)

Após criar o ambiente, dados externos precisam ser importados. O Flink oferece múltiplos conectores:

// Ler arquivo de texto
DataSource<String> texto = ambiente.readTextFile("file:///caminho/arquivo");

// Criar stream a partir de coleção
DataSource<Pessoa> colecao = ambiente.fromCollection(listaPessoas);

// Stream de socket (uso menos comum)
DataSource<String> socket = ambiente.socketTextStream("localhost", 777);

// Elementos diretos para testes
DataSource<Object> elementos = ambiente.fromElements(1, 4, "teste");

Para Kafka, é necessário adicionar a dependência do connector correspondente:

// Configuração do consumidor Kafka
Properties config = new Properties();
config.put("bootstrap.servers", "servidor1:9092");
config.put("group.id", "grupo-consumidor");
config.put("enable.auto.commit", "true");
config.put("key.deserializer", StringDeserializer.class);
config.put("value.deserializer", StringDeserializer.class);

// Criar consumidor Kafka
KafkaConsumer<String> consumidor = new FlinkKafkaConsumer<>(
    "meu-topico", 
    new SimpleStringSchema(), 
    config
);

DataSource<String> streamKafka = ambiente.addSource(consumidor);

Fonte Personalizada

Para fontes especiais ou dados de teste, pode-se implementar uma SourceFunction:

public class GeradorDadosSensor {
    public static void main(String[] args) {
        StreamEnvironment env = StreamEnvironment.getEnvironment();
        
        // Adicionar fonte personalizada
        DataSource<LeituraSensor> stream = env.addSource(new GeradorSensor());
        
        stream.output();
        env.run("geração de dados sensor");
    }
}

class GeradorSensor implements SourceFunction<LeituraSensor> {
    private boolean ativo = true;
    private Random random = new Random();
    
    @Override
    public void run(SourceContext<LeituraSensor> ctx) {
        // Gerar 10 sensores iniciais
        List<Temperatura> medicoes = IntStream.range(0, 10)
            .mapToObj(i -> new Temperatura("sensor_" + i, 60 + random.nextGaussian() * 10))
            .collect(Collectors.toList());
        
        while (ativo) {
            // Atualizar temperaturas com pequenas variações
            medicoes = medicoes.stream()
                .map(t -> new Temperatura(
                    t.id, 
                    t.valor + random.nextGaussian()
                ))
                .collect(Collectors.toList());
            
            long momento = System.currentTimeMillis();
            medicoes.forEach(m -> 
                ctx.collect(new LeituraSensor(m.id, momento, m.valor))
            );
            
            Thread.sleep(1000);
        }
    }
    
    @Override
    public void cancel() {
        ativo = false;
    }
}

class LeituraSensor {
    public String id;
    public long timestamp;
    public double temperatura;
    
    public LeituraSensor(String id, long timestamp, double temperatura) {
        this.id = id;
        this.timestamp = timestamp;
        this.temperatura = temperatura;
    }
}

Transformações (Transform)

As transformações convertem dados de uma forma para outra. Cada operador implementa uma interface Function para definir a lógica de processamento:

// Exemplo de transformação com FlatMap
DataFlow<Par<String, Integer>> resultado = fonte
    .flatMap(new SeparadorToken())
    .groupBy(0)
    .sum(1);

class SeparadorToken implements FlatMapFunction<String, Par<String, Integer>> {
    public void flatMap(String linha, Collector<Par<String, Integer>> saida) {
        Arrays.asList(linha.split(" ")).forEach(token -> 
            saida.collect(new Par<>(token, 1))
        );
    }
}

Existem diferentes formas de definir funções:

1. Classe Implementando Function

// Implementação via classe
fonte.map(new TransformadorMaiusculas());

class TransformadorMaiusculas implements MapFunction<String, String> {
    public String map(String entrada) {
        return entrada.toUpperCase();
    }
}

2. Classe Anônima

// Implementação anônima
fonte.map(new MapFunction<String, String>() {
    public String map(String valor) {
        return valor.toUpperCase();
    }
});

3. RichFunction

// RichFunction permite acesso a contexto de runtime
fonte.map(new RichMapFunction<String, String>() {
    public void open(Configuration config) {
        // Inicialização de recursos
    }
    
    public void close() {
        // Limpeza de recursos
    }
    
    public String map(String valor) {
        return valor.toUpperCase();
    }
});

Especificação de Chaves de Partição

1. Por Posição de Campo

// Particionar pelo primeiro campo, agregar segundo campo
fluxo.groupBy(0).max(1);

// Para streams, usar keyBy
fluxo.keyBy(0).sum(1);

2. Por Nome de Campo

// Usar nomes de campos (requer Tuple ou POJO)
DataSource<Pessoa> pessoas = ambiente.fromElements(
    new Pessoa("joao", 25)
);

pessoas.keyBy("nome").sum("idade");

// Equivalente usando Tuplas
fluxo.keyBy("_1").sum("_2");

3. Por Seletor de Chave

// Definir chave via KeySelector
DataSource<Pessoa> dados = ambiente.fromElements(
    new Pessoa("maria", 30)
);

dados.keyBy(new KeySelector<Pessoa, String>() {
    public String getKey(Pessoa p) {
        return p.getNome();
    }
}).sum("idade");

Compreendendo KeyedStream

O KeyedStream representa um stream particionado por chave. Dados com a mesma chave são processados no mesmo分区:

public class ProcessamentoSensor {
    public static void main(String[] args) {
        StreamEnvironment env = StreamEnvironment.getEnvironment();
        
        DataSource<String> arquivo = env.readTextFile("dados/sensores.txt");
        
        DataFlow<LeituraSensor> processado = arquivo
            .map(linha -> {
                String[] partes = linha.split(",");
                return new LeituraSensor(
                    partes[0], 
                    Long.parseLong(partes[1]), 
                    Double.parseDouble(partes[2])
                );
            })
            .keyBy(new ExtratorIdSensor())
            .reduce(new CombinadorLeituras());
        
        processado.output();
        env.run("processamento sensor");
    }
}

class ExtratorIdSensor implements KeySelector<LeituraSensor, String> {
    public String getKey(LeituraSensor leitura) {
        return leitura.id;
    }
}

class CombinadorLeituras implements ReduceFunction<LeituraSensor> {
    public LeituraSensor reduce(LeituraSensor atual, LeituraSensor nova) {
        return new LeituraSensor(
            atual.id,
            Math.max(atual.timestamp, nova.timestamp),
            Math.min(atual.temperatura, nova.temperatura)
        );
    }
}

Split - Divisão de Streams

A operação split permite dividir um DataStream em múltiplos streams baseado em condições:

// Dividir stream por temperatura
SplitStream<LeituraSensor> dividido = fluxo.split(dado -> {
    if (dado.temperatura > 30) {
        return Collections.singletonList("quente");
    } else {
        return Collections.singletonList("frio");
    }
});

// Selecionar streams específicos
DataSource<LeituraSensor> streamQuente = dividido.select("quente");
DataSource<LeituraSensor> streamFrio = dividido.select("frio");
DataSource<LeituraSensor> streamTodos = dividido.select("quente", "frio");

Connect - Conexão de Streams

O método connect junta dois streams mantendo seus tipos originais:

// Streams devem ter tipos diferentes após conexão
ConnectedStreams<Alerta, LeituraSensor> conectado = alertaStream
    .connect(sensorStream);

DataFlow<Object> resultado = conectado.map(
    // Processar stream de alertas
    alerta -> Tupla.of(alerta.mensagem, "ALERTA"),
    // Processar stream de sensores
    sensor -> Tupla.of(sensor.id, "NORMAL")
);

resultado.output();

Union - União de Streams

Union combina múltiplos streams do mesmo tipo em um único:

// Combinar múltiplos streams
DataSource<LeituraSensor> alta = dividido.select("quente");
DataSource<LeituraSensor> baixa = dividido.select("frio");
DataSource<LeituraSensor> ambos = dividido.select("quente", "frio");

// Union de todos os streams
DataSource<LeituraSensor> unificado = alta.union(baixa).union(ambos);

Sink - Saída de Dados

Após as transformações, os dados precisam ser enviados para destinos:

// Exibir no console
fluxo.output();

// Salvar em arquivo
fluxo.writeAsText("file:///caminho/saida");

// Enviar para socket
fluxo.writeToSocket("hostname", porta, esquema);

// Usar sink personalizado
entrada.addSink(StreamingFileSink.forRowFormat(
    new Path("/diretorio/saida"),
    new SimpleStringEncoder<>("UTF-8")
).build());

// Enviar para Kafka
fluxo.addSink(new FlinkKafkaProducer<>(
    "localhost:9092",
    "topico-saida",
    new SimpleStringSchema()
));

Execução do Programa

O método execute() força a construção do grafo de execução e inicia o processamento:

// Executar aplicação de stream
ambienteStream.execute("Minha Aplicação Flink");

// Para DataSet, não é necessário chamar explicitamente
// pois os métodos de saída já incluem a execução

Agrupamento Físico

O Flink oferece várias estratégias de distribuição de dados entre tasks:

• global(): Todos os dados vão para a primeira task
• broadcast(): Dados replicados para todas as tasks
• forward(): Envio direto entre tasks com mesma paralelização
• shuffle(): Distribuição aleatória uniforme
• rebalance(): Distribuição round-robin
• rescale(): Distribuição local entre subconjunto de tasks
• partitionCustom(): Distribuição definida pelo usuário

Sistema de Tipos

O Flink utiliza tipagem forte para otimizar serialização e deserialização. Cada DataStream possui TypeInformation específico, permitindo otimizações internas do engine.

Princípios da API

Quando uma transformação como map é chamada, o Flink cria internamente um OneInputTransformation que contém o operador correspondiente. O operador possui uma Function definida pelo usuário, que contém a lógica de processamento específica.