Apache Kafka - Tích hợp với Storm

Trong chương này, chúng ta sẽ học cách tích hợp Kafka với Apache Storm.

Về Storm

Storm ban đầu được tạo ra bởi Nathan Marz và nhóm tại BackType. Trong một thời gian ngắn, Apache Storm đã trở thành một tiêu chuẩn cho hệ thống xử lý thời gian thực phân tán cho phép bạn xử lý một khối lượng dữ liệu khổng lồ. Storm rất nhanh và một điểm chuẩn đã giúp nó đạt hơn một triệu bộ giá trị được xử lý mỗi giây trên mỗi nút. Apache Storm chạy liên tục, tiêu thụ dữ liệu từ các nguồn được cấu hình (Spouts) và chuyển dữ liệu xuống đường ống xử lý (Bu lông). Com-bined, Spouts và Bolts tạo nên một cấu trúc liên kết.

Tích hợp với Storm

Kafka và Storm bổ sung cho nhau một cách tự nhiên và sự hợp tác mạnh mẽ của họ cho phép phân tích phát trực tuyến theo thời gian thực cho dữ liệu lớn chuyển động nhanh. Tích hợp Kafka và Storm là để giúp các nhà phát triển nhập và xuất bản các luồng dữ liệu từ cấu trúc liên kết Storm dễ dàng hơn.

Dòng khái niệm

Vòi là một nguồn suối. Ví dụ: một vòi có thể đọc các bộ từ một Chủ đề Kafka và phát chúng dưới dạng một luồng. Một bu lông tiêu thụ các luồng đầu vào, xử lý và có thể phát ra các luồng mới. Bu lông có thể làm bất cứ điều gì từ chạy các chức năng, lọc bộ dữ liệu, tổng hợp phát trực tuyến, liên kết trực tuyến, nói chuyện với cơ sở dữ liệu, v.v. Mỗi nút trong cấu trúc liên kết Storm thực thi song song. Một cấu trúc liên kết chạy vô thời hạn cho đến khi bạn kết thúc nó. Storm sẽ tự động giao lại bất kỳ nhiệm vụ không thành công nào. Ngoài ra, Storm đảm bảo rằng sẽ không bị mất dữ liệu, ngay cả khi máy bị trục trặc và tin nhắn bị rớt.

Hãy để chúng tôi tìm hiểu chi tiết về API tích hợp Kafka-Storm. Có ba lớp chính để tích hợp Kafka với Storm. Chúng như sau:

BrokerHosts - ZkHosts & StaticHosts

BrokerHosts là một giao diện và ZkHosts và StaticHosts là hai cách triển khai chính của nó. ZkHosts được sử dụng để theo dõi động các nhà môi giới Kafka bằng cách duy trì các chi tiết trong ZooKeeper, trong khi StaticHosts được sử dụng để thiết lập thủ công / tĩnh các nhà môi giới Kafka và các chi tiết của nó. ZkHosts là cách đơn giản và nhanh chóng để truy cập nhà môi giới Kafka.

Chữ ký của ZkHosts như sau:

public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)

Trong đó brokerZkStr là máy chủ ZooKeeper và brokerZkPath là đường dẫn ZooKeeper để duy trì các chi tiết của nhà môi giới Kafka.

API KafkaConfig

API này được sử dụng để xác định cài đặt cấu hình cho cụm Kafka. Chữ ký của Kafka Con-fig được xác định như sau

public KafkaConfig(BrokerHosts hosts, string topic)

Hosts - Các BrokerHosts có thể là ZkHosts / StaticHosts.

Topic - tên chủ đề.

API SpoutConfig

Spoutconfig là một phần mở rộng của KafkaConfig hỗ trợ thêm thông tin ZooKeeper.

public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)

• Hosts - BrokerHosts có thể là bất kỳ triển khai nào của giao diện BrokerHosts

• Topic - tên chủ đề.

• zkRoot - Đường dẫn gốc ZooKeeper.

• id −Vòi lưu trữ trạng thái của phần bù được tiêu thụ trong Zookeeper. Id sẽ xác định duy nhất vòi của bạn.

SchemeAsMultiScheme

SchemeAsMultiScheme là một giao diện chỉ định cách ByteBuffer tiêu thụ từ Kafka được biến đổi thành một bộ bão. Nó có nguồn gốc từ MultiScheme và chấp nhận triển khai lớp Scheme. Có rất nhiều cách triển khai của lớp Scheme và một trong những cách triển khai như vậy là StringScheme, phân tích cú pháp byte như một chuỗi đơn giản. Nó cũng kiểm soát việc đặt tên cho trường đầu ra của bạn. Chữ ký được định nghĩa như sau.

public SchemeAsMultiScheme(Scheme scheme)

• Scheme - bộ đệm byte được sử dụng từ kafka.

API KafkaSpout

KafkaSpout là triển khai spout của chúng tôi, sẽ tích hợp với Storm. Nó lấy các mes-sage từ chủ đề kafka và phát nó vào hệ sinh thái Storm dưới dạng các bộ giá trị. KafkaSpout lấy chi tiết cấu hình của nó từ SpoutConfig.

Dưới đây là mã mẫu để tạo một vòi Kafka đơn giản.

// ZooKeeper connection string
BrokerHosts hosts = new ZkHosts(zkConnString);

//Creating SpoutConfig Object
SpoutConfig spoutConfig = new SpoutConfig(hosts, 
   topicName, "/" + topicName UUID.randomUUID().toString());

//convert the ByteBuffer to String.
spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

//Assign SpoutConfig to KafkaSpout.
KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);

Tạo bu lông

Bolt là một thành phần lấy các bộ giá trị làm đầu vào, xử lý bộ giá trị và tạo ra các bộ giá trị mới làm đầu ra. Bu lông sẽ thực hiện giao diện IRichBolt. Trong chương trình này, hai lớp bu lông WordSplitter-Bolt và WordCounterBolt được sử dụng để thực hiện các hoạt động.

Giao diện IRichBolt có các phương thức sau:

• Prepare- Cung cấp cho bu lông một môi trường để thực thi. Những người thực thi sẽ chạy phương thức này để khởi tạo vòi.

• Execute - Xử lý một bộ dữ liệu đầu vào.

• Cleanup - Được gọi khi có tia chớp tắt.

• declareOutputFields - Khai báo lược đồ đầu ra của bộ tuple.

Chúng ta hãy tạo SplitBolt.java, thực hiện logic để tách một câu thành các từ và CountBolt.java, thực hiện logic để tách các từ duy nhất và đếm số lần xuất hiện của nó.

SplitBolt.java

import java.util.Map;

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

import backtype.storm.task.OutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.IRichBolt;
import backtype.storm.task.TopologyContext;

public class SplitBolt implements IRichBolt {
   private OutputCollector collector;
   
   @Override
   public void prepare(Map stormConf, TopologyContext context,
      OutputCollector collector) {
      this.collector = collector;
   }
   
   @Override
   public void execute(Tuple input) {
      String sentence = input.getString(0);
      String[] words = sentence.split(" ");
      
      for(String word: words) {
         word = word.trim();
         
         if(!word.isEmpty()) {
            word = word.toLowerCase();
            collector.emit(new Values(word));
         }
         
      }

      collector.ack(input);
   }
   
   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
      declarer.declare(new Fields("word"));
   }

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

CountBolt.java

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

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

public class CountBolt implements IRichBolt{
   Map<String, Integer> counters;
   private OutputCollector collector;
   
   @Override
   public void prepare(Map stormConf, TopologyContext context,
   OutputCollector collector) {
      this.counters = new HashMap<String, Integer>();
      this.collector = collector;
   }

   @Override
   public void execute(Tuple input) {
      String str = input.getString(0);
      
      if(!counters.containsKey(str)){
         counters.put(str, 1);
      }else {
         Integer c = counters.get(str) +1;
         counters.put(str, c);
      }
   
      collector.ack(input);
   }

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

   @Override
   public void declareOutputFields(OutputFieldsDeclarer declarer) {
   
   }

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

Trình lên cấu trúc liên kết

Cấu trúc liên kết Storm về cơ bản là một cấu trúc Tiết kiệm. Lớp TopologyBuilder cung cấp các phương thức đơn giản và dễ dàng để tạo các cấu trúc liên kết phức tạp. Lớp TopologyBuilder có các phương thức để thiết lập vòi (setSpout) và đặt chốt (setBolt). Cuối cùng, TopologyBuilder có createTopology để tạo thành-pology. shuffleGrouping và các phương thức fieldsGrouping giúp thiết lập nhóm luồng cho vòi và bu lông.

Local Cluster- Đối với mục đích phát triển, chúng ta có thể tạo ra một cụm địa phương sử dụng LocalCluster đối tượng và sau đó gửi topo sử dụng submitTopology phương pháp LocalCluster lớp.

KafkaStormSample.java

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

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import backtype.storm.spout.SchemeAsMultiScheme;
import storm.kafka.trident.GlobalPartitionInformation;
import storm.kafka.ZkHosts;
import storm.kafka.Broker;
import storm.kafka.StaticHosts;
import storm.kafka.BrokerHosts;
import storm.kafka.SpoutConfig;
import storm.kafka.KafkaConfig;
import storm.kafka.KafkaSpout;
import storm.kafka.StringScheme;

public class KafkaStormSample {
   public static void main(String[] args) throws Exception{
      Config config = new Config();
      config.setDebug(true);
      config.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
      String zkConnString = "localhost:2181";
      String topic = "my-first-topic";
      BrokerHosts hosts = new ZkHosts(zkConnString);
      
      SpoutConfig kafkaSpoutConfig = new SpoutConfig (hosts, topic, "/" + topic,    
         UUID.randomUUID().toString());
      kafkaSpoutConfig.bufferSizeBytes = 1024 * 1024 * 4;
      kafkaSpoutConfig.fetchSizeBytes = 1024 * 1024 * 4;
      kafkaSpoutConfig.forceFromStart = true;
      kafkaSpoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

      TopologyBuilder builder = new TopologyBuilder();
      builder.setSpout("kafka-spout", new KafkaSpout(kafkaSpoutCon-fig));
      builder.setBolt("word-spitter", new SplitBolt()).shuffleGroup-ing("kafka-spout");
      builder.setBolt("word-counter", new CountBolt()).shuffleGroup-ing("word-spitter");
         
      LocalCluster cluster = new LocalCluster();
      cluster.submitTopology("KafkaStormSample", config, builder.create-Topology());

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

Trước khi chuyển quá trình biên dịch, tích hợp Kakfa-Storm cần có thư viện java ứng dụng khách ZooKeeper của người phụ trách. Người quản lý phiên bản 2.9.1 hỗ trợ Apache Storm phiên bản 0.9.5 (chúng tôi sử dụng trong hướng dẫn này). Tải xuống các tệp jar được chỉ định bên dưới và đặt nó vào đường dẫn lớp java.

• curator-client-2.9.1.jar
• curator-framework-2.9.1.jar

Sau khi bao gồm các tệp phụ thuộc, hãy biên dịch chương trình bằng lệnh sau,

javac -cp "/path/to/Kafka/apache-storm-0.9.5/lib/*" *.java

Chấp hành

Khởi động Kafka Producer CLI (đã giải thích ở chương trước), tạo một chủ đề mới được gọi là chủ đề đầu tiên của tôi và cung cấp một số thông báo mẫu như hình dưới đây -

hello
kafka
storm
spark
test message
another test message

Bây giờ thực thi ứng dụng bằng lệnh sau:

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

Đầu ra mẫu của ứng dụng này được chỉ định bên dưới:

storm : 1
test : 2
spark : 1
another : 1
kafka : 1
hello : 1
message : 2