Apache Storm - Contoh Kerja

Kami telah membahas detail teknis inti dari Apache Storm dan sekarang saatnya untuk membuat kode beberapa skenario sederhana.

Skenario - Penganalisis Log Panggilan Seluler

Panggilan seluler dan durasinya akan diberikan sebagai masukan ke Apache Storm dan Storm akan memproses dan mengelompokkan panggilan antara pemanggil dan penerima yang sama dan jumlah total panggilan mereka.

Pembuatan Cerat

Spout adalah komponen yang digunakan untuk pembuatan data. Pada dasarnya, cerat akan mengimplementasikan antarmuka IRichSpout. Antarmuka "IRichSpout" memiliki metode penting berikut -

• open- Menyediakan cerat dengan lingkungan untuk dieksekusi. Pelaksana akan menjalankan metode ini untuk menginisialisasi cerat.

• nextTuple - Memancarkan data yang dihasilkan melalui kolektor.

• close - Metode ini dipanggil saat cerat akan dimatikan.

• declareOutputFields - Mendeklarasikan skema keluaran tupel.

• ack - Mengakui bahwa tupel tertentu diproses

• fail - Menentukan bahwa tupel tertentu tidak diproses dan tidak akan diproses ulang.

Buka

Tanda tangan dari open metode adalah sebagai berikut -

open(Map conf, TopologyContext context, SpoutOutputCollector collector)

• conf - Menyediakan konfigurasi badai untuk cerat ini.

• context - Memberikan informasi lengkap tentang tempat cerat dalam topologi, id tugasnya, informasi masukan dan keluaran.

• collector - Memungkinkan kami memancarkan tupel yang akan diproses oleh baut.

nextTuple

Tanda tangan dari nextTuple metode adalah sebagai berikut -

nextTuple()

nextTuple () dipanggil secara berkala dari loop yang sama dengan metode ack () dan fail (). Ini harus melepaskan kontrol utas ketika tidak ada pekerjaan yang harus dilakukan, sehingga metode lain memiliki kesempatan untuk dipanggil. Jadi baris pertama pemeriksaan nextTuple untuk melihat apakah pemrosesan telah selesai. Jika demikian, prosesor harus tidur setidaknya satu milidetik untuk mengurangi beban pada prosesor sebelum kembali.

Menutup

Tanda tangan dari close metode adalah sebagai berikut -

close()

mendeklarasikanOutputFields

Tanda tangan dari declareOutputFields metode adalah sebagai berikut -

declareOutputFields(OutputFieldsDeclarer declarer)

declarer - Digunakan untuk mendeklarasikan id aliran keluaran, bidang keluaran, dll.

Metode ini digunakan untuk menentukan skema keluaran tupel.

ack

Tanda tangan dari ack metode adalah sebagai berikut -

ack(Object msgId)

Metode ini mengakui bahwa tupel tertentu telah diproses.

gagal

Tanda tangan dari nextTuple metode adalah sebagai berikut -

ack(Object msgId)

Metode ini menginformasikan bahwa tupel tertentu belum sepenuhnya diproses. Storm akan memproses ulang tupel tertentu.

FakeCallLogReaderSpout

Dalam skenario kami, kami perlu mengumpulkan detail log panggilan. Informasi dari log panggilan berisi.

• nomor pemanggil
• nomor penerima
• duration

Karena, kami tidak memiliki informasi log panggilan waktu nyata, kami akan membuat log panggilan palsu. Informasi palsu akan dibuat menggunakan kelas Acak. Kode program lengkap diberikan di bawah ini.

Coding - FakeCallLogReaderSpout.java

import java.util.*;
//import storm tuple packages
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

//import Spout interface packages
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;

//Create a class FakeLogReaderSpout which implement IRichSpout interface 
   to access functionalities
	
public class FakeCallLogReaderSpout implements IRichSpout {
   //Create instance for SpoutOutputCollector which passes tuples to bolt.
   private SpoutOutputCollector collector;
   private boolean completed = false;
	
   //Create instance for TopologyContext which contains topology data.
   private TopologyContext context;
	
   //Create instance for Random class.
   private Random randomGenerator = new Random();
   private Integer idx = 0;

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

   @Override
   public void nextTuple() {
      if(this.idx <= 1000) {
         List<String> mobileNumbers = new ArrayList<String>();
         mobileNumbers.add("1234123401");
         mobileNumbers.add("1234123402");
         mobileNumbers.add("1234123403");
         mobileNumbers.add("1234123404");

         Integer localIdx = 0;
         while(localIdx++ < 100 && this.idx++ < 1000) {
            String fromMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
            String toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
				
            while(fromMobileNumber == toMobileNumber) {
               toMobileNumber = mobileNumbers.get(randomGenerator.nextInt(4));
            }
				
            Integer duration = randomGenerator.nextInt(60);
            this.collector.emit(new Values(fromMobileNumber, toMobileNumber, duration));
         }
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("from", "to", "duration"));
   }

   //Override all the interface methods
   @Override
   public void close() {}

   public boolean isDistributed() {
      return false;
   }

   @Override
   public void activate() {}

   @Override 
   public void deactivate() {}

   @Override
   public void ack(Object msgId) {}

   @Override
   public void fail(Object msgId) {}

   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

Pembuatan Baut

Bolt merupakan komponen yang mengambil tupel sebagai masukan, memproses tupel, dan menghasilkan tupel baru sebagai keluaran. Baut akan diterapkanIRichBoltantarmuka. Dalam program ini, dua kelas bautCallLogCreatorBolt dan CallLogCounterBolt digunakan untuk melakukan operasi.

Antarmuka IRichBolt memiliki metode berikut -

• prepare- Menyediakan baut dengan lingkungan untuk dieksekusi. Pelaksana akan menjalankan metode ini untuk menginisialisasi cerat.

• execute - Memproses satu tupel input.

• cleanup - Dipanggil saat baut akan dimatikan.

• declareOutputFields - Mendeklarasikan skema keluaran tupel.

Mempersiapkan

Tanda tangan dari prepare metode adalah sebagai berikut -

prepare(Map conf, TopologyContext context, OutputCollector collector)

• conf - Menyediakan konfigurasi Storm untuk baut ini.

• context - Memberikan informasi lengkap tentang tempat baut dalam topologi, id tugasnya, informasi input dan output, dll.

• collector - Memungkinkan kami memancarkan tupel yang diproses.

menjalankan

Tanda tangan dari execute metode adalah sebagai berikut -

execute(Tuple tuple)

Sini tuple adalah tupel masukan yang akan diproses.

Itu executemetode memproses tupel tunggal pada satu waktu. Data tuple dapat diakses dengan metode getValue kelas Tuple. Tidak perlu memproses tupel masukan dengan segera. Beberapa tupel dapat diproses dan dikeluarkan sebagai tupel keluaran tunggal. Tupel yang diproses dapat dipancarkan dengan menggunakan kelas OutputCollector.

membersihkan

Tanda tangan dari cleanup metode adalah sebagai berikut -

cleanup()

mendeklarasikanOutputFields

Tanda tangan dari declareOutputFields metode adalah sebagai berikut -

declareOutputFields(OutputFieldsDeclarer declarer)

Berikut parameternya declarer digunakan untuk mendeklarasikan id aliran keluaran, bidang keluaran, dll.

Metode ini digunakan untuk menentukan skema keluaran tupel

Call log Creator Bolt

Baut pembuat log panggilan menerima tupel log panggilan. Tupel log panggilan memiliki nomor pemanggil, nomor penerima, dan durasi panggilan. Baut ini hanya membuat nilai baru dengan menggabungkan nomor pemanggil dan nomor penerima. Format nilai baru adalah "Nomor pemanggil - Nomor penerima" dan dinamai sebagai kolom baru, "panggilan". Kode lengkap diberikan di bawah ini.

Coding - CallLogCreatorBolt.java

//import util packages
import java.util.HashMap;
import java.util.Map;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;

//import Storm IRichBolt package
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

//Create a class CallLogCreatorBolt which implement IRichBolt interface
public class CallLogCreatorBolt implements IRichBolt {
   //Create instance for OutputCollector which collects and emits tuples to produce output
   private OutputCollector collector;

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

   @Override
   public void execute(Tuple tuple) {
      String from = tuple.getString(0);
      String to = tuple.getString(1);
      Integer duration = tuple.getInteger(2);
      collector.emit(new Values(from + " - " + to, duration));
   }

   @Override
   public void cleanup() {}

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("call", "duration"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
}

Baut Penghitung Log Panggilan

Baut penghitung log panggilan menerima panggilan dan durasinya sebagai tupel. Baut ini menginisialisasi objek kamus (Peta) dalam metode persiapan. Diexecutemetode, ia memeriksa tupel dan membuat entri baru dalam objek kamus untuk setiap nilai "panggilan" baru dalam tupel dan menetapkan nilai 1 dalam objek kamus. Untuk entri yang sudah tersedia di kamus, itu hanya menaikkan nilainya. Secara sederhana, baut ini menyimpan panggilan dan hitungannya dalam objek kamus. Alih-alih menyimpan panggilan dan hitungannya di kamus, kita juga bisa menyimpannya ke sumber data. Kode program lengkapnya adalah sebagai berikut -

Coding - CallLogCounterBolt.java

import java.util.HashMap;
import java.util.Map;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichBolt;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Tuple;

public class CallLogCounterBolt implements IRichBolt {
   Map<String, Integer> counterMap;
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.counterMap = new HashMap<String, Integer>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String call = tuple.getString(0);
      Integer duration = tuple.getInteger(1);
		
      if(!counterMap.containsKey(call)){
         counterMap.put(call, 1);
      }else{
         Integer c = counterMap.get(call) + 1;
         counterMap.put(call, c);
      }
		
      collector.ack(tuple);
   }

   @Override
   public void cleanup() {
      for(Map.Entry<String, Integer> entry:counterMap.entrySet()){
         System.out.println(entry.getKey()+" : " + entry.getValue());
      }
   }

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("call"));
   }
	
   @Override
   public Map<String, Object> getComponentConfiguration() {
      return null;
   }
	
}

Membuat Topologi

Topologi Storm pada dasarnya adalah struktur Hemat. Kelas TopologyBuilder menyediakan metode sederhana dan mudah untuk membuat topologi yang kompleks. Kelas TopologyBuilder memiliki metode untuk mengatur cerat(setSpout) dan untuk memasang baut (setBolt). Akhirnya, TopologyBuilder memiliki createTopology untuk membuat topologi. Gunakan potongan kode berikut untuk membuat topologi -

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
   .shuffleGrouping("call-log-reader-spout");

builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
   .fieldsGrouping("call-log-creator-bolt", new Fields("call"));

shuffleGrouping dan fieldsGrouping Metode membantu mengatur pengelompokan aliran untuk cerat dan baut.

Cluster Lokal

Untuk tujuan pengembangan, kita dapat membuat cluster lokal menggunakan objek "LocalCluster" dan kemudian mengirimkan topologi menggunakan metode "submitTopology" dari kelas "LocalCluster". Salah satu argumen untuk "submitTopology" adalah turunan dari kelas "Config". Kelas "Config" digunakan untuk mengatur opsi konfigurasi sebelum mengirimkan topologi. Opsi konfigurasi ini akan digabungkan dengan konfigurasi cluster pada saat run time dan dikirim ke semua tugas (cerat dan baut) dengan metode persiapan. Setelah topologi dikirimkan ke cluster, kita akan menunggu 10 detik untuk cluster menghitung topologi yang dikirimkan dan kemudian menutup cluster menggunakan metode "shutdown" dari "LocalCluster". Kode program lengkapnya adalah sebagai berikut -

Coding - LogAnalyserStorm.java

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

//import storm configuration packages
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

//Create main class LogAnalyserStorm submit topology.
public class LogAnalyserStorm {
   public static void main(String[] args) throws Exception{
      //Create Config instance for cluster configuration
      Config config = new Config();
      config.setDebug(true);
		
      //
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("call-log-reader-spout", new FakeCallLogReaderSpout());

      builder.setBolt("call-log-creator-bolt", new CallLogCreatorBolt())
         .shuffleGrouping("call-log-reader-spout");

      builder.setBolt("call-log-counter-bolt", new CallLogCounterBolt())
         .fieldsGrouping("call-log-creator-bolt", new Fields("call"));
			
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("LogAnalyserStorm", config, builder.createTopology());
      Thread.sleep(10000);
		
      //Stop the topology
		
      cluster.shutdown();
   }
}

Membangun dan Menjalankan Aplikasi

Aplikasi lengkap memiliki empat kode Java. Mereka adalah -

• FakeCallLogReaderSpout.java
• CallLogCreaterBolt.java
• CallLogCounterBolt.java
• LogAnalyerStorm.java

Aplikasi dapat dibangun menggunakan perintah berikut -

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*” *.java

Aplikasi dapat dijalankan menggunakan perintah berikut -

java -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:. LogAnalyserStorm

Keluaran

Setelah aplikasi dimulai, ini akan menampilkan detail lengkap tentang proses startup cluster, pemrosesan spout dan baut, dan terakhir, proses penutupan cluster. Di "CallLogCounterBolt", kami telah mencetak panggilan dan detail hitungannya. Informasi ini akan ditampilkan di konsol sebagai berikut -

1234123402 - 1234123401 : 78
1234123402 - 1234123404 : 88
1234123402 - 1234123403 : 105
1234123401 - 1234123404 : 74
1234123401 - 1234123403 : 81
1234123401 - 1234123402 : 81
1234123403 - 1234123404 : 86
1234123404 - 1234123401 : 63
1234123404 - 1234123402 : 82
1234123403 - 1234123402 : 83
1234123404 - 1234123403 : 86
1234123403 - 1234123401 : 93

Bahasa non-JVM

Topologi badai diimplementasikan oleh antarmuka Thrift yang memudahkan pengiriman topologi dalam bahasa apa pun. Storm mendukung Ruby, Python, dan banyak bahasa lainnya. Mari kita lihat pengikatan python.

Pengikatan Python

Python adalah bahasa pemrograman yang ditafsirkan untuk tujuan umum, interaktif, berorientasi objek, dan tingkat tinggi. Storm mendukung Python untuk mengimplementasikan topologinya. Python mendukung operasi emitting, anchoring, acking, dan logging.

Seperti yang Anda ketahui, baut dapat didefinisikan dalam bahasa apa pun. Baut yang ditulis dalam bahasa lain dijalankan sebagai sub-proses, dan Storm berkomunikasi dengan sub-proses tersebut dengan pesan JSON melalui stdin / stdout. Pertama ambil contoh WordCount baut yang mendukung pengikatan python.

public static class WordCount implements IRichBolt {
   public WordSplit() {
      super("python", "splitword.py");
   }
	
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("word"));
   }
}

Di sini kelasnya WordCount mengimplementasikan IRichBoltantarmuka dan berjalan dengan implementasi python yang ditentukan argumen metode super "splitword.py". Sekarang buat implementasi python bernama "splitword.py".

import storm
   class WordCountBolt(storm.BasicBolt):
      def process(self, tup):
         words = tup.values[0].split(" ")
         for word in words:
         storm.emit([word])
WordCountBolt().run()

Ini adalah contoh implementasi untuk Python yang menghitung kata-kata dalam kalimat tertentu. Demikian pula, Anda juga dapat mengikat dengan bahasa pendukung lainnya.