Apache Storm - Panduan Cepat

Apa itu Apache Storm?

Apache Storm adalah sistem pemrosesan data besar waktu nyata terdistribusi. Storm dirancang untuk memproses data dalam jumlah besar dalam metode yang toleran terhadap kesalahan dan dapat diskalakan secara horizontal. Ini adalah kerangka data streaming yang memiliki kemampuan tingkat penyerapan tertinggi. Meskipun Storm tidak memiliki kewarganegaraan, ia mengelola lingkungan terdistribusi dan status cluster melalui Apache ZooKeeper. Ini sederhana dan Anda dapat menjalankan semua jenis manipulasi pada data real-time secara paralel.

Apache Storm terus menjadi pemimpin dalam analisis data waktu nyata. Storm mudah diatur, dioperasikan, dan menjamin bahwa setiap pesan akan diproses melalui topologi setidaknya sekali.

Apache Storm vs Hadoop

Pada dasarnya kerangka kerja Hadoop dan Storm digunakan untuk menganalisis data besar. Keduanya saling melengkapi dan berbeda dalam beberapa aspek. Apache Storm melakukan semua operasi kecuali persistensi, sedangkan Hadoop bagus dalam segala hal, tetapi tertinggal dalam komputasi waktu nyata. Tabel berikut membandingkan atribut Storm dan Hadoop.

Badai Hadoop
Pemrosesan aliran waktu nyata Pemrosesan batch
Tanpa kewarganegaraan Stateful
Arsitektur Master / Slave dengan koordinasi berbasis Zookeeper. Node master disebut sebagainimbus dan budak adalah supervisors. Arsitektur master-slave dengan / tanpa koordinasi berbasis ZooKeeper. Node master adalahjob tracker dan simpul budak adalah task tracker.
Proses streaming Storm dapat mengakses puluhan ribu pesan per detik di cluster. Hadoop Distributed File System (HDFS) menggunakan kerangka kerja MapReduce untuk memproses data dalam jumlah besar yang memerlukan beberapa menit atau jam.
Topologi Storm berjalan hingga pengguna dimatikan atau terjadi kegagalan tak terduga yang tidak dapat dipulihkan Pekerjaan MapReduce dijalankan dalam urutan berurutan dan akhirnya diselesaikan.
Both are distributed and fault-tolerant
Jika nimbus / supervisor mati, memulai ulang akan melanjutkan dari tempat berhenti, maka tidak ada yang terpengaruh. Jika JobTracker mati, semua pekerjaan yang berjalan hilang.

Kasus Penggunaan Apache Storm

Apache Storm sangat terkenal dengan pemrosesan aliran data besar secara real-time. Untuk alasan ini, sebagian besar perusahaan menggunakan Storm sebagai bagian integral dari sistem mereka. Beberapa contoh penting adalah sebagai berikut -

Twitter- Twitter menggunakan Apache Storm untuk rangkaian "produk Analisis Penerbit". "Produk Analisis Penerbit" memproses setiap tweet dan klik di Platform Twitter. Apache Storm sangat terintegrasi dengan infrastruktur Twitter.

NaviSite- NaviSite menggunakan Storm untuk sistem pemantauan / audit log Peristiwa. Setiap log yang dihasilkan di sistem akan melalui Storm. Storm akan memeriksa pesan terhadap kumpulan ekspresi reguler yang dikonfigurasi dan jika ada yang cocok, maka pesan tersebut akan disimpan ke database.

Wego- Wego adalah mesin pencari perjalanan yang berlokasi di Singapura. Data terkait perjalanan berasal dari banyak sumber di seluruh dunia dengan waktu yang berbeda. Storm membantu Wego mencari data real-time, menyelesaikan masalah konkurensi dan menemukan kecocokan terbaik untuk pengguna akhir.

Manfaat Apache Storm

Berikut adalah daftar manfaat yang ditawarkan Apache Storm -

  • Storm adalah open source, kuat, dan ramah pengguna. Ini bisa digunakan di perusahaan kecil maupun perusahaan besar.

  • Storm toleran terhadap kesalahan, fleksibel, andal, dan mendukung bahasa pemrograman apa pun.

  • Memungkinkan pemrosesan streaming waktu nyata.

  • Storm luar biasa cepat karena memiliki kekuatan pemrosesan data yang sangat besar.

  • Storm dapat mempertahankan kinerja bahkan di bawah beban yang meningkat dengan menambahkan sumber daya secara linier. Ini sangat skalabel.

  • Storm melakukan penyegaran data dan respons pengiriman ujung ke ujung dalam hitungan detik atau menit tergantung pada masalahnya. Ini memiliki latensi yang sangat rendah.

  • Storm memiliki kecerdasan operasional.

  • Storm menyediakan pemrosesan data yang terjamin bahkan jika salah satu node yang terhubung di cluster mati atau pesan hilang.

Apache Storm membaca aliran mentah data real-time dari satu ujung dan meneruskannya melalui serangkaian unit pemrosesan kecil dan mengeluarkan informasi yang diproses / berguna di ujung lainnya.

Diagram berikut menggambarkan konsep inti Apache Storm.

Sekarang mari kita lihat lebih dekat komponen Apache Storm -

Komponen Deskripsi
Tuple Tuple adalah struktur data utama di Storm. Ini adalah daftar elemen yang dipesan. Secara default, Tuple mendukung semua tipe data. Umumnya, ini dimodelkan sebagai sekumpulan nilai yang dipisahkan koma dan diteruskan ke cluster Storm.
Aliran Stream adalah urutan tupel yang tidak berurutan.
Cerat Sumber aliran. Umumnya, Storm menerima data masukan dari sumber data mentah seperti Twitter Streaming API, antrean Apache Kafka, antrean Kestrel, dll. Jika tidak, Anda dapat menulis spout untuk membaca data dari sumber data. "ISpout" adalah antarmuka inti untuk mengimplementasikan spouts. Beberapa antarmuka spesifik adalah IRichSpout, BaseRichSpout, KafkaSpout, dll.
Baut Baut adalah unit pemrosesan logis. Cerat melewatkan data ke proses baut dan baut dan menghasilkan aliran keluaran baru. Baut dapat melakukan operasi pemfilteran, agregasi, penggabungan, interaksi dengan sumber data dan database. Bolt menerima data dan memancarkan ke satu atau lebih baut. "IBolt" adalah antarmuka inti untuk mengimplementasikan baut. Beberapa antarmuka umum adalah IRichBolt, IBasicBolt, dll.

Mari kita ambil contoh real-time dari "Analisis Twitter" dan lihat bagaimana hal itu dapat dimodelkan di Apache Storm. Diagram berikut menggambarkan strukturnya.

Masukan untuk "Analisis Twitter" berasal dari Twitter Streaming API. Spout akan membaca tweet pengguna menggunakan Twitter Streaming API dan output sebagai aliran tuple. Satu tupel dari cerat akan memiliki nama pengguna twitter dan satu tweet sebagai nilai yang dipisahkan koma. Kemudian, uap tupel ini akan diteruskan ke Bolt dan Bolt akan membagi tweet menjadi kata individu, menghitung jumlah kata, dan menyimpan informasi ke sumber data yang dikonfigurasi. Sekarang, kita bisa dengan mudah mendapatkan hasilnya dengan menanyakan sumber data.

Topologi

Spout dan baut dihubungkan bersama dan membentuk topologi. Logika aplikasi waktu nyata ditentukan di dalam topologi Storm. Dengan kata sederhana, topologi adalah grafik berarah di mana simpul adalah komputasi dan tepi adalah aliran data.

Topologi sederhana dimulai dengan cerat. Cerat memancarkan data ke satu atau lebih baut. Baut mewakili simpul dalam topologi yang memiliki logika pemrosesan terkecil dan keluaran dari sebuah baut dapat dipancarkan ke baut lain sebagai masukan.

Storm membuat topologi selalu berjalan, sampai Anda mematikan topologi tersebut. Tugas utama Apache Storm adalah menjalankan topologi dan menjalankan sejumlah topologi pada waktu tertentu.

Tugas

Sekarang Anda memiliki ide dasar tentang cerat dan baut. Mereka adalah unit logis terkecil dari topologi dan topologi dibuat dengan menggunakan cerat tunggal dan serangkaian baut. Mereka harus dijalankan dengan benar dalam urutan tertentu agar topologi dapat berjalan dengan sukses. Eksekusi setiap cerat dan baut oleh Storm disebut sebagai "Tasks". Dengan kata sederhana, tugas adalah eksekusi cerat atau baut. Pada waktu tertentu, setiap cerat dan baut dapat memiliki beberapa contoh yang berjalan di beberapa utas terpisah.

Pekerja

Topologi berjalan secara terdistribusi, di beberapa node pekerja. Storm menyebarkan tugas secara merata di semua node pekerja. Peran node pekerja adalah mendengarkan pekerjaan dan memulai atau menghentikan proses setiap kali ada pekerjaan baru.

Pengelompokan Aliran

Aliran data mengalir dari cerat ke baut atau dari satu baut ke baut lainnya. Pengelompokan aliran mengontrol bagaimana tupel dirutekan dalam topologi dan membantu kita memahami aliran tupel dalam topologi. Ada empat pengelompokan bawaan seperti yang dijelaskan di bawah ini.

Pengelompokan Acak

Dalam pengelompokan acak, jumlah tupel yang sama didistribusikan secara acak ke semua pekerja yang menjalankan baut. Diagram berikut menggambarkan strukturnya.

Pengelompokan Lapangan

Bidang dengan nilai yang sama dalam tupel dikelompokkan bersama dan tupel yang tersisa disimpan di luar. Kemudian, tupel dengan nilai bidang yang sama dikirim ke pekerja yang sama yang menjalankan baut. Misalnya, jika aliran dikelompokkan berdasarkan bidang "kata", maka tupel dengan string yang sama, "Halo" akan pindah ke pekerja yang sama. Diagram berikut memperlihatkan cara kerja Pengelompokan Bidang.

Pengelompokan Global

Semua aliran dapat dikelompokkan dan diteruskan ke satu baut. Pengelompokan ini mengirimkan tupel yang dihasilkan oleh semua instance sumber ke satu instance target (khususnya, pilih pekerja dengan ID terendah).

Semua Pengelompokan

Semua Pengelompokan mengirimkan satu salinan dari setiap tupel ke semua contoh baut penerima. Pengelompokan semacam ini digunakan untuk mengirim sinyal ke baut. Semua pengelompokan berguna untuk operasi gabungan.

Salah satu sorotan utama dari Apache Storm adalah ia toleran terhadap kesalahan, cepat tanpa aplikasi terdistribusi "Titik Kegagalan Tunggal" (SPOF). Kita dapat menginstal Apache Storm di banyak sistem yang diperlukan untuk meningkatkan kapasitas aplikasi.

Mari kita lihat bagaimana cluster Apache Storm dirancang dan arsitektur internalnya. Diagram berikut menggambarkan desain cluster.

Apache Storm memiliki dua jenis node, Nimbus (node ​​master) dan Supervisor(node ​​pekerja). Nimbus adalah komponen utama dari Apache Storm. Tugas utama Nimbus adalah menjalankan topologi Storm. Nimbus menganalisis topologi dan mengumpulkan tugas yang akan dijalankan. Kemudian, itu akan mendistribusikan tugas ke supervisor yang tersedia.

Seorang supervisor akan memiliki satu atau lebih proses pekerja. Supervisor akan mendelegasikan tugas ke proses pekerja. Proses pekerja akan menelurkan pelaksana sebanyak yang dibutuhkan dan menjalankan tugas. Apache Storm menggunakan sistem pesan terdistribusi internal untuk komunikasi antara nimbus dan supervisor.

Komponen Deskripsi
Nimbus Nimbus adalah node master dari cluster Storm. Semua node lain dalam cluster disebut sebagaiworker nodes. Node master bertanggung jawab untuk mendistribusikan data di antara semua node pekerja, menetapkan tugas ke node pekerja dan memantau kegagalan.
Pengawas Node yang mengikuti instruksi yang diberikan oleh nimbus disebut sebagai Supervisor. SEBUAHsupervisor memiliki banyak proses pekerja dan mengatur proses pekerja untuk menyelesaikan tugas yang diberikan oleh nimbus.
Proses pekerja Proses pekerja akan menjalankan tugas yang terkait dengan topologi tertentu. Proses pekerja tidak akan menjalankan tugas dengan sendirinya, melainkan membuatexecutorsdan meminta mereka untuk melakukan tugas tertentu. Sebuah proses pekerja akan memiliki banyak pelaksana.
Pelaksana Eksekutor tidak lain adalah satu utas yang muncul oleh proses pekerja. Pelaksana menjalankan satu atau lebih tugas tetapi hanya untuk cerat atau baut tertentu.
Tugas Sebuah tugas melakukan pemrosesan data aktual. Jadi, itu bisa berupa cerat atau baut.
Kerangka ZooKeeper

Apache ZooKeeper adalah layanan yang digunakan oleh cluster (sekelompok node) untuk berkoordinasi di antara mereka sendiri dan memelihara data bersama dengan teknik sinkronisasi yang kuat. Nimbus tidak memiliki kewarganegaraan, jadi bergantung pada Zookeeper untuk memantau status node yang berfungsi.

ZooKeeper membantu supervisor untuk berinteraksi dengan nimbus. Bertanggung jawab untuk menjaga status nimbus dan supervisor.

Badai tidak memiliki kewarganegaraan. Meskipun sifat stateless memiliki kekurangannya sendiri, itu sebenarnya membantu Storm memproses data real-time dengan cara terbaik dan tercepat.

Badai tidak sepenuhnya tanpa kewarganegaraan. Ini menyimpan statusnya di Apache ZooKeeper. Karena status tersedia di Apache ZooKeeper, nimbus yang gagal dapat dimulai ulang dan berfungsi dari tempatnya. Biasanya, alat pemantauan layanan sepertimonit akan memantau Nimbus dan memulai ulang jika ada kegagalan.

Apache Storm juga memiliki topologi lanjutan yang disebut Trident Topologydengan pemeliharaan negara dan juga menyediakan API tingkat tinggi seperti Pig. Kami akan membahas semua fitur ini di bab-bab selanjutnya.

Cluster Storm yang berfungsi harus memiliki satu nimbus dan satu atau lebih supervisor. Node penting lainnya adalah Apache ZooKeeper, yang akan digunakan untuk koordinasi antara nimbus dan pengawas.

Mari kita lihat lebih dekat alur kerja Apache Storm -

  • Awalnya, nimbus akan menunggu sampai "Topologi Badai" dikirimkan ke sana.

  • Setelah topologi dikirim, itu akan memproses topologi dan mengumpulkan semua tugas yang akan dilakukan dan urutan tugas yang akan dijalankan.

  • Kemudian, nimbus akan mendistribusikan tugas secara merata kepada semua pengawas yang ada.

  • Pada interval waktu tertentu, semua pengawas akan mengirimkan detak jantung ke nimbus untuk menginformasikan bahwa mereka masih hidup.

  • Ketika seorang supervisor meninggal dan tidak mengirimkan detak jantung ke nimbus, maka nimbus memberikan tugas tersebut kepada supervisor lainnya.

  • Saat nimbus itu sendiri mati, supervisor akan mengerjakan tugas yang telah diberikan tanpa masalah apa pun.

  • Setelah semua tugas selesai, supervisor akan menunggu tugas baru masuk.

  • Sementara itu, nimbus yang mati akan dimulai ulang secara otomatis oleh alat pemantauan layanan.

  • Nimbus yang dimulai ulang akan melanjutkan dari tempatnya berhenti. Demikian pula, supervisor yang sudah mati juga dapat dimulai ulang secara otomatis. Karena nimbus dan supervisor dapat dimulai ulang secara otomatis dan keduanya akan berlanjut seperti sebelumnya, Storm dijamin akan memproses semua tugas setidaknya sekali.

  • Setelah semua topologi diproses, nimbus menunggu topologi baru tiba dan demikian pula pengawas menunggu tugas baru.

Secara default, ada dua mode dalam cluster Storm -

  • Local mode- Mode ini digunakan untuk pengembangan, pengujian, dan debugging karena ini adalah cara termudah untuk melihat semua komponen topologi bekerja bersama. Dalam mode ini, kita dapat menyesuaikan parameter yang memungkinkan kita untuk melihat bagaimana topologi kita berjalan di lingkungan konfigurasi Storm yang berbeda. Dalam mode Lokal, topologi badai berjalan di mesin lokal dalam JVM tunggal.

  • Production mode- Dalam mode ini, kami mengirimkan topologi kami ke cluster working storm, yang terdiri dari banyak proses, biasanya berjalan pada mesin yang berbeda. Seperti yang dibahas dalam alur kerja badai, cluster kerja akan berjalan tanpa batas hingga ditutup.

Apache Storm memproses data waktu nyata dan masukan biasanya berasal dari sistem antrian pesan. Sistem pesan terdistribusi eksternal akan memberikan masukan yang diperlukan untuk komputasi waktu nyata. Spout akan membaca data dari sistem pesan dan mengubahnya menjadi tupel dan memasukkannya ke dalam Apache Storm. Fakta yang menarik adalah bahwa Apache Storm menggunakan sistem pesan terdistribusi sendiri secara internal untuk komunikasi antara nimbus dan supervisornya.

Apa itu Sistem Pesan Terdistribusi?

Pesan terdistribusi didasarkan pada konsep antrian pesan yang andal. Pesan diantrekan secara asinkron antara aplikasi klien dan sistem perpesanan. Sistem perpesanan terdistribusi memberikan manfaat keandalan, skalabilitas, dan persistensi.

Sebagian besar pola pesan mengikuti publish-subscribe model (sederhana Pub-Sub) tempat pengirim pesan dipanggil publishers dan mereka yang ingin menerima pesan dipanggil subscribers.

Setelah pesan dipublikasikan oleh pengirim, pelanggan dapat menerima pesan yang dipilih dengan bantuan opsi pemfilteran. Biasanya kami memiliki dua jenis pemfilteran, satu adalahtopic-based filtering dan satu lagi content-based filtering.

Perhatikan bahwa model pub-sub hanya dapat berkomunikasi melalui pesan. Ini adalah arsitektur yang sangat longgar; bahkan pengirimnya tidak tahu siapa pelanggannya. Banyak dari pola pesan memungkinkan dengan broker pesan untuk bertukar mempublikasikan pesan untuk akses tepat waktu oleh banyak pelanggan. Contoh kehidupan nyata adalah Dish TV, yang menerbitkan saluran berbeda seperti olahraga, film, musik, dll., Dan siapa pun dapat berlangganan saluran mereka sendiri dan mendapatkannya setiap kali saluran langganan mereka tersedia.

Tabel berikut menjelaskan beberapa sistem pesan throughput tinggi yang populer -

Sistem pesan terdistribusi Deskripsi
Apache Kafka Kafka dikembangkan di perusahaan LinkedIn dan kemudian menjadi sub-proyek Apache. Apache Kafka didasarkan pada model publikasi-langganan yang diaktifkan perantara, persisten, dan terdistribusi. Kafka cepat, terukur, dan sangat efisien.
RabbitMQ RabbitMQ adalah aplikasi perpesanan tangguh terdistribusi terbuka. Mudah digunakan dan dijalankan di semua platform.
JMS (Layanan Pesan Java) JMS adalah API sumber terbuka yang mendukung pembuatan, pembacaan, dan pengiriman pesan dari satu aplikasi ke aplikasi lainnya. Ini memberikan pengiriman pesan yang terjamin dan mengikuti model publish-subscribe.
ActiveMQ Sistem pesan ActiveMQ adalah API JMS open source.
ZeroMQ ZeroMQ adalah pemrosesan pesan peer-peer tanpa broker. Ini menyediakan pola pesan dorong-tarik, router-dealer.
Kestrel Kestrel adalah antrian pesan terdistribusi yang cepat, andal, dan sederhana.

Protokol Hemat

Penghematan dibuat di Facebook untuk pengembangan layanan lintas bahasa dan panggilan prosedur jarak jauh (RPC). Kemudian, ini menjadi proyek Apache open source. Apache Thrift adalah fileInterface Definition Language dan memungkinkan untuk menentukan tipe data baru dan implementasi layanan di atas tipe data yang ditentukan dengan cara yang mudah.

Apache Thrift juga merupakan kerangka kerja komunikasi yang mendukung sistem tertanam, aplikasi seluler, aplikasi web, dan banyak bahasa pemrograman lainnya. Beberapa fitur utama yang terkait dengan Apache Thrift adalah modularitas, fleksibilitas, dan performa tinggi. Selain itu, dapat melakukan streaming, perpesanan, dan RPC dalam aplikasi terdistribusi.

Storm secara ekstensif menggunakan Thrift Protocol untuk komunikasi internal dan definisi datanya. Topologi badai sederhana sajaThrift Structs. Storm Nimbus yang menjalankan topologi di Apache Storm adalah aThrift service.

Sekarang mari kita lihat cara menginstal framework Apache Storm di komputer Anda. Ada tiga langkah majo di sini -

  • Instal Java di sistem Anda, jika Anda belum memilikinya.
  • Instal kerangka ZooKeeper.
  • Instal kerangka Apache Storm.

Langkah 1 - Memverifikasi Instalasi Java

Gunakan perintah berikut untuk memeriksa apakah Anda telah menginstal Java di sistem Anda.

$ java -version

Jika Java sudah ada, maka Anda akan melihat nomor versinya. Atau, unduh JDK versi terbaru.

Langkah 1.1 - Unduh JDK

Unduh versi terbaru JDK dengan menggunakan tautan berikut - www.oracle.com

Versi terbaru adalah JDK 8u 60 dan filenya adalah “jdk-8u60-linux-x64.tar.gz”. Unduh file di komputer Anda.

Langkah 1.2 - Ekstrak file

Umumnya file sedang diunduh ke file downloadsmap. Ekstrak pengaturan tar menggunakan perintah berikut.

$ cd /go/to/download/path
$ tar -zxf jdk-8u60-linux-x64.gz

Langkah 1.3 - Pindah ke direktori opt

Agar Java tersedia untuk semua pengguna, pindahkan konten java yang diekstrak ke folder "/ usr / local / java".

$ su
password: (type password of root user)
$ mkdir /opt/jdk
$ mv jdk-1.8.0_60 /opt/jdk/

Langkah 1.4 - Tetapkan jalur

Untuk mengatur path dan variabel JAVA_HOME, tambahkan perintah berikut ke file ~ / .bashrc.

export JAVA_HOME =/usr/jdk/jdk-1.8.0_60
export PATH=$PATH:$JAVA_HOME/bin

Sekarang terapkan semua perubahan ke sistem yang sedang berjalan.

$ source ~/.bashrc

Langkah 1.5 - Alternatif Java

Gunakan perintah berikut untuk mengubah alternatif Java.

update-alternatives --install /usr/bin/java java /opt/jdk/jdk1.8.0_60/bin/java 100

Langkah 1.6

Sekarang verifikasi instalasi Java menggunakan perintah verifikasi (java -version) dijelaskan pada Langkah 1.

Langkah 2 - Instalasi Kerangka ZooKeeper

Langkah 2.1 - Unduh ZooKeeper

Untuk memasang kerangka kerja Zookeeper di komputer Anda, kunjungi tautan berikut dan unduh versi terbaru Zookeeper http://zookeeper.apache.org/releases.html

Saat ini, versi terbaru ZooKeeper adalah 3.4.6 (ZooKeeper-3.4.6.tar.gz).

Langkah 2.2 - Ekstrak file tar

Ekstrak file tar menggunakan perintah berikut -

$ cd opt/
$ tar -zxf zookeeper-3.4.6.tar.gz
$ cd zookeeper-3.4.6
$ mkdir data

Langkah 2.3 - Buat file konfigurasi

Buka file konfigurasi bernama “conf / zoo.cfg” menggunakan perintah "vi conf / zoo.cfg" dan atur semua parameter berikut sebagai titik awal.

$ vi conf/zoo.cfg
tickTime=2000
dataDir=/path/to/zookeeper/data
clientPort=2181
initLimit=5
syncLimit=2

Setelah file konfigurasi berhasil disimpan, Anda dapat memulai server Zookeeper.

Langkah 2.4 - Mulai Server ZooKeeper

Gunakan perintah berikut untuk memulai server Zookeeper.

$ bin/zkServer.sh start

Setelah menjalankan perintah ini, Anda akan mendapatkan respons sebagai berikut -

$ JMX enabled by default
$ Using config: /Users/../zookeeper-3.4.6/bin/../conf/zoo.cfg
$ Starting zookeeper ... STARTED

Langkah 2.5 - Mulai CLI

Gunakan perintah berikut untuk memulai CLI.

$ bin/zkCli.sh

Setelah menjalankan perintah di atas, Anda akan terhubung ke server Zookeeper dan mendapatkan respons berikut.

Connecting to localhost:2181
................
................
................
Welcome to ZooKeeper!
................
................
WATCHER::
WatchedEvent state:SyncConnected type: None path:null
[zk: localhost:2181(CONNECTED) 0]

Langkah 2.6 - Hentikan Server ZooKeeper

Setelah menghubungkan server dan melakukan semua operasi, Anda dapat menghentikan server Zookeeper dengan menggunakan perintah berikut.

bin/zkServer.sh stop

Anda telah berhasil menginstal Java dan Zookeeper di komputer Anda. Sekarang mari kita lihat langkah-langkah untuk menginstal framework Apache Storm.

Langkah 3 - Instalasi Kerangka Apache Storm

Langkah 3.1 Unduh Storm

Untuk menginstal kerangka kerja Storm di komputer Anda, kunjungi tautan berikut dan unduh versi terbaru Storm http://storm.apache.org/downloads.html

Untuk saat ini, versi terbaru dari Storm adalah “apache-storm-0.9.5.tar.gz”.

Langkah 3.2 - Ekstrak file tar

Ekstrak file tar menggunakan perintah berikut -

$ cd opt/
$ tar -zxf apache-storm-0.9.5.tar.gz
$ cd apache-storm-0.9.5
$ mkdir data

Langkah 3.3 - Buka file konfigurasi

Rilis Storm saat ini berisi file di “conf / storm.yaml” yang mengkonfigurasi daemon Storm. Tambahkan informasi berikut ke file itu.

$ vi conf/storm.yaml
storm.zookeeper.servers:
 - "localhost"
storm.local.dir: “/path/to/storm/data(any path)”
nimbus.host: "localhost"
supervisor.slots.ports:
 - 6700
 - 6701
 - 6702
 - 6703

Setelah menerapkan semua perubahan, simpan dan kembali ke terminal.

Langkah 3.4 - Mulai Nimbus

$ bin/storm nimbus

Langkah 3.5 - Mulai Supervisor

$ bin/storm supervisor

Langkah 3.6 Mulai UI

$ bin/storm ui

Setelah memulai aplikasi antarmuka pengguna Storm, ketik URL-nya http://localhost:8080di browser favorit Anda dan Anda dapat melihat informasi cluster Storm dan topologi yang sedang berjalan. Halaman tersebut akan terlihat seperti screenshot berikut.

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 serta 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 mengembalikannya.

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, kami 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.

Trident adalah perpanjangan dari Storm. Seperti Storm, Trident juga dikembangkan oleh Twitter. Alasan utama di balik pengembangan Trident adalah untuk menyediakan abstraksi tingkat tinggi di atas Storm bersama dengan pemrosesan aliran stateful dan kueri terdistribusi latensi rendah.

Trident menggunakan cerat dan baut, tetapi komponen tingkat rendah ini dibuat secara otomatis oleh Trident sebelum dieksekusi. Trident memiliki fungsi, filter, gabungan, pengelompokan, dan agregasi.

Trident memproses aliran sebagai serangkaian batch yang disebut transaksi. Umumnya ukuran batch kecil itu akan berada di urutan ribuan atau jutaan tupel, tergantung pada aliran input. Dengan cara ini, Trident berbeda dari Storm, yang melakukan pemrosesan tuple-by-tuple.

Konsep pemrosesan batch sangat mirip dengan transaksi database. Setiap transaksi diberi ID transaksi. Transaksi dianggap berhasil, setelah semua prosesnya selesai. Namun, kegagalan dalam memproses salah satu tupel transaksi akan menyebabkan seluruh transaksi dikirim ulang. Untuk setiap batch, Trident akan memanggil beginCommit di awal transaksi, dan berkomitmen di akhir transaksi.

Topologi Trisula

Trident API menampilkan opsi mudah untuk membuat topologi Trident menggunakan kelas “TridentTopology”. Pada dasarnya, topologi Trident menerima aliran input dari cerat dan melakukan urutan operasi yang teratur (filter, agregasi, pengelompokan, dll.,) Di aliran. Storm Tuple digantikan oleh Trident Tuple dan Bolts diganti dengan pengoperasian. Topologi Trident sederhana dapat dibuat sebagai berikut -

TridentTopology topology = new TridentTopology();

Trident Tuple

Trident tuple adalah daftar nilai yang diberi nama. Antarmuka TridentTuple adalah model data dari topologi Trident. Antarmuka TridentTuple adalah unit dasar data yang dapat diproses oleh topologi Trident.

Trident Spout

Trident spout mirip dengan Storm spout, dengan opsi tambahan untuk menggunakan fitur Trident. Sebenarnya kita masih bisa menggunakan IRichSpout yang sudah kita pakai di topologi Storm, tapi sifatnya non-transaksional dan kita tidak akan bisa menggunakan kelebihan yang diberikan oleh Trident.

Cerat dasar yang memiliki semua fungsi untuk menggunakan fitur Trident adalah "ITridentSpout". Ini mendukung semantik transaksional transaksional dan buram. Spout lainnya adalah IBatchSpout, IPartitionedTridentSpout, dan IOpaquePartitionedTridentSpout.

Selain cerat generik ini, Trident memiliki banyak contoh penerapan cerat trisula. Salah satunya adalah FeederBatchSpout spout, yang dapat kita gunakan untuk mengirim daftar tupel trident dengan mudah tanpa perlu khawatir tentang pemrosesan batch, paralelisme, dll.

Pembuatan FeederBatchSpout dan pengumpanan data dapat dilakukan seperti yang ditunjukkan di bawah ini -

TridentTopology topology = new TridentTopology();
FeederBatchSpout testSpout = new FeederBatchSpout(
   ImmutableList.of("fromMobileNumber", "toMobileNumber", “duration”));
topology.newStream("fixed-batch-spout", testSpout)
testSpout.feed(ImmutableList.of(new Values("1234123401", "1234123402", 20)));

Operasi Trident

Trident mengandalkan "Operasi Trident" untuk memproses aliran input tupel trisula. Trident API memiliki sejumlah operasi built-in untuk menangani pemrosesan streaming yang sederhana hingga kompleks. Operasi ini berkisar dari validasi sederhana hingga pengelompokan kompleks dan agregasi tupel trisula. Mari kita bahas operasi yang paling penting dan sering digunakan.

Saring

Filter adalah sebuah objek yang digunakan untuk melakukan tugas validasi input. Filter Trident mendapatkan subset dari bidang tupel trisula sebagai masukan dan menampilkan benar atau salah bergantung pada apakah kondisi tertentu terpenuhi atau tidak. Jika true dikembalikan, maka tupel disimpan dalam aliran keluaran; jika tidak, tupel akan dihapus dari aliran. Filter pada dasarnya akan mewarisi dariBaseFilter kelas dan menerapkan isKeepmetode. Berikut adalah contoh implementasi operasi filter -

public class MyFilter extends BaseFilter {
   public boolean isKeep(TridentTuple tuple) {
      return tuple.getInteger(1) % 2 == 0;
   }
}

input

[1, 2]
[1, 3]
[1, 4]

output

[1, 2]
[1, 4]

Fungsi filter dapat dipanggil dalam topologi menggunakan metode “each”. Kelas "Fields" dapat digunakan untuk menentukan input (bagian dari tupel trisula). Kode sampelnya adalah sebagai berikut -

TridentTopology topology = new TridentTopology();
topology.newStream("spout", spout)
.each(new Fields("a", "b"), new MyFilter())

Fungsi

Functionadalah objek yang digunakan untuk melakukan operasi sederhana pada tupel trisula tunggal. Dibutuhkan subset bidang tupel trisula dan memancarkan nol atau lebih bidang tupel trisula baru.

Function pada dasarnya mewarisi dari BaseFunction kelas dan mengimplementasikan executemetode. Contoh implementasi diberikan di bawah ini -

public class MyFunction extends BaseFunction {
   public void execute(TridentTuple tuple, TridentCollector collector) {
      int a = tuple.getInteger(0);
      int b = tuple.getInteger(1);
      collector.emit(new Values(a + b));
   }
}

input

[1, 2]
[1, 3]
[1, 4]

output

[1, 2, 3]
[1, 3, 4]
[1, 4, 5]

Sama seperti operasi Filter, operasi Fungsi dapat dipanggil dalam topologi menggunakan eachmetode. Kode sampelnya adalah sebagai berikut -

TridentTopology topology = new TridentTopology();
topology.newStream("spout", spout)
   .each(new Fields(“a, b"), new MyFunction(), new Fields(“d")));

Pengumpulan

Agregasi adalah objek yang digunakan untuk melakukan operasi agregasi pada batch input atau partisi atau aliran. Trisula memiliki tiga jenis agregasi. Mereka adalah sebagai berikut -

  • aggregate- Mengumpulkan setiap batch tupel trisula secara terpisah. Selama proses agregat, tupel awalnya dipartisi ulang menggunakan pengelompokan global untuk menggabungkan semua partisi dari batch yang sama ke dalam satu partisi.

  • partitionAggregate- Mengumpulkan setiap partisi, bukan seluruh batch tupel trisula. Keluaran agregat partisi sepenuhnya menggantikan tupel masukan. Keluaran agregat partisi berisi tupel bidang tunggal.

  • persistentaggregate - Agregat pada semua tupel trisula di semua batch dan menyimpan hasilnya di memori atau database.

TridentTopology topology = new TridentTopology();

// aggregate operation
topology.newStream("spout", spout)
   .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
   .aggregate(new Count(), new Fields(“count”))
	
// partitionAggregate operation
topology.newStream("spout", spout)
   .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
   .partitionAggregate(new Count(), new Fields(“count"))
	
// persistentAggregate - saving the count to memory
topology.newStream("spout", spout)
   .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
   .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"));

Operasi agregasi dapat dibuat menggunakan antarmuka CombinerAggregator, ReducerAggregator, atau generik Agregator. Agregator "count" yang digunakan dalam contoh di atas adalah salah satu agregator build-in. Agregator ini diterapkan menggunakan "CombinerAggregator". Penerapannya adalah sebagai berikut -

public class Count implements CombinerAggregator<Long> {
   @Override
   public Long init(TridentTuple tuple) {
      return 1L;
   }
	
   @Override
   public Long combine(Long val1, Long val2) {
      return val1 + val2;
   }
	
   @Override
   public Long zero() {
      return 0L;
   }
}

Pengelompokan

Operasi pengelompokan adalah operasi bawaan dan dapat dipanggil oleh groupBymetode. Metode groupBy mempartisi ulang aliran dengan melakukan partisiBy pada bidang yang ditentukan, dan kemudian dalam setiap partisi, ia mengelompokkan tupel bersama yang bidang kelompoknya sama. Biasanya, kami menggunakan "groupBy" bersama dengan "persistentAggregate" untuk mendapatkan agregasi yang dikelompokkan. Kode sampelnya adalah sebagai berikut -

TridentTopology topology = new TridentTopology();

// persistentAggregate - saving the count to memory
topology.newStream("spout", spout)
   .each(new Fields(“a, b"), new MyFunction(), new Fields(“d”))
   .groupBy(new Fields(“d”)
   .persistentAggregate(new MemoryMapState.Factory(), new Count(), new Fields("count"));

Penggabungan dan Bergabung

Penggabungan dan penggabungan dapat dilakukan dengan menggunakan metode “merge” dan “join”. Penggabungan menggabungkan satu atau lebih aliran. Penggabungan mirip dengan penggabungan, kecuali fakta bahwa penggabungan menggunakan bidang tupel trisula dari kedua sisi untuk memeriksa dan menggabungkan dua aliran. Selain itu, bergabung hanya akan bekerja di bawah tingkat batch. Kode sampelnya adalah sebagai berikut -

TridentTopology topology = new TridentTopology();
topology.merge(stream1, stream2, stream3);
topology.join(stream1, new Fields("key"), stream2, new Fields("x"), 
   new Fields("key", "a", "b", "c"));

Pemeliharaan Negara

Trident menyediakan mekanisme untuk pemeliharaan negara. Informasi status dapat disimpan di topologi itu sendiri, jika tidak, Anda juga dapat menyimpannya di database terpisah. Alasannya adalah untuk mempertahankan status bahwa jika ada tupel yang gagal selama pemrosesan, maka tupel yang gagal akan dicoba kembali. Ini menimbulkan masalah saat memperbarui status karena Anda tidak yakin apakah status tupel ini telah diperbarui sebelumnya atau tidak. Jika tupel gagal sebelum memperbarui status, maka mencoba kembali tupel akan membuat status stabil. Namun, jika tupel gagal setelah memperbarui status, maka mencoba kembali tupel yang sama akan meningkatkan jumlah dalam database dan membuat status tidak stabil. Seseorang perlu melakukan langkah-langkah berikut untuk memastikan pesan diproses hanya sekali -

  • Proses tupel dalam kelompok kecil.

  • Tetapkan ID unik untuk setiap kelompok. Jika batch dicoba lagi, batch tersebut diberi ID unik yang sama.

  • Pembaruan status dipesan di antara batch. Misalnya, pembaruan status untuk batch kedua tidak akan memungkinkan hingga pembaruan status untuk batch pertama telah selesai.

RPC Terdistribusi

RPC terdistribusi digunakan untuk melakukan kueri dan mengambil hasil dari topologi Trident. Storm memiliki server RPC terdistribusi bawaan. Server RPC terdistribusi menerima permintaan RPC dari klien dan meneruskannya ke topologi. Topologi memproses permintaan dan mengirimkan hasilnya ke server RPC terdistribusi, yang diarahkan oleh server RPC terdistribusi ke klien. Kueri RPC terdistribusi Trident mengeksekusi seperti kueri RPC normal, kecuali fakta bahwa kueri ini dijalankan secara paralel.

Kapan Menggunakan Trident?

Seperti dalam banyak kasus penggunaan, jika persyaratannya adalah memproses kueri hanya sekali, kita bisa mencapainya dengan menulis topologi di Trident. Di sisi lain, akan sulit untuk mencapai proses tepat setelah diproses dalam kasus Storm. Karenanya Trident akan berguna untuk kasus penggunaan di mana Anda memerlukan tepat sekali pemrosesan. Trident tidak untuk semua kasus penggunaan, terutama kasus penggunaan berperforma tinggi karena Trident menambahkan kerumitan pada Storm dan mengelola status.

Contoh Pengerjaan Trident

Kami akan mengonversi aplikasi penganalisis log panggilan kami yang bekerja di bagian sebelumnya ke kerangka Trident. Aplikasi Trident relatif lebih mudah dibandingkan dengan badai biasa, berkat API tingkat tinggi. Storm pada dasarnya akan diminta untuk melakukan salah satu dari operasi Fungsi, Filter, Agregat, GroupBy, Gabung, dan Gabung di Trident. Akhirnya kita akan memulai Server DRPC menggunakanLocalDRPC kelas dan cari beberapa kata kunci menggunakan execute metode kelas LocalDRPC.

Memformat informasi panggilan

Tujuan dari kelas FormatCall adalah untuk memformat informasi panggilan yang terdiri dari "Nomor pemanggil" dan "nomor Penerima". Kode program lengkapnya adalah sebagai berikut -

Coding: FormatCall.java

import backtype.storm.tuple.Values;

import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

public class FormatCall extends BaseFunction {
   @Override
   public void execute(TridentTuple tuple, TridentCollector collector) {
      String fromMobileNumber = tuple.getString(0);
      String toMobileNumber = tuple.getString(1);
      collector.emit(new Values(fromMobileNumber + " - " + toMobileNumber));
   }
}

CSVSplit

Tujuan dari kelas CSVSplit adalah untuk membagi string input berdasarkan "koma (,)" dan mengeluarkan setiap kata dalam string tersebut. Fungsi ini digunakan untuk mengurai argumen input kueri terdistribusi. Kode lengkapnya adalah sebagai berikut -

Pengkodean: CSVSplit.java

import backtype.storm.tuple.Values;

import storm.trident.operation.BaseFunction;
import storm.trident.operation.TridentCollector;
import storm.trident.tuple.TridentTuple;

public class CSVSplit extends BaseFunction {
   @Override
   public void execute(TridentTuple tuple, TridentCollector collector) {
      for(String word: tuple.getString(0).split(",")) {
         if(word.length() > 0) {
            collector.emit(new Values(word));
         }
      }
   }
}

Penganalisis Log

Ini adalah aplikasi utama. Awalnya, aplikasi akan menginisialisasi TridentTopology dan memberi makan informasi pemanggil menggunakanFeederBatchSpout. Aliran topologi trisula dapat dibuat menggunakannewStreammetode kelas TridentTopology. Demikian pula, aliran DRPC topologi Trident dapat dibuat menggunakannewDRCPStreammetode kelas TridentTopology. Server DRCP sederhana dapat dibuat menggunakan kelas LocalDRPC.LocalDRPCtelah menjalankan metode untuk mencari beberapa kata kunci. Kode lengkap diberikan di bawah ini.

Coding: LogAnalyserTrident.java

import java.util.*;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.LocalDRPC;
import backtype.storm.utils.DRPCClient;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import storm.trident.TridentState;
import storm.trident.TridentTopology;
import storm.trident.tuple.TridentTuple;

import storm.trident.operation.builtin.FilterNull;
import storm.trident.operation.builtin.Count;
import storm.trident.operation.builtin.Sum;
import storm.trident.operation.builtin.MapGet;
import storm.trident.operation.builtin.Debug;
import storm.trident.operation.BaseFilter;

import storm.trident.testing.FixedBatchSpout;
import storm.trident.testing.FeederBatchSpout;
import storm.trident.testing.Split;
import storm.trident.testing.MemoryMapState;

import com.google.common.collect.ImmutableList;

public class LogAnalyserTrident {
   public static void main(String[] args) throws Exception {
      System.out.println("Log Analyser Trident");
      TridentTopology topology = new TridentTopology();
		
      FeederBatchSpout testSpout = new FeederBatchSpout(ImmutableList.of("fromMobileNumber",
         "toMobileNumber", "duration"));

      TridentState callCounts = topology
         .newStream("fixed-batch-spout", testSpout)
         .each(new Fields("fromMobileNumber", "toMobileNumber"), 
         new FormatCall(), new Fields("call"))
         .groupBy(new Fields("call"))
         .persistentAggregate(new MemoryMapState.Factory(), new Count(), 
         new Fields("count"));

      LocalDRPC drpc = new LocalDRPC();

      topology.newDRPCStream("call_count", drpc)
         .stateQuery(callCounts, new Fields("args"), new MapGet(), new Fields("count"));

      topology.newDRPCStream("multiple_call_count", drpc)
         .each(new Fields("args"), new CSVSplit(), new Fields("call"))
         .groupBy(new Fields("call"))
         .stateQuery(callCounts, new Fields("call"), new MapGet(), 
         new Fields("count"))
         .each(new Fields("call", "count"), new Debug())
         .each(new Fields("count"), new FilterNull())
         .aggregate(new Fields("count"), new Sum(), new Fields("sum"));

      Config conf = new Config();
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("trident", conf, topology.build());
      Random randomGenerator = new Random();
      int idx = 0;
		
      while(idx < 10) {
         testSpout.feed(ImmutableList.of(new Values("1234123401", 
            "1234123402", randomGenerator.nextInt(60))));

         testSpout.feed(ImmutableList.of(new Values("1234123401", 
            "1234123403", randomGenerator.nextInt(60))));

         testSpout.feed(ImmutableList.of(new Values("1234123401", 
            "1234123404", randomGenerator.nextInt(60))));

         testSpout.feed(ImmutableList.of(new Values("1234123402", 
            "1234123403", randomGenerator.nextInt(60))));

         idx = idx + 1;
      }

      System.out.println("DRPC : Query starts");
      System.out.println(drpc.execute("call_count","1234123401 - 1234123402"));
      System.out.println(drpc.execute("multiple_call_count", "1234123401 -
         1234123402,1234123401 - 1234123403"));
      System.out.println("DRPC : Query ends");

      cluster.shutdown();
      drpc.shutdown();

      // DRPCClient client = new DRPCClient("drpc.server.location", 3772);
   }
}

Membangun dan Menjalankan Aplikasi

Aplikasi lengkap memiliki tiga kode Java. Mereka adalah sebagai berikut -

  • FormatCall.java
  • CSVSplit.java
  • LogAnalyerTrident.java

Aplikasi dapat dibangun dengan menggunakan perintah berikut -

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

Aplikasi dapat dijalankan dengan menggunakan perintah berikut -

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

Keluaran

Setelah aplikasi dimulai, aplikasi akan menampilkan detail lengkap tentang proses startup cluster, pemrosesan operasi, DRPC Server dan informasi klien, dan akhirnya, proses penutupan cluster. Output ini akan ditampilkan di konsol seperti yang ditunjukkan di bawah ini.

DRPC : Query starts
[["1234123401 - 1234123402",10]]
DEBUG: [1234123401 - 1234123402, 10]
DEBUG: [1234123401 - 1234123403, 10]
[[20]]
DRPC : Query ends

Di sini, di bab ini, kita akan membahas aplikasi Apache Storm real-time. Kita akan melihat bagaimana Storm digunakan di Twitter.

Indonesia

Twitter adalah layanan jejaring sosial online yang menyediakan platform untuk mengirim dan menerima tweet pengguna. Pengguna terdaftar dapat membaca dan memposting tweet, tetapi pengguna yang tidak terdaftar hanya dapat membaca tweet. Hashtag digunakan untuk mengkategorikan tweet dengan kata kunci dengan menambahkan # sebelum kata kunci yang relevan. Sekarang mari kita ambil skenario waktu nyata untuk menemukan hashtag yang paling banyak digunakan per topik.

Pembuatan Cerat

Tujuan dari corot adalah untuk mendapatkan tweet yang dikirimkan oleh orang-orang secepatnya. Twitter menyediakan "Twitter Streaming API", alat berbasis layanan web untuk mengambil tweet yang dikirimkan oleh orang-orang secara real time. Twitter Streaming API dapat diakses dalam bahasa pemrograman apa pun.

twitter4j adalah pustaka Java open source tidak resmi, yang menyediakan modul berbasis Java untuk mengakses API Streaming Twitter dengan mudah. twitter4jmenyediakan kerangka kerja berbasis pendengar untuk mengakses tweet. Untuk mengakses API Streaming Twitter, kita perlu masuk ke akun pengembang Twitter dan mendapatkan detail otentikasi OAuth berikut.

  • Customerkey
  • CustomerSecret
  • AccessToken
  • AccessTookenSecret

Storm memberikan cerat twitter, TwitterSampleSpout,di starter kitnya. Kami akan menggunakannya untuk mengambil tweet. Cerat membutuhkan detail otentikasi OAuth dan setidaknya satu kata kunci. Cerat akan memancarkan tweet real-time berdasarkan kata kunci. Kode program lengkap diberikan di bawah ini.

Coding: TwitterSampleSpout.java

import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;

import twitter4j.FilterQuery;
import twitter4j.StallWarning;
import twitter4j.Status;
import twitter4j.StatusDeletionNotice;
import twitter4j.StatusListener;

import twitter4j.TwitterStream;
import twitter4j.TwitterStreamFactory;
import twitter4j.auth.AccessToken;
import twitter4j.conf.ConfigurationBuilder;

import backtype.storm.Config;
import backtype.storm.spout.SpoutOutputCollector;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseRichSpout;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;

import backtype.storm.utils.Utils;

@SuppressWarnings("serial")
public class TwitterSampleSpout extends BaseRichSpout {
   SpoutOutputCollector _collector;
   LinkedBlockingQueue<Status> queue = null;
   TwitterStream _twitterStream;
		
   String consumerKey;
   String consumerSecret;
   String accessToken;
   String accessTokenSecret;
   String[] keyWords;
		
   public TwitterSampleSpout(String consumerKey, String consumerSecret,
      String accessToken, String accessTokenSecret, String[] keyWords) {
         this.consumerKey = consumerKey;
         this.consumerSecret = consumerSecret;
         this.accessToken = accessToken;
         this.accessTokenSecret = accessTokenSecret;
         this.keyWords = keyWords;
   }
		
   public TwitterSampleSpout() {
      // TODO Auto-generated constructor stub
   }
		
   @Override
   public void open(Map conf, TopologyContext context,
      SpoutOutputCollector collector) {
         queue = new LinkedBlockingQueue<Status>(1000);
         _collector = collector;
         StatusListener listener = new StatusListener() {
            @Override
            public void onStatus(Status status) {
               queue.offer(status);
            }
					
            @Override
            public void onDeletionNotice(StatusDeletionNotice sdn) {}
					
            @Override
            public void onTrackLimitationNotice(int i) {}
					
            @Override
            public void onScrubGeo(long l, long l1) {}
					
            @Override
            public void onException(Exception ex) {}
					
            @Override
            public void onStallWarning(StallWarning arg0) {
               // TODO Auto-generated method stub
            }
         };
				
         ConfigurationBuilder cb = new ConfigurationBuilder();
				
         cb.setDebugEnabled(true)
            .setOAuthConsumerKey(consumerKey)
            .setOAuthConsumerSecret(consumerSecret)
            .setOAuthAccessToken(accessToken)
            .setOAuthAccessTokenSecret(accessTokenSecret);
					
         _twitterStream = new TwitterStreamFactory(cb.build()).getInstance();
         _twitterStream.addListener(listener);
				
         if (keyWords.length == 0) {
            _twitterStream.sample();
         }else {
            FilterQuery query = new FilterQuery().track(keyWords);
            _twitterStream.filter(query);
         }
   }
			
   @Override
   public void nextTuple() {
      Status ret = queue.poll();
				
      if (ret == null) {
         Utils.sleep(50);
      } else {
         _collector.emit(new Values(ret));
      }
   }
			
   @Override
   public void close() {
      _twitterStream.shutdown();
   }
			
   @Override
   public Map<String, Object> getComponentConfiguration() {
      Config ret = new Config();
      ret.setMaxTaskParallelism(1);
      return ret;
   }
			
   @Override
   public void ack(Object id) {}
			
   @Override
   public void fail(Object id) {}
			
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("tweet"));
   }
}

Baut Pembaca Hashtag

Kicauan yang dipancarkan oleh cerat akan diteruskan ke HashtagReaderBolt, yang akan memproses tweet dan memancarkan semua hashtag yang tersedia. HashtagReaderBolt menggunakangetHashTagEntitiesmetode yang disediakan oleh twitter4j. getHashTagEntities membaca tweet dan mengembalikan daftar hashtag. Kode program lengkapnya adalah sebagai berikut -

Coding: HashtagReaderBolt.java

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

import twitter4j.*;
import twitter4j.conf.*;

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 HashtagReaderBolt implements IRichBolt {
   private OutputCollector collector;

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

   @Override
   public void execute(Tuple tuple) {
      Status tweet = (Status) tuple.getValueByField("tweet");
      for(HashtagEntity hashtage : tweet.getHashtagEntities()) {
         System.out.println("Hashtag: " + hashtage.getText());
         this.collector.emit(new Values(hashtage.getText()));
      }
   }

   @Override
   public void cleanup() {}

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

Baut Penghitung Hashtag

Hashtag yang dikeluarkan akan diteruskan ke HashtagCounterBolt. Baut ini akan memproses semua tagar dan menyimpan setiap tagar dan jumlahnya di memori menggunakan objek Java Map. Kode program lengkap diberikan di bawah ini.

Pengkodean: HashtagCounterBolt.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 HashtagCounterBolt 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 key = tuple.getString(0);

      if(!counterMap.containsKey(key)){
         counterMap.put(key, 1);
      }else{
         Integer c = counterMap.get(key) + 1;
         counterMap.put(key, c);
      }
		
      collector.ack(tuple);
   }

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

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

Mengirimkan Topologi

Mengirimkan topologi adalah aplikasi utama. Topologi Twitter terdiri dariTwitterSampleSpout, HashtagReaderBolt, dan HashtagCounterBolt. Kode program berikut menunjukkan cara mengirimkan topologi.

Coding: TwitterHashtagStorm.java

import java.util.*;

import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

public class TwitterHashtagStorm {
   public static void main(String[] args) throws Exception{
      String consumerKey = args[0];
      String consumerSecret = args[1];
		
      String accessToken = args[2];
      String accessTokenSecret = args[3];
		
      String[] arguments = args.clone();
      String[] keyWords = Arrays.copyOfRange(arguments, 4, arguments.length);
		
      Config config = new Config();
      config.setDebug(true);
		
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("twitter-spout", new TwitterSampleSpout(consumerKey,
         consumerSecret, accessToken, accessTokenSecret, keyWords));

      builder.setBolt("twitter-hashtag-reader-bolt", new HashtagReaderBolt())
         .shuffleGrouping("twitter-spout");

      builder.setBolt("twitter-hashtag-counter-bolt", new HashtagCounterBolt())
         .fieldsGrouping("twitter-hashtag-reader-bolt", new Fields("hashtag"));
			
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("TwitterHashtagStorm", config,
         builder.createTopology());
      Thread.sleep(10000);
      cluster.shutdown();
   }
}

Membangun dan Menjalankan Aplikasi

Aplikasi lengkap memiliki empat kode Java. Mereka adalah sebagai berikut -

  • TwitterSampleSpout.java
  • HashtagReaderBolt.java
  • HashtagCounterBolt.java
  • TwitterHashtagStorm.java

Anda dapat mengkompilasi aplikasi menggunakan perintah berikut -

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

Jalankan aplikasi menggunakan perintah berikut -

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:”/path/to/twitter4j/lib/*”:.
TwitterHashtagStorm <customerkey> <customersecret> <accesstoken> <accesstokensecret>
<keyword1> <keyword2> … <keywordN>

Keluaran

Aplikasi akan mencetak hashtag yang tersedia saat ini dan jumlahnya. Outputnya harus serupa dengan berikut -

Result: jazztastic : 1
Result: foodie : 1
Result: Redskins : 1
Result: Recipe : 1
Result: cook : 1
Result: android : 1
Result: food : 2
Result: NoToxicHorseMeat : 1
Result: Purrs4Peace : 1
Result: livemusic : 1
Result: VIPremium : 1
Result: Frome : 1
Result: SundayRoast : 1
Result: Millennials : 1
Result: HealthWithKier : 1
Result: LPs30DaysofGratitude : 1
Result: cooking : 1
Result: gameinsight : 1
Result: Countryfile : 1
Result: androidgames : 1

Yahoo! Keuangan adalah situs web berita bisnis dan data keuangan terkemuka di Internet. Ini adalah bagian dari Yahoo! dan memberikan informasi tentang berita keuangan, statistik pasar, data pasar internasional, dan informasi lain tentang sumber daya keuangan yang dapat diakses siapa saja.

Jika Anda adalah Yahoo! pengguna, lalu Anda dapat menyesuaikan Yahoo! Keuangan untuk memanfaatkan penawaran tertentu. Yahoo! Finance API digunakan untuk menanyakan data keuangan dari Yahoo!

API ini menampilkan data yang tertunda selama 15 menit dari waktu nyata, dan memperbarui database-nya setiap 1 menit, untuk mengakses informasi terkait saham saat ini. Sekarang mari kita ambil skenario waktu nyata dari sebuah perusahaan dan lihat bagaimana cara meningkatkan peringatan ketika nilai sahamnya turun di bawah 100.

Pembuatan Cerat

Tujuan cerat adalah untuk mendapatkan detail perusahaan dan mengirimkan harga ke baut. Anda dapat menggunakan kode program berikut untuk membuat cerat.

Coding: YahooFinanceSpout.java

import java.util.*;
import java.io.*;
import java.math.BigDecimal;

//import yahoofinace packages
import yahoofinance.YahooFinance;
import yahoofinance.Stock;

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

import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;

import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;

public class YahooFinanceSpout implements IRichSpout {
   private SpoutOutputCollector collector;
   private boolean completed = false;
   private TopologyContext context;
	
   @Override
   public void open(Map conf, TopologyContext context, SpoutOutputCollector collector){
      this.context = context;
      this.collector = collector;
   }

   @Override
   public void nextTuple() {
      try {
         Stock stock = YahooFinance.get("INTC");
         BigDecimal price = stock.getQuote().getPrice();

         this.collector.emit(new Values("INTC", price.doubleValue()));
         stock = YahooFinance.get("GOOGL");
         price = stock.getQuote().getPrice();

         this.collector.emit(new Values("GOOGL", price.doubleValue()));
         stock = YahooFinance.get("AAPL");
         price = stock.getQuote().getPrice();

         this.collector.emit(new Values("AAPL", price.doubleValue()));
      } catch(Exception e) {}
   }

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

   @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

Di sini tujuan baut adalah untuk memproses harga perusahaan tertentu ketika harga jatuh di bawah 100. Ini menggunakan objek Peta Jawa untuk mengatur peringatan batas harga cutoff sebagai truesaat harga saham turun di bawah 100; jika tidak salah. Kode program lengkapnya adalah sebagai berikut -

Coding: PriceCutOffBolt.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 PriceCutOffBolt implements IRichBolt {
   Map<String, Integer> cutOffMap;
   Map<String, Boolean> resultMap;
	
   private OutputCollector collector;

   @Override
   public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
      this.cutOffMap = new HashMap <String, Integer>();
      this.cutOffMap.put("INTC", 100);
      this.cutOffMap.put("AAPL", 100);
      this.cutOffMap.put("GOOGL", 100);

      this.resultMap = new HashMap<String, Boolean>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple tuple) {
      String company = tuple.getString(0);
      Double price = tuple.getDouble(1);

      if(this.cutOffMap.containsKey(company)){
         Integer cutOffPrice = this.cutOffMap.get(company);

         if(price < cutOffPrice) {
            this.resultMap.put(company, true);
         } else {
            this.resultMap.put(company, false);
         }
      }
		
      collector.ack(tuple);
   }

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

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

Mengirimkan Topologi

Ini adalah aplikasi utama tempat YahooFinanceSpout.java dan PriceCutOffBolt.java dihubungkan bersama dan menghasilkan topologi. Kode program berikut menunjukkan bagaimana Anda dapat mengirimkan topologi.

Pengkodean: YahooFinanceStorm.java

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

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;

public class YahooFinanceStorm {
   public static void main(String[] args) throws Exception{
      Config config = new Config();
      config.setDebug(true);
		
      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("yahoo-finance-spout", new YahooFinanceSpout());

      builder.setBolt("price-cutoff-bolt", new PriceCutOffBolt())
         .fieldsGrouping("yahoo-finance-spout", new Fields("company"));
			
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("YahooFinanceStorm", config, builder.createTopology());
      Thread.sleep(10000);
      cluster.shutdown();
   }
}

Membangun dan Menjalankan Aplikasi

Aplikasi lengkap memiliki tiga kode Java. Mereka adalah sebagai berikut -

  • YahooFinanceSpout.java
  • PriceCutOffBolt.java
  • YahooFinanceStorm.java

Aplikasi dapat dibangun menggunakan perintah berikut -

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

Aplikasi dapat dijalankan menggunakan perintah berikut -

javac -cp “/path/to/storm/apache-storm-0.9.5/lib/*”:”/path/to/yahoofinance/lib/*”:.
YahooFinanceStorm

Keluaran

Outputnya akan mirip dengan berikut -

GOOGL : false
AAPL : false
INTC : true

Kerangka kerja Apache Storm mendukung banyak aplikasi industri terbaik saat ini. Kami akan memberikan gambaran yang sangat singkat tentang beberapa aplikasi Storm yang paling terkenal di bab ini.

Klout

Klout adalah aplikasi yang menggunakan analitik media sosial untuk menentukan peringkat penggunanya berdasarkan pengaruh sosial online melalui Klout Score, yang merupakan nilai numerik antara 1 dan 100. Klout menggunakan abstraksi Trident bawaan Apache Storm untuk membuat topologi kompleks yang mengalirkan data.

Saluran cuaca

Weather Channel menggunakan topologi Storm untuk mencerna data cuaca. Ini telah terikat dengan Twitter untuk mengaktifkan iklan informasi cuaca di Twitter dan aplikasi seluler.OpenSignal adalah perusahaan yang mengkhususkan diri dalam pemetaan jangkauan nirkabel. StormTag dan WeatherSignaladalah proyek berbasis cuaca yang dibuat oleh OpenSignal. StormTag adalah stasiun cuaca Bluetooth yang dilampirkan ke rantai kunci. Data cuaca yang dikumpulkan oleh perangkat dikirim ke aplikasi WeatherSignal dan server OpenSignal.

Industri Telekomunikasi

Penyedia telekomunikasi memproses jutaan panggilan telepon per detik. Mereka melakukan forensik pada panggilan terputus dan kualitas suara yang buruk. Catatan detail panggilan mengalir dengan kecepatan jutaan per detik dan Apache Storm memprosesnya secara real-time dan mengidentifikasi pola yang mengganggu. Analisis badai dapat digunakan untuk terus meningkatkan kualitas panggilan.

ja/tutorial
es/tutorial
de/tutorial
fr/tutorial
th/tutorial
pt/tutorial
ru/tutorial
vi/tutorial
it/tutorial
ko/tutorial
tr/tutorial
id/tutorial
pl/tutorial
hi/tutorial
japost
espost
depost
frpost
thpost
ptpost
rupost
vipost
itpost
kopost
trpost
idpost
plpost
hipost

© Copyright 2021 - 2024 | All Rights Reserved