Apache Kafka: integración con Storm
En este capítulo, aprenderemos cómo integrar Kafka con Apache Storm.
Acerca de Storm
Storm fue creado originalmente por Nathan Marz y el equipo de BackType. En poco tiempo, Apache Storm se convirtió en un estándar para el sistema de procesamiento distribuido en tiempo real que le permite procesar un gran volumen de datos. Storm es muy rápido y un punto de referencia lo registró en más de un millón de tuplas procesadas por segundo por nodo. Apache Storm se ejecuta de forma continua, consumiendo datos de las fuentes configuradas (Spouts) y transmite los datos por la tubería de procesamiento (Bolts). Combinados, picos y pernos forman una topología.
Integración con Storm
Kafka y Storm se complementan naturalmente entre sí, y su poderosa cooperación permite el análisis de transmisión en tiempo real para big data en rápido movimiento. La integración de Kafka y Storm facilita a los desarrolladores la ingesta y la publicación de flujos de datos de topologías de Storm.
Flujo conceptual
Un pico es una fuente de corrientes. Por ejemplo, un canalón puede leer tuplas de un tema de Kafka y emitirlas como un flujo. Un perno consume flujos de entrada, procesa y posiblemente emite nuevos flujos. Los bolts pueden hacer cualquier cosa, desde ejecutar funciones, filtrar tuplas, realizar agregaciones de transmisión, uniones de transmisión, hablar con bases de datos y más. Cada nodo de una topología Storm se ejecuta en paralelo. Una topología se ejecuta indefinidamente hasta que la finaliza. Storm reasignará automáticamente cualquier tarea fallida. Además, Storm garantiza que no habrá pérdida de datos, incluso si las máquinas se apagan y los mensajes se eliminan.
Veamos en detalle las API de integración de Kafka-Storm. Hay tres clases principales para integrar Kafka con Storm. Son los siguientes:
BrokerHosts - ZkHosts y StaticHosts
BrokerHosts es una interfaz y ZkHosts y StaticHosts son sus dos implementaciones principales. ZkHosts se utiliza para realizar un seguimiento dinámico de los corredores de Kafka manteniendo los detalles en ZooKeeper, mientras que StaticHosts se utiliza para configurar de forma manual / estática los corredores de Kafka y sus detalles. ZkHosts es la forma sencilla y rápida de acceder al corredor de Kafka.
La firma de ZkHosts es la siguiente:
public ZkHosts(String brokerZkStr, String brokerZkPath)
public ZkHosts(String brokerZkStr)Donde brokerZkStr es el host de ZooKeeper y brokerZkPath es la ruta de ZooKeeper para mantener los detalles del agente de Kafka.
API KafkaConfig
Esta API se utiliza para definir opciones de configuración para el clúster de Kafka. La firma de Kafka Con-fig se define de la siguiente manera
public KafkaConfig(BrokerHosts hosts, string topic)Hosts - Los BrokerHosts pueden ser ZkHosts / StaticHosts.
Topic - nombre del tema.
API SpoutConfig
Spoutconfig es una extensión de KafkaConfig que admite información adicional de ZooKeeper.
public SpoutConfig(BrokerHosts hosts, string topic, string zkRoot, string id)Hosts - Los BrokerHosts pueden ser cualquier implementación de la interfaz BrokerHosts
Topic - nombre del tema.
zkRoot - Ruta raíz de ZooKeeper.
id −El pico almacena el estado de las compensaciones que consume en Zookeeper. La identificación debe identificar de forma única su pico.
SchemeAsMultiScheme
SchemeAsMultiScheme es una interfaz que dicta cómo el ByteBuffer consumido de Kafka se transforma en una tupla de tormenta. Se deriva de MultiScheme y acepta la implementación de la clase Scheme. Hay muchas implementaciones de la clase Scheme y una de esas implementaciones es StringScheme, que analiza el byte como una cadena simple. También controla el nombre de su campo de salida. La firma se define de la siguiente manera.
public SchemeAsMultiScheme(Scheme scheme)Scheme - búfer de bytes consumido de kafka.
API de KafkaSpout
KafkaSpout es nuestra implementación de spout, que se integrará con Storm. Obtiene los mensajes del tema kafka y los emite en el ecosistema Storm como tuplas. KafkaSpout obtiene sus detalles de configuración de SpoutConfig.
A continuación se muestra un código de muestra para crear un canalón Kafka simple.
// 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);Creación de pernos
Bolt es un componente que toma tuplas como entrada, procesa la tupla y produce nuevas tuplas como salida. Bolts implementará la interfaz IRichBolt. En este programa, se utilizan dos clases de tornillos WordSplitter-Bolt y WordCounterBolt para realizar las operaciones.
La interfaz IRichBolt tiene los siguientes métodos:
Prepare- Proporciona al cerrojo un entorno para ejecutar. Los ejecutores ejecutarán este método para inicializar el pico.
Execute - Procesar una sola tupla de entrada.
Cleanup - Se llama cuando un cerrojo se va a apagar.
declareOutputFields - Declara el esquema de salida de la tupla.
Creemos SplitBolt.java, que implementa la lógica para dividir una oración en palabras y CountBolt.java, que implementa la lógica para separar palabras únicas y contar su ocurrencia.
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;
}
}Sometiéndose a la topología
La topología de Storm es básicamente una estructura de ahorro. La clase TopologyBuilder proporciona métodos sencillos y sencillos para crear topologías complejas. La clase TopologyBuilder tiene métodos para configurar spout (setSpout) y para configurar bolt (setBolt). Finalmente, TopologyBuilder tiene createTopology para crear topología. Los métodos shuffleGrouping y fieldsGrouping ayudan a configurar la agrupación de corrientes para spout y bolts.
Local Cluster- Para fines de desarrollo, podemos crear un clúster local utilizando LocalCluster
objeto y luego enviar la topología utilizando submitTopology
método de LocalCluster
clase.
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();
}
}Antes de mover la compilación, la integración de Kakfa-Storm necesita la biblioteca java del cliente ZooKeeper del curador. La versión 2.9.1 de Curator es compatible con la versión 0.9.5 de Apache Storm (que usamos en este tutorial). Descargue los archivos jar especificados a continuación y colóquelos en la ruta de clases de Java.
- curator-client-2.9.1.jar
- curator-framework-2.9.1.jar
Después de incluir los archivos de dependencia, compile el programa con el siguiente comando,
javac -cp "/path/to/Kafka/apache-storm-0.9.5/lib/*" *.javaEjecución
Inicie Kafka Producer CLI (explicado en el capítulo anterior), cree un nuevo tema llamado my-first-topic
y proporcione algunos mensajes de muestra como se muestra a continuación:
hello
kafka
storm
spark
test message
another test messageAhora ejecute la aplicación usando el siguiente comando:
java -cp “/path/to/Kafka/apache-storm-0.9.5/lib/*”:. KafkaStormSampleLa salida de muestra de esta aplicación se especifica a continuación:
storm : 1
test : 2
spark : 1
another : 1
kafka : 1
hello : 1
message : 2