El Projecte | Sobre nosaltres | Contribuir | Donacions | Llicència

HOME

Introducció

En aquest projecte, aprendrem a processar dades en temps real utilitzant l'ecosistema Hadoop. El processament de dades en temps real és crucial per a moltes aplicacions modernes, com ara la detecció de fraus, l'anàlisi de xarxes socials i el monitoratge de sistemes. Utilitzarem Apache Flume i Apache Storm per recollir, processar i analitzar dades en temps real.

Objectius del Projecte

Configurar Apache Flume per recollir dades en temps real.
Utilitzar Apache Storm per processar les dades recollides.
Emmagatzemar els resultats processats en HDFS per a una anàlisi posterior.

Requisits Previs

Abans de començar aquest projecte, assegura't de tenir els següents components instal·lats i configurats:

Hadoop
Apache Flume
Apache Storm
Java Development Kit (JDK)

Pas 1: Configuració d'Apache Flume

1.1 Instal·lació d'Apache Flume

Descarrega i instal·la Apache Flume des del lloc oficial: Apache Flume.

1.2 Configuració d'un Agent Flume

Crea un fitxer de configuració per a l'agent Flume. Anomena'l flume-conf.properties i afegeix el següent contingut:

# Define a source, channel, and sink
agent.sources = r1
agent.channels = c1
agent.sinks = k1

# Configure the source
agent.sources.r1.type = netcat
agent.sources.r1.bind = localhost
agent.sources.r1.port = 44444

# Configure the channel
agent.channels.c1.type = memory
agent.channels.c1.capacity = 1000
agent.channels.c1.transactionCapacity = 100

# Configure the sink
agent.sinks.k1.type = hdfs
agent.sinks.k1.hdfs.path = hdfs://localhost:9000/flume/events
agent.sinks.k1.hdfs.fileType = DataStream
agent.sinks.k1.hdfs.writeFormat = Text
agent.sinks.k1.hdfs.batchSize = 1000
agent.sinks.k1.hdfs.rollSize = 0
agent.sinks.k1.hdfs.rollCount = 10000

# Bind the source and sink to the channel
agent.sources.r1.channels = c1
agent.sinks.k1.channel = c1

1.3 Executar l'Agent Flume

Executa l'agent Flume amb la següent comanda:

flume-ng agent --conf ./conf --conf-file flume-conf.properties --name agent -Dflume.root.logger=INFO,console

Pas 2: Configuració d'Apache Storm

2.1 Instal·lació d'Apache Storm

Descarrega i instal·la Apache Storm des del lloc oficial: Apache Storm.

2.2 Creació d'un Topology de Storm

Crea un fitxer Java per definir el Topology de Storm. Anomena'l RealTimeProcessingTopology.java i afegeix el següent contingut:

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.tuple.Fields;
import org.apache.storm.utils.Utils;

public class RealTimeProcessingTopology {
    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();

        // Define the spout
        builder.setSpout("data-spout", new DataSpout(), 1);

        // Define the bolt
        builder.setBolt("process-bolt", new ProcessBolt(), 1)
               .shuffleGrouping("data-spout");

        Config conf = new Config();
        conf.setDebug(true);

        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("real-time-processing", conf, builder.createTopology());

        Utils.sleep(10000);
        cluster.shutdown();
    }
}

2.3 Creació del Spout

Crea un fitxer Java per al Spout. Anomena'l DataSpout.java i afegeix el següent contingut:

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class DataSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void nextTuple() {
        // Simulate data stream
        String data = "example data";
        collector.emit(new Values(data));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("data"));
    }
}

2.4 Creació del Bolt

Crea un fitxer Java per al Bolt. Anomena'l ProcessBolt.java i afegeix el següent contingut:

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.Map;

public class ProcessBolt extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        String data = input.getStringByField("data");
        // Process the data
        System.out.println("Processed data: " + data);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // No output fields
    }
}

2.5 Executar el Topology de Storm

Compila i executa el Topology de Storm amb les següents comandes:

javac -cp ".:path/to/storm/lib/*" RealTimeProcessingTopology.java DataSpout.java ProcessBolt.java
java -cp ".:path/to/storm/lib/*" RealTimeProcessingTopology

Pas 3: Emmagatzematge de Resultats en HDFS

3.1 Configuració del Sink de HDFS

Ja hem configurat el Sink de HDFS en el fitxer flume-conf.properties. Assegura't que el directori hdfs://localhost:9000/flume/events existeixi en HDFS:

hdfs dfs -mkdir -p /flume/events

3.2 Verificació dels Resultats

Després d'executar l'agent Flume i el Topology de Storm, verifica que els resultats processats s'han emmagatzemat en HDFS:

hdfs dfs -ls /flume/events

Conclusió

En aquest projecte, hem après a configurar Apache Flume per recollir dades en temps real, utilitzar Apache Storm per processar aquestes dades i emmagatzemar els resultats en HDFS. Aquest flux de treball és essencial per a moltes aplicacions que requereixen processament de dades en temps real. Practica amb diferents tipus de dades i ajusta els components per adaptar-los a les teves necessitats específiques.

Exercicis Pràctics

Modifica el Spout per recollir dades d'una font externa, com ara una API de xarxes socials.
Afegeix més Bolts al Topology per realitzar diferents tipus de processament de dades.
Configura el Sink de Flume per emmagatzemar els resultats en una base de dades NoSQL com Apache HBase.

Solucions als Exercicis

Exercici 1: Modificar el Spout

@Override
public void nextTuple() {
    // Simulate data stream from an external API
    String data = fetchDataFromAPI();
    collector.emit(new Values(data));
}

private String fetchDataFromAPI() {
    // Implement API call and return data
    return "data from API";
}

Exercici 2: Afegeix més Bolts

builder.setBolt("filter-bolt", new FilterBolt(), 1)
       .shuffleGrouping("data-spout");

builder.setBolt("aggregate-bolt", new AggregateBolt(), 1)
       .shuffleGrouping("filter-bolt");

Exercici 3: Configurar el Sink de Flume per HBase

agent.sinks.k1.type = org.apache.flume.sink.hbase.HBaseSink
agent.sinks.k1.table = my_table
agent.sinks.k1.columnFamily = my_cf
agent.sinks.k1.serializer = org.apache.flume.sink.hbase.SimpleHbaseEventSerializer
agent.sinks.k1.channel = c1

Amb aquests exercicis, podràs aprofundir en el processament de dades en temps real amb Hadoop i adaptar les solucions a les teves necessitats específiques.

Projecte 3: Processament de dades en temps real amb Hadoop