[java] Java를 사용한 Apache Storm의 데이터 분산 처리 도구 개발

16 Nov 2023

java

Apache Storm은 빅데이터 처리를 위한 분산 스트리밍 플랫폼으로, 대규모 데이터를 실시간으로 처리할 수 있는 도구입니다. 이러한 Storm을 사용하여 데이터를 분산처리하는 도구를 개발해보겠습니다.

1. 개발 환경 설정

Apache Storm을 사용하기 위해서는 개발 환경을 설정해야 합니다.

JDK 설치: Java 개발을 위해 JDK를 설치합니다.
Apache Storm 다운로드: Apache Storm의 최신 버전을 다운로드합니다.
환경 변수 설정: JDK와 Storm의 경로를 환경 변수에 추가합니다.

2. Storm Topology 개발

Storm의 핵심 개념은 Topology입니다. Topology는 데이터 처리 흐름을 정의하는 그래프로, Spout와 Bolt로 구성됩니다.

Spout: 데이터 소스로부터 데이터를 읽어옵니다. 예를 들어 Kafka나 RabbitMQ와 같은 메시지 큐에서 데이터를 가져올 수 있습니다.
Bolt: Spout로부터 받은 데이터를 가공하거나 다른 Bolt로 전달합니다. 여러 개의 Bolt를 연결하여 복잡한 데이터 처리 로직을 구현할 수 있습니다.

개발할 Storm Topology의 형태와 각 Bolt와 Spout의 역할을 고민한 후, Java 코드로 구현합니다.

public class DataProcessingTopology {

    public static void main(String[] args) throws Exception {
        // Topology 빌더 생성
        TopologyBuilder builder = new TopologyBuilder();

        // Spout와 Bolt 설정
        builder.setSpout("dataSpout", new DataSpout(), 1);
        builder.setBolt("dataBolt", new DataBolt(), 1).shuffleGrouping("dataSpout");

        // Config 생성
        Config config = new Config();
        config.setDebug(true);

        // Topology 실행
        LocalCluster cluster = new LocalCluster();
        cluster.submitTopology("data-processing-topology", config, builder.createTopology());

        // Topology 실행 시간 설정 (예: 10분)
        Thread.sleep(600000);

        // Topology 종료
        cluster.killTopology("data-processing-topology");
        cluster.shutdown();
    }
}

3. Spout 개발

Spout는 데이터 소스로부터 데이터를 읽어오는 역할을 합니다. 예를 들어 Kafka를 사용하여 데이터를 가져온다면, KafkaSpout를 개발하여 데이터를 읽어올 수 있습니다.

public class DataSpout extends BaseRichSpout {

    private SpoutOutputCollector collector;
    private BufferedReader reader;

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector = collector;
        try {
            // 데이터 소스에서 데이터를 읽어와서 BufferedReader 초기화
            this.reader = new BufferedReader(new FileReader("data.txt"));
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void nextTuple() {
        try {
            // 데이터를 한 줄씩 읽어와서 Bolt로 emit
            String line = reader.readLine();
            if (line != null) {
                collector.emit(new Values(line));
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

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

4. Bolt 개발

Bolt는 Spout로부터 받은 데이터를 가공하거나 다른 Bolt로 전달하는 역할을 합니다. 예를 들어 받은 데이터를 분석하여 특정 조건에 맞는 데이터만 추출한다면, 해당 기능을 구현한 Bolt를 개발합니다.

public class DataBolt extends BaseBasicBolt {

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        // 받은 데이터를 가공하거나 다른 Bolt로 전달하는 로직을 구현
        String data = input.getStringByField("data");
        if (data.contains("important")) {
            collector.emit(new Values(data));
        }
    }

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

5. Topology 실행

모든 개발이 완료되었다면, Topology를 실행하여 데이터 분산 처리를 시작할 수 있습니다. 위에서 작성한 DataProcessingTopology 클래스의 main 메서드를 실행하면 됩니다.

public class DataProcessingTopology {

    public static void main(String[] args) throws Exception {
        // ...
    }
}

결론

이처럼 Apache Storm을 사용하여 데이터를 분산 처리할 수 있는 도구를 Java로 개발하는 방법을 살펴보았습니다. Storm을 활용하여 대규모 실시간 데이터를 처리하는 시스템을 구축하고자 한다면, Topology 개발을 통해 필요한 데이터 처리 로직을 구현해보세요.