Apache Storm - Guía rápida

¿Qué es Apache Storm?

Apache Storm es un sistema de procesamiento de big data distribuido en tiempo real. Storm está diseñado para procesar una gran cantidad de datos en un método escalable horizontal y tolerante a fallas. Es un marco de transmisión de datos que tiene la capacidad de las tasas de ingestión más altas. Aunque Storm no tiene estado, administra el entorno distribuido y el estado del clúster a través de Apache ZooKeeper. Es simple y puede ejecutar todo tipo de manipulaciones en datos en tiempo real en paralelo.

Apache Storm continúa siendo líder en análisis de datos en tiempo real. Storm es fácil de configurar, operar y garantiza que cada mensaje se procesará a través de la topología al menos una vez.

Apache Storm contra Hadoop

Básicamente, los frameworks Hadoop y Storm se utilizan para analizar big data. Ambos se complementan y difieren en algunos aspectos. Apache Storm realiza todas las operaciones excepto la persistencia, mientras que Hadoop es bueno en todo, pero se retrasa en el cálculo en tiempo real. La siguiente tabla compara los atributos de Storm y Hadoop.

Tormenta Hadoop
Procesamiento de secuencias en tiempo real Procesamiento por lotes
Apátrida Con estado
Arquitectura maestro / esclavo con coordinación basada en ZooKeeper. El nodo maestro se llama comonimbus y los esclavos son supervisors. Arquitectura maestro-esclavo con / sin coordinación basada en ZooKeeper. El nodo maestro esjob tracker y el nodo esclavo es task tracker.
Un proceso de transmisión de Storm puede acceder a decenas de miles de mensajes por segundo en el clúster. El sistema de archivos distribuido de Hadoop (HDFS) utiliza el marco MapReduce para procesar una gran cantidad de datos que lleva minutos u horas.
La topología de tormenta se ejecuta hasta que el usuario la apaga o hasta que se produce una falla irrecuperable inesperada. Los trabajos de MapReduce se ejecutan en orden secuencial y se completan eventualmente.
Both are distributed and fault-tolerant
Si nimbus / supervisor muere, el reinicio hace que continúe desde donde se detuvo, por lo que nada se ve afectado. Si JobTracker muere, todos los trabajos en ejecución se pierden.

Casos de uso de Apache Storm

Apache Storm es muy famoso por el procesamiento de flujos de big data en tiempo real. Por esta razón, la mayoría de las empresas utilizan Storm como parte integral de su sistema. Algunos ejemplos notables son los siguientes:

Twitter- Twitter está utilizando Apache Storm para su gama de "productos de Publisher Analytics". Los “Productos de análisis de editores” procesan todos y cada uno de los tweets y clics en la plataforma de Twitter. Apache Storm está profundamente integrado con la infraestructura de Twitter.

NaviSite- NaviSite utiliza Storm para el sistema de supervisión / auditoría de registros de eventos. Todos los registros generados en el sistema pasarán por Storm. Storm comparará el mensaje con el conjunto configurado de expresiones regulares y, si hay una coincidencia, ese mensaje en particular se guardará en la base de datos.

Wego- Wego es un metabuscador de viajes ubicado en Singapur. Los datos relacionados con los viajes provienen de muchas fuentes en todo el mundo con diferentes tiempos. Storm ayuda a Wego a buscar datos en tiempo real, resuelve problemas de concurrencia y encuentra la mejor coincidencia para el usuario final.

Beneficios de Apache Storm

Aquí hay una lista de los beneficios que ofrece Apache Storm:

  • Storm es de código abierto, robusto y fácil de usar. Podría utilizarse tanto en pequeñas empresas como en grandes corporaciones.

  • Storm es tolerante a fallas, flexible, confiable y es compatible con cualquier lenguaje de programación.

  • Permite el procesamiento de transmisiones en tiempo real.

  • Storm es increíblemente rápido porque tiene un enorme poder de procesamiento de datos.

  • Storm puede mantener el rendimiento incluso bajo una carga creciente al agregar recursos de forma lineal. Es altamente escalable.

  • Storm realiza la actualización de datos y la respuesta de entrega de un extremo a otro en segundos o minutos, según el problema. Tiene una latencia muy baja.

  • Storm tiene inteligencia operativa.

  • Storm proporciona un procesamiento de datos garantizado incluso si alguno de los nodos conectados en el clúster muere o se pierden mensajes.

Apache Storm lee el flujo sin procesar de datos en tiempo real desde un extremo, lo pasa a través de una secuencia de pequeñas unidades de procesamiento y genera la información procesada / útil en el otro extremo.

El siguiente diagrama muestra el concepto central de Apache Storm.

Echemos ahora un vistazo más de cerca a los componentes de Apache Storm:

Componentes Descripción
Tupla Tuple es la estructura de datos principal en Storm. Es una lista de elementos ordenados. De forma predeterminada, una tupla admite todos los tipos de datos. Generalmente, se modela como un conjunto de valores separados por comas y se pasa a un clúster de Storm.
Corriente Stream es una secuencia desordenada de tuplas.
Caños Fuente de arroyo. Generalmente, Storm acepta datos de entrada de fuentes de datos sin procesar como Twitter Streaming API, cola de Apache Kafka, cola de Kestrel, etc. De lo contrario, puede escribir spouts para leer datos de fuentes de datos. "ISpout" es la interfaz principal para implementar spouts. Algunas de las interfaces específicas son IRichSpout, BaseRichSpout, KafkaSpout, etc.
Pernos Los pernos son unidades lógicas de procesamiento. Los surtidores pasan datos al proceso de pernos y pernos y producen un nuevo flujo de salida. Bolts puede realizar las operaciones de filtrado, agregación, unión e interacción con fuentes de datos y bases de datos. Bolt recibe datos y emite a uno o más tornillos. “IBolt” es la interfaz principal para implementar pernos. Algunas de las interfaces comunes son IRichBolt, IBasicBolt, etc.

Tomemos un ejemplo en tiempo real de “Análisis de Twitter” y veamos cómo se puede modelar en Apache Storm. El siguiente diagrama muestra la estructura.

La entrada para el "Análisis de Twitter" proviene de Twitter Streaming API. Spout leerá los tweets de los usuarios que utilizan la API de transmisión de Twitter y los generará como un flujo de tuplas. Una sola tupla del pico tendrá un nombre de usuario de Twitter y un solo tweet como valores separados por comas. Luego, este grupo de tuplas se enviará a Bolt y Bolt dividirá el tweet en palabras individuales, calculará el recuento de palabras y mantendrá la información en una fuente de datos configurada. Ahora, podemos obtener fácilmente el resultado consultando la fuente de datos.

Topología

Los picos y los pernos están conectados entre sí y forman una topología. La lógica de la aplicación en tiempo real se especifica dentro de la topología de Storm. En palabras simples, una topología es un gráfico dirigido donde los vértices son cálculos y los bordes son un flujo de datos.

Una topología simple comienza con picos. Spout emite los datos a uno o más tornillos. El perno representa un nodo en la topología que tiene la lógica de procesamiento más pequeña y la salida de un perno se puede emitir a otro perno como entrada.

Storm mantiene la topología siempre en ejecución, hasta que mata la topología. El trabajo principal de Apache Storm es ejecutar la topología y ejecutará cualquier número de topología en un momento dado.

Tareas

Ahora tienes una idea básica sobre picos y pernos. Son la unidad lógica más pequeña de la topología y una topología se construye utilizando un solo pico y una serie de pernos. Deben ejecutarse correctamente en un orden particular para que la topología se ejecute correctamente. La ejecución de todos y cada uno de los caños y cerrojos de Storm se denomina "Tareas". En palabras simples, una tarea es la ejecución de un pico o un cerrojo. En un momento dado, cada pico y perno puede tener varias instancias ejecutándose en varios subprocesos separados.

Trabajadores

Una topología se ejecuta de forma distribuida en varios nodos trabajadores. Storm distribuye las tareas de manera uniforme en todos los nodos de trabajadores. La función del nodo trabajador es escuchar trabajos e iniciar o detener los procesos cada vez que llega un nuevo trabajo.

Agrupación de transmisiones

El flujo de datos fluye de los picos a los pernos o de un perno a otro. La agrupación de flujos controla cómo se enrutan las tuplas en la topología y nos ayuda a comprender el flujo de tuplas en la topología. Hay cuatro agrupaciones integradas como se explica a continuación.

Agrupación aleatoria

En la agrupación aleatoria, se distribuye aleatoriamente un número igual de tuplas entre todos los trabajadores que ejecutan los pernos. El siguiente diagrama muestra la estructura.

Agrupación de campos

Los campos con los mismos valores en tuplas se agrupan y las tuplas restantes se mantienen fuera. Luego, las tuplas con los mismos valores de campo se envían al mismo trabajador que ejecuta los pernos. Por ejemplo, si la secuencia está agrupada por el campo "palabra", las tuplas con la misma cadena, "Hola" se moverán al mismo trabajador. El siguiente diagrama muestra cómo funciona la agrupación de campos.

Agrupación global

Todas las corrientes se pueden agrupar y enviar a un solo perno. Esta agrupación envía tuplas generadas por todas las instancias del origen a una única instancia de destino (específicamente, elija el trabajador con el ID más bajo).

Todo agrupamiento

All Grouping envía una única copia de cada tupla a todas las instancias del perno receptor. Este tipo de agrupación se utiliza para enviar señales a los tornillos. Toda la agrupación es útil para las operaciones de combinación.

Uno de los aspectos más destacados de Apache Storm es que es una aplicación distribuida rápida, tolerante a fallas y sin "punto único de falla" (SPOF). Podemos instalar Apache Storm en tantos sistemas como sea necesario para aumentar la capacidad de la aplicación.

Echemos un vistazo a cómo está diseñado el clúster de Apache Storm y su arquitectura interna. El siguiente diagrama muestra el diseño del clúster.

Apache Storm tiene dos tipos de nodos, Nimbus (nodo maestro) y Supervisor(nodo trabajador). Nimbus es el componente central de Apache Storm. El trabajo principal de Nimbus es ejecutar la topología Storm. Nimbus analiza la topología y recopila la tarea a ejecutar. Luego, distribuirá la tarea a un supervisor disponible.

Un supervisor tendrá uno o más procesos de trabajador. El supervisor delegará las tareas a los procesos de los trabajadores. El proceso de trabajo generará tantos ejecutores como sea necesario y ejecutará la tarea. Apache Storm utiliza un sistema de mensajería distribuido interno para la comunicación entre nimbus y supervisores.

Componentes Descripción
Nimbo Nimbus es un nodo maestro del clúster Storm. Todos los demás nodos del clúster se denominanworker nodes. El nodo maestro es responsable de distribuir datos entre todos los nodos trabajadores, asignar tareas a los nodos trabajadores y monitorear fallas.
Supervisor Los nodos que siguen las instrucciones dadas por el nimbus se denominan Supervisores. UNsupervisor tiene múltiples procesos de trabajo y gobierna los procesos de trabajo para completar las tareas asignadas por el nimbus.
Proceso de trabajo Un proceso de trabajo ejecutará tareas relacionadas con una topología específica. Un proceso de trabajo no ejecutará una tarea por sí mismo, sino que creaexecutorsy les pide que realicen una tarea en particular. Un proceso de trabajo tendrá varios ejecutores.
Ejecutor Un ejecutor no es más que un hilo único generado por un proceso de trabajo. Un ejecutor ejecuta una o más tareas, pero solo para un pico o cerrojo específico.
Tarea Una tarea realiza el procesamiento de datos real. Entonces, es un pico o un perno.
Marco de ZooKeeper

Apache ZooKeeper es un servicio utilizado por un clúster (grupo de nodos) para coordinarse entre sí y mantener los datos compartidos con técnicas de sincronización sólidas. Nimbus no tiene estado, por lo que depende de ZooKeeper monitorear el estado del nodo de trabajo.

ZooKeeper ayuda al supervisor a interactuar con el nimbus. Es responsable de mantener el estado de nimbus y supervisor.

Storm es de naturaleza apátrida. Aunque la naturaleza sin estado tiene sus propias desventajas, en realidad ayuda a Storm a procesar datos en tiempo real de la mejor manera posible y más rápida.

Sin embargo, Storm no es del todo apátrida. Almacena su estado en Apache ZooKeeper. Dado que el estado está disponible en Apache ZooKeeper, un nimbus fallido se puede reiniciar y hacer que funcione desde donde salió. Por lo general, las herramientas de monitoreo de servicios comomonit monitoreará Nimbus y lo reiniciará si hay alguna falla.

Apache Storm también tiene una topología avanzada llamada Trident Topologycon mantenimiento de estado y también proporciona una API de alto nivel como Pig. Discutiremos todas estas características en los próximos capítulos.

Un clúster de Storm en funcionamiento debe tener un nimbus y uno o más supervisores. Otro nodo importante es Apache ZooKeeper, que se utilizará para la coordinación entre el nimbus y los supervisores.

Echemos ahora un vistazo de cerca al flujo de trabajo de Apache Storm:

  • Inicialmente, el nimbus esperará a que se le envíe la “Topología de tormenta”.

  • Una vez que se envía una topología, procesará la topología y recopilará todas las tareas que se van a realizar y el orden en el que se ejecutará la tarea.

  • Luego, el nimbus distribuirá uniformemente las tareas a todos los supervisores disponibles.

  • En un intervalo de tiempo particular, todos los supervisores enviarán latidos al nimbus para informar que todavía están vivos.

  • Cuando un supervisor muere y no envía un latido al nimbus, el nimbus asigna las tareas a otro supervisor.

  • Cuando el nimbus mismo muere, los supervisores trabajarán en la tarea ya asignada sin ningún problema.

  • Una vez completadas todas las tareas, el supervisor esperará a que entre una nueva tarea.

  • Mientras tanto, las herramientas de supervisión del servicio reiniciarán automáticamente el nimbus muerto.

  • El nimbus reiniciado continuará desde donde se detuvo. Del mismo modo, el supervisor muerto también se puede reiniciar automáticamente. Dado que tanto el nimbus como el supervisor se pueden reiniciar automáticamente y ambos continuarán como antes, Storm tiene la garantía de procesar todas las tareas al menos una vez.

  • Una vez que se procesan todas las topologías, el nimbus espera que llegue una nueva topología y, de manera similar, el supervisor espera nuevas tareas.

De forma predeterminada, hay dos modos en un clúster de Storm:

  • Local mode- Este modo se utiliza para desarrollo, pruebas y depuración porque es la forma más fácil de ver todos los componentes de topología trabajando juntos. En este modo, podemos ajustar parámetros que nos permiten ver cómo se ejecuta nuestra topología en diferentes entornos de configuración de Storm. En el modo local, las topologías de tormenta se ejecutan en la máquina local en una única JVM.

  • Production mode- En este modo, enviamos nuestra topología al clúster de tormenta en funcionamiento, que se compone de muchos procesos, que generalmente se ejecutan en diferentes máquinas. Como se discutió en el flujo de trabajo de Storm, un clúster en funcionamiento se ejecutará indefinidamente hasta que se cierre.

Apache Storm procesa datos en tiempo real y la entrada normalmente proviene de un sistema de cola de mensajes. Un sistema de mensajería distribuido externo proporcionará la entrada necesaria para el cálculo en tiempo real. Spout leerá los datos del sistema de mensajería y los convertirá en tuplas y los ingresará en Apache Storm. Lo interesante es que Apache Storm utiliza su propio sistema de mensajería distribuida internamente para la comunicación entre su nimbus y su supervisor.

¿Qué es el sistema de mensajería distribuida?

La mensajería distribuida se basa en el concepto de cola de mensajes confiable. Los mensajes se ponen en cola de forma asincrónica entre las aplicaciones cliente y los sistemas de mensajería. Un sistema de mensajería distribuida proporciona los beneficios de confiabilidad, escalabilidad y persistencia.

La mayoría de los patrones de mensajería siguen las publish-subscribe modelo (simplemente Pub-Sub) donde se llama a los remitentes de los mensajes publishers y los que quieran recibir los mensajes se llaman subscribers.

Una vez que el remitente ha publicado el mensaje, los suscriptores pueden recibir el mensaje seleccionado con la ayuda de una opción de filtrado. Normalmente tenemos dos tipos de filtrado, uno estopic-based filtering y otro es content-based filtering.

Tenga en cuenta que el modelo pub-sub solo puede comunicarse a través de mensajes. Es una arquitectura muy débilmente acoplada; incluso los remitentes no saben quiénes son sus suscriptores. Muchos de los patrones de mensajes permiten al intermediario de mensajes intercambiar mensajes de publicación para que muchos suscriptores tengan acceso oportuno. Un ejemplo de la vida real es Dish TV, que publica diferentes canales como deportes, películas, música, etc., y cualquiera puede suscribirse a su propio conjunto de canales y obtenerlos siempre que sus canales suscritos estén disponibles.

La siguiente tabla describe algunos de los sistemas de mensajería de alto rendimiento más populares:

Sistema de mensajería distribuida Descripción
Apache Kafka Kafka se desarrolló en la corporación LinkedIn y luego se convirtió en un subproyecto de Apache. Apache Kafka se basa en un modelo de publicación-suscripción distribuido, persistente y habilitado para la ruptura. Kafka es rápido, escalable y altamente eficiente.
RabbitMQ RabbitMQ es una aplicación de mensajería robusta distribuida de código abierto. Es fácil de usar y se ejecuta en todas las plataformas.
JMS (servicio de mensajes Java) JMS es una API de código abierto que admite la creación, lectura y envío de mensajes de una aplicación a otra. Proporciona entrega de mensajes garantizada y sigue el modelo de publicación-suscripción.
ActiveMQ El sistema de mensajería ActiveMQ es una API de código abierto de JMS.
ZeroMQ ZeroMQ es un procesamiento de mensajes de igual a igual sin intermediario. Proporciona patrones de mensajes push-pull, enrutador-distribuidor.
Cernícalo Kestrel es una cola de mensajes distribuida rápida, confiable y simple.

Protocolo de ahorro

Thrift se creó en Facebook para el desarrollo de servicios en varios idiomas y la llamada a procedimiento remoto (RPC). Más tarde, se convirtió en un proyecto Apache de código abierto. Apache Thrift es unInterface Definition Language y permite definir nuevos tipos de datos y la implementación de servicios sobre los tipos de datos definidos de una manera sencilla.

Apache Thrift también es un marco de comunicación que admite sistemas integrados, aplicaciones móviles, aplicaciones web y muchos otros lenguajes de programación. Algunas de las características clave asociadas con Apache Thrift son su modularidad, flexibilidad y alto rendimiento. Además, puede realizar streaming, mensajería y RPC en aplicaciones distribuidas.

Storm utiliza ampliamente Thrift Protocol para su comunicación interna y definición de datos. La topología de tormenta es simplementeThrift Structs. Storm Nimbus que ejecuta la topología en Apache Storm es unaThrift service.

Veamos ahora cómo instalar el marco Apache Storm en su máquina. Aquí hay tres grandes pasos:

  • Instale Java en su sistema, si aún no lo tiene.
  • Instale el marco de ZooKeeper.
  • Instale el marco de trabajo de Apache Storm.

Paso 1: verificar la instalación de Java

Utilice el siguiente comando para comprobar si ya tiene Java instalado en su sistema.

$ java -version

Si Java ya está allí, verá su número de versión. De lo contrario, descargue la última versión de JDK.

Paso 1.1 - Descarga JDK

Descargue la última versión de JDK utilizando el siguiente enlace: www.oracle.com

La última versión es JDK 8u 60 y el archivo es “jdk-8u60-linux-x64.tar.gz”. Descargue el archivo en su máquina.

Paso 1.2 - Extraer archivos

Generalmente, los archivos se descargan en el downloadscarpeta. Extraiga la configuración de tar usando los siguientes comandos.

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

Paso 1.3 - Mover al directorio de opciones

Para que Java esté disponible para todos los usuarios, mueva el contenido java extraído a la carpeta “/ usr / local / java”.

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

Paso 1.4 - Establecer ruta

Para establecer la ruta y las variables JAVA_HOME, agregue los siguientes comandos al archivo ~ / .bashrc.

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

Ahora aplique todos los cambios en el sistema en ejecución actual.

$ source ~/.bashrc

Paso 1.5 - Alternativas de Java

Utilice el siguiente comando para cambiar las alternativas de Java.

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

Paso 1.6

Ahora verifique la instalación de Java usando el comando de verificación (java -version) explicado en el Paso 1.

Paso 2: instalación de ZooKeeper Framework

Paso 2.1 - Descarga ZooKeeper

Para instalar el framework ZooKeeper en su máquina, visite el siguiente enlace y descargue la última versión de ZooKeeper http://zookeeper.apache.org/releases.html

A partir de ahora, la última versión de ZooKeeper es 3.4.6 (ZooKeeper-3.4.6.tar.gz).

Paso 2.2 - Extraiga el archivo tar

Extraiga el archivo tar usando los siguientes comandos:

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

Paso 2.3 - Crear archivo de configuración

Abra el archivo de configuración llamado "conf / zoo.cfg" usando el comando "vi conf / zoo.cfg" y estableciendo todos los siguientes parámetros como punto de partida.

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

Una vez que el archivo de configuración se haya guardado correctamente, puede iniciar el servidor de ZooKeeper.

Paso 2.4 - Inicie ZooKeeper Server

Utilice el siguiente comando para iniciar el servidor de ZooKeeper.

$ bin/zkServer.sh start

Después de ejecutar este comando, obtendrá una respuesta de la siguiente manera:

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

Paso 2.5 - Inicie CLI

Utilice el siguiente comando para iniciar la CLI.

$ bin/zkCli.sh

Después de ejecutar el comando anterior, se conectará al servidor de ZooKeeper y obtendrá la siguiente respuesta.

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

Paso 2.6 - Detenga el servidor ZooKeeper

Después de conectar el servidor y realizar todas las operaciones, puede detener el servidor de ZooKeeper usando el siguiente comando.

bin/zkServer.sh stop

Ha instalado con éxito Java y ZooKeeper en su máquina. Veamos ahora los pasos para instalar el framework Apache Storm.

Paso 3: instalación de Apache Storm Framework

Paso 3.1 Descarga Storm

Para instalar Storm framework en su máquina, visite el siguiente enlace y descargue la última versión de Storm http://storm.apache.org/downloads.html

A partir de ahora, la última versión de Storm es "apache-storm-0.9.5.tar.gz".

Paso 3.2 - Extraiga el archivo tar

Extraiga el archivo tar usando los siguientes comandos:

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

Paso 3.3 - Abrir archivo de configuración

La versión actual de Storm contiene un archivo en “conf / storm.yaml” que configura los demonios de Storm. Agregue la siguiente información a ese archivo.

$ 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

Después de aplicar todos los cambios, guarde y regrese a la terminal.

Paso 3.4 - Inicie el Nimbus

$ bin/storm nimbus

Paso 3.5 - Inicie el supervisor

$ bin/storm supervisor

Paso 3.6 Inicie la interfaz de usuario

$ bin/storm ui

Después de iniciar la aplicación de interfaz de usuario de Storm, escriba la URL http://localhost:8080en su navegador favorito y podría ver la información del clúster de Storm y su topología en ejecución. La página debe verse similar a la siguiente captura de pantalla.

Hemos revisado los detalles técnicos básicos de Apache Storm y ahora es el momento de codificar algunos escenarios simples.

Escenario: analizador de registro de llamadas móviles

La llamada móvil y su duración se proporcionarán como entrada a Apache Storm y Storm procesará y agrupará la llamada entre la misma persona que llama y el mismo receptor y su número total de llamadas.

Creación de pico

Spout es un componente que se utiliza para la generación de datos. Básicamente, un pico implementará una interfaz IRichSpout. La interfaz "IRichSpout" tiene los siguientes métodos importantes:

  • open- Proporciona al pico un entorno para ejecutar. Los ejecutores ejecutarán este método para inicializar el pico.

  • nextTuple - Emite los datos generados a través del recolector.

  • close - Este método se llama cuando se va a apagar un pico.

  • declareOutputFields - Declara el esquema de salida de la tupla.

  • ack - Reconoce que se procesa una tupla específica

  • fail - Especifica que una tupla específica no se procesa y no se reprocesará.

Abierto

La firma del open El método es el siguiente:

open(Map conf, TopologyContext context, SpoutOutputCollector collector)
  • conf - Proporciona configuración de tormenta para este pico.

  • context - Proporciona información completa sobre el lugar del pico dentro de la topología, su identificación de tarea, información de entrada y salida.

  • collector - Nos permite emitir la tupla que será procesada por los pernos.

nextTuple

La firma del nextTuple El método es el siguiente:

nextTuple()

nextTuple () se llama periódicamente desde el mismo ciclo que los métodos ack () y fail (). Debe liberar el control del hilo cuando no hay trabajo por hacer, para que los otros métodos tengan la oportunidad de ser llamados. Entonces, la primera línea de nextTuple verifica si el procesamiento ha terminado. Si es así, debería dormir durante al menos un milisegundo para reducir la carga en el procesador antes de regresar.

cerca

La firma del close El método es el siguiente:

close()

declareOutputFields

La firma del declareOutputFields El método es el siguiente:

declareOutputFields(OutputFieldsDeclarer declarer)

declarer - Se utiliza para declarar ID de flujo de salida, campos de salida, etc.

Este método se utiliza para especificar el esquema de salida de la tupla.

ack

La firma del ack El método es el siguiente:

ack(Object msgId)

Este método reconoce que se ha procesado una tupla específica.

fallar

La firma del nextTuple El método es el siguiente:

ack(Object msgId)

Este método informa que una tupla específica no se ha procesado por completo. Storm reprocesará la tupla específica.

FakeCallLogReaderSpout

En nuestro escenario, necesitamos recopilar los detalles del registro de llamadas. La información del registro de llamadas contiene.

  • número de la persona que llama
  • número de receptor
  • duration

Dado que no tenemos información en tiempo real de los registros de llamadas, generaremos registros de llamadas falsos. La información falsa se creará usando la clase Random. El código completo del programa se proporciona a continuación.

Codificación - 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;
   }
}

Creación de pernos

Bolt es un componente que toma tuplas como entrada, procesa la tupla y produce nuevas tuplas como salida. Los pernos implementaránIRichBoltinterfaz. En este programa, dos clases de tornillosCallLogCreatorBolt y CallLogCounterBolt se utilizan 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 - Llamado cuando un cerrojo se va a apagar.

  • declareOutputFields - Declara el esquema de salida de la tupla.

Preparar

La firma del prepare El método es el siguiente:

prepare(Map conf, TopologyContext context, OutputCollector collector)
  • conf - Proporciona configuración Storm para este perno.

  • context - Proporciona información completa sobre el lugar del perno dentro de la topología, su identificación de tarea, información de entrada y salida, etc.

  • collector - Nos permite emitir la tupla procesada.

ejecutar

La firma del execute El método es el siguiente:

execute(Tuple tuple)

aquí tuple es la tupla de entrada que se procesará.

los executeEl método procesa una sola tupla a la vez. Se puede acceder a los datos de tupla mediante el método getValue de la clase Tuple. No es necesario procesar la tupla de entrada inmediatamente. Se pueden procesar múltiples tuplas y generar como una única tupla de salida. La tupla procesada se puede emitir utilizando la clase OutputCollector.

limpiar

La firma del cleanup El método es el siguiente:

cleanup()

declareOutputFields

La firma del declareOutputFields El método es el siguiente:

declareOutputFields(OutputFieldsDeclarer declarer)

Aquí el parámetro declarer se utiliza para declarar ID de flujo de salida, campos de salida, etc.

Este método se utiliza para especificar el esquema de salida de la tupla.

Registro de llamadas Creator Bolt

El perno del creador del registro de llamadas recibe la tupla del registro de llamadas. La tupla del registro de llamadas tiene el número de la persona que llama, el número del receptor y la duración de la llamada. Este perno simplemente crea un nuevo valor combinando el número de la persona que llama y el número del receptor. El formato del nuevo valor es "Número de llamante - Número de receptor" y se denomina como nuevo campo, "llamada". El código completo se proporciona a continuación.

Codificación - 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;
   }
}

Perno contador de registro de llamadas

El perno del contador del registro de llamadas recibe la llamada y su duración como una tupla. Este perno inicializa un objeto de diccionario (mapa) en el método de preparación. Enexecute, comprueba la tupla y crea una nueva entrada en el objeto de diccionario para cada nuevo valor de "llamada" en la tupla y establece un valor 1 en el objeto de diccionario. Para la entrada ya disponible en el diccionario, simplemente incrementa su valor. En términos simples, este perno guarda la llamada y su recuento en el objeto de diccionario. En lugar de guardar la llamada y su recuento en el diccionario, también podemos guardarla en una fuente de datos. El código completo del programa es el siguiente:

Codificación - 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;
   }
	
}

Crear 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 fijar el perno (setBolt). Finalmente, TopologyBuilder tiene createTopology para crear topología. Utilice el siguiente fragmento de código para crear una topología:

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 y fieldsGrouping Los métodos ayudan a configurar la agrupación de arroyos para picos y pernos.

Clúster local

Para propósitos de desarrollo, podemos crear un clúster local usando el objeto "LocalCluster" y luego enviar la topología usando el método "submitTopology" de la clase "LocalCluster". Uno de los argumentos para "submitTopology" es una instancia de la clase "Config". La clase "Config" se utiliza para establecer opciones de configuración antes de enviar la topología. Esta opción de configuración se fusionará con la configuración del clúster en el tiempo de ejecución y se enviará a todas las tareas (pico y cerrojo) con el método de preparación. Una vez que la topología se envía al clúster, esperaremos 10 segundos para que el clúster calcule la topología enviada y luego apagaremos el clúster mediante el método de "apagado" de "LocalCluster". El código completo del programa es el siguiente:

Codificación - 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();
   }
}

Crear y ejecutar la aplicación

La aplicación completa tiene cuatro códigos Java. Ellos son -

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

La aplicación se puede construir usando el siguiente comando:

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

La aplicación se puede ejecutar usando el siguiente comando:

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

Salida

Una vez que se inicia la aplicación, se mostrarán los detalles completos sobre el proceso de inicio del clúster, el procesamiento del canal y el perno y, finalmente, el proceso de cierre del clúster. En "CallLogCounterBolt", hemos impreso la llamada y sus detalles de recuento. Esta información se mostrará en la consola de la siguiente manera:

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

Idiomas que no son JVM

Las topologías de tormenta se implementan mediante interfaces Thrift, lo que facilita el envío de topologías en cualquier idioma. Storm admite Ruby, Python y muchos otros lenguajes. Echemos un vistazo al enlace de Python.

Enlace de Python

Python es un lenguaje de programación de uso general interpretado, interactivo, orientado a objetos y de alto nivel. Storm admite Python para implementar su topología. Python admite operaciones de emisión, anclaje, seguimiento y registro.

Como sabe, los tornillos se pueden definir en cualquier idioma. Los pernos escritos en otro idioma se ejecutan como subprocesos, y Storm se comunica con esos subprocesos con mensajes JSON a través de stdin / stdout. Primero, tome un WordCount de perno de muestra que admita el enlace de Python.

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

Aqui la clase WordCount implementa el IRichBoltinterfaz y se ejecuta con la implementación de Python especificada como argumento del súper método "splitword.py". Ahora cree una implementación de Python llamada "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()

Esta es la implementación de muestra para Python que cuenta las palabras en una oración determinada. De manera similar, también puede vincularse con otros idiomas de soporte.

Trident es una extensión de Storm. Al igual que Storm, Trident también fue desarrollado por Twitter. La razón principal detrás del desarrollo de Trident es proporcionar una abstracción de alto nivel sobre Storm junto con procesamiento de flujo con estado y consultas distribuidas de baja latencia.

Trident usa pico y cerrojo, pero estos componentes de bajo nivel son generados automáticamente por Trident antes de la ejecución. Trident tiene funciones, filtros, uniones, agrupación y agregación.

Trident procesa los flujos como una serie de lotes que se denominan transacciones. Generalmente, el tamaño de esos pequeños lotes será del orden de miles o millones de tuplas, según el flujo de entrada. De esta manera, Trident es diferente de Storm, que realiza el procesamiento tupla por tupla.

El concepto de procesamiento por lotes es muy similar al de las transacciones de bases de datos. A cada transacción se le asigna un ID de transacción. La transacción se considera exitosa, una vez que se completa todo su procesamiento. Sin embargo, una falla en el procesamiento de una de las tuplas de la transacción hará que se retransmita toda la transacción. Para cada lote, Trident llamará a beginCommit al comienzo de la transacción y se comprometerá al final de la misma.

Topología tridente

Trident API expone una opción fácil para crear topología Trident usando la clase "TridentTopology". Básicamente, la topología Trident recibe el flujo de entrada del canal y realiza una secuencia ordenada de operación (filtro, agregación, agrupación, etc.) en el flujo. Storm Tuple se reemplaza por Trident Tuple y Bolts se reemplazan por operaciones. Se puede crear una topología Trident simple de la siguiente manera:

TridentTopology topology = new TridentTopology();

Tuplas tridentes

La tupla tridente es una lista de valores con nombre. La interfaz TridentTuple es el modelo de datos de una topología Trident. La interfaz TridentTuple es la unidad básica de datos que puede procesar una topología Trident.

Caño tridente

El pico Trident es similar al pico Storm, con opciones adicionales para usar las funciones de Trident. De hecho, todavía podemos usar el IRichSpout, que hemos usado en la topología de Storm, pero será de naturaleza no transaccional y no podremos usar las ventajas proporcionadas por Trident.

El pico básico que tiene todas las funciones para utilizar las funciones de Trident es "ITridentSpout". Es compatible con semántica transaccional transaccional y opaca. Los otros surtidores son IBatchSpout, IPartitionedTridentSpout e IOpaquePartitionedTridentSpout.

Además de estos picos genéricos, Trident tiene muchos ejemplos de implementación del pico tridente. Uno de ellos es el pico FeederBatchSpout, que podemos usar para enviar listas nombradas de tuplas tridentes fácilmente sin preocuparnos por el procesamiento por lotes, el paralelismo, etc.

La creación de FeederBatchSpout y la alimentación de datos se pueden realizar como se muestra a continuación:

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)));

Operaciones de tridente

Trident se basa en la "Operación Trident" para procesar el flujo de entrada de tuplas trident. La API Trident tiene una serie de operaciones integradas para manejar el procesamiento de flujos de simples a complejos. Estas operaciones van desde la validación simple hasta la agrupación compleja y agregación de tuplas de tridentes. Repasemos las operaciones más importantes y de uso frecuente.

Filtrar

El filtro es un objeto utilizado para realizar la tarea de validación de entrada. Un filtro Trident obtiene un subconjunto de campos de tupla trident como entrada y devuelve verdadero o falso dependiendo de si se cumplen ciertas condiciones o no. Si se devuelve true, la tupla se mantiene en el flujo de salida; de lo contrario, la tupla se elimina de la secuencia. El filtro básicamente heredará delBaseFilter clase e implementar el isKeepmétodo. Aquí hay una implementación de muestra del funcionamiento del filtro:

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]

La función de filtro se puede llamar en la topología utilizando el método "cada". La clase "Campos" se puede utilizar para especificar la entrada (subconjunto de la tupla tridente). El código de muestra es el siguiente:

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

Función

Functiones un objeto que se utiliza para realizar una operación simple en una tupla de un solo tridente. Toma un subconjunto de campos de tupla tridente y emite cero o más campos de tupla tridente nuevos.

Function básicamente hereda de la BaseFunction clase e implementa el executemétodo. A continuación se ofrece una implementación de muestra:

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]

Al igual que la operación de filtro, la operación de función se puede llamar en una topología utilizando el eachmétodo. El código de muestra es el siguiente:

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

Agregación

La agregación es un objeto que se utiliza para realizar operaciones de agregación en un lote de entrada, una partición o un flujo. Trident tiene tres tipos de agregación. Son los siguientes:

  • aggregate- Agrega cada lote de tupla tridente de forma aislada. Durante el proceso de agregación, las tuplas se reparten inicialmente utilizando la agrupación global para combinar todas las particiones del mismo lote en una sola partición.

  • partitionAggregate- Agrega cada partición en lugar de todo el lote de tupla tridente. La salida del agregado de la partición reemplaza completamente la tupla de entrada. La salida del agregado de partición contiene una única tupla de campo.

  • persistentaggregate - Agrega todas las tuplas de tridentes en todos los lotes y almacena el resultado en la memoria o en la base de datos.

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"));

La operación de agregación se puede crear usando CombinerAggregator, ReducerAggregator o la interfaz de agregación genérica. El agregador de "recuento" utilizado en el ejemplo anterior es uno de los agregadores integrados. Se implementa mediante "CombinerAggregator". La implementación es la siguiente:

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;
   }
}

Agrupamiento

La operación de agrupamiento es una operación incorporada y puede ser invocada por el groupBymétodo. El método groupBy reparte la secuencia haciendo un partitionBy en los campos especificados y luego dentro de cada partición, agrupa las tuplas cuyos campos de grupo son iguales. Normalmente, usamos "groupBy" junto con "persistentAggregate" para obtener la agregación agrupada. El código de muestra es el siguiente:

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"));

Fusión y unión

La fusión y la unión se pueden realizar mediante el método "fusionar" y "unión", respectivamente. La fusión combina una o más corrientes. La unión es similar a la fusión, excepto por el hecho de que la unión utiliza un campo de tupla tridente de ambos lados para verificar y unir dos corrientes. Además, la unión solo funcionará a nivel de lote. El código de muestra es el siguiente:

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"));

Mantenimiento del estado

Trident proporciona un mecanismo para el mantenimiento del estado. La información de estado se puede almacenar en la topología misma; de lo contrario, también puede almacenarla en una base de datos separada. El motivo es mantener un estado en el que si alguna tupla falla durante el procesamiento, se reintentará la tupla fallida. Esto crea un problema al actualizar el estado porque no está seguro de si el estado de esta tupla se ha actualizado previamente o no. Si la tupla ha fallado antes de actualizar el estado, volver a intentar la tupla hará que el estado sea estable. Sin embargo, si la tupla ha fallado después de actualizar el estado, volver a intentar la misma tupla aumentará nuevamente el recuento en la base de datos y hará que el estado sea inestable. Es necesario realizar los siguientes pasos para garantizar que un mensaje se procese solo una vez:

  • Procese las tuplas en pequeños lotes.

  • Asigne una identificación única a cada lote. Si se vuelve a intentar el lote, se le asigna el mismo ID único.

  • Las actualizaciones de estado se ordenan entre lotes. Por ejemplo, la actualización del estado del segundo lote no será posible hasta que se complete la actualización del estado del primer lote.

RPC distribuido

La RPC distribuida se utiliza para consultar y recuperar el resultado de la topología Trident. Storm tiene un servidor RPC distribuido incorporado. El servidor RPC distribuido recibe la solicitud RPC del cliente y la pasa a la topología. La topología procesa la solicitud y envía el resultado al servidor RPC distribuido, que es redirigido por el servidor RPC distribuido al cliente. La consulta RPC distribuida de Trident se ejecuta como una consulta RPC normal, excepto por el hecho de que estas consultas se ejecutan en paralelo.

¿Cuándo usar Trident?

Como en muchos casos de uso, si el requisito es procesar una consulta solo una vez, podemos lograrlo escribiendo una topología en Trident. Por otro lado, será difícil lograr exactamente una vez procesado en el caso de Storm. Por lo tanto, Trident será útil para aquellos casos de uso en los que requiera exactamente un procesamiento. Trident no es para todos los casos de uso, especialmente los casos de uso de alto rendimiento porque agrega complejidad a Storm y administra el estado.

Ejemplo de trabajo de Trident

Vamos a convertir nuestra aplicación analizadora de registro de llamadas desarrollada en la sección anterior al marco Trident. La aplicación Trident será relativamente fácil en comparación con la simple tormenta, gracias a su API de alto nivel. Básicamente, se requerirá que Storm realice cualquiera de las operaciones de Función, Filtro, Agregado, GroupBy, Join y Merge en Trident. Finalmente iniciaremos el servidor DRPC usando elLocalDRPC class y busque alguna palabra clave usando la execute método de la clase LocalDRPC.

Formatear la información de la llamada

El propósito de la clase FormatCall es formatear la información de la llamada que comprende "Número de llamante" y "Número de receptor". El código completo del programa es el siguiente:

Codificación: 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

El propósito de la clase CSVSplit es dividir la cadena de entrada basándose en "coma (,)" y emitir cada palabra en la cadena. Esta función se utiliza para analizar el argumento de entrada de la consulta distribuida. El código completo es el siguiente:

Codificación: 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));
         }
      }
   }
}

Analizador de registros

Esta es la aplicación principal. Inicialmente, la aplicación inicializará TridentTopology y alimentará la información de la persona que llama usandoFeederBatchSpout. El flujo de topología tridente se puede crear utilizando elnewStreammétodo de la clase TridentTopology. De manera similar, el flujo DRPC de topología Trident se puede crear utilizando elnewDRCPStreammétodo de la clase TridentTopology. Se puede crear un servidor DRCP simple usando la clase LocalDRPC.LocalDRPCtiene un método de ejecución para buscar alguna palabra clave. El código completo se proporciona a continuación.

Codificación: 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);
   }
}

Crear y ejecutar la aplicación

La aplicación completa tiene tres códigos Java. Son los siguientes:

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

La aplicación se puede construir usando el siguiente comando:

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

La aplicación se puede ejecutar usando el siguiente comando:

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

Salida

Una vez iniciada la aplicación, la aplicación generará los detalles completos sobre el proceso de inicio del clúster, el procesamiento de operaciones, el servidor DRPC y la información del cliente y, finalmente, el proceso de cierre del clúster. Esta salida se mostrará en la consola como se muestra a continuación.

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

Aquí, en este capítulo, discutiremos una aplicación en tiempo real de Apache Storm. Veremos cómo se usa Storm en Twitter.

Gorjeo

Twitter es un servicio de redes sociales en línea que proporciona una plataforma para enviar y recibir tweets de usuarios. Los usuarios registrados pueden leer y publicar tweets, pero los usuarios no registrados solo pueden leer tweets. El hashtag se usa para categorizar tweets por palabra clave agregando # antes de la palabra clave relevante. Ahora tomemos un escenario en tiempo real para encontrar el hashtag más utilizado por tema.

Creación de pico

El propósito de spout es que las personas envíen los tweets lo antes posible. Twitter proporciona "Twitter Streaming API", una herramienta basada en servicios web para recuperar los tweets enviados por personas en tiempo real. Se puede acceder a Twitter Streaming API en cualquier lenguaje de programación.

twitter4j es una biblioteca de Java no oficial de código abierto, que proporciona un módulo basado en Java para acceder fácilmente a la API de transmisión de Twitter. twitter4jproporciona un marco basado en oyentes para acceder a los tweets. Para acceder a la API de transmisión de Twitter, debemos iniciar sesión en la cuenta de desarrollador de Twitter y obtener los siguientes detalles de autenticación de OAuth.

  • Customerkey
  • CustomerSecret
  • AccessToken
  • AccessTookenSecret

Storm proporciona un canal de Twitter, TwitterSampleSpout,en su kit de inicio. Lo usaremos para recuperar los tweets. El canalón necesita detalles de autenticación OAuth y al menos una palabra clave. El canal emitirá tweets en tiempo real basados ​​en palabras clave. El código completo del programa se proporciona a continuación.

Codificación: 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"));
   }
}

Perno del lector de hashtags

El tweet emitido por spout se reenviará a HashtagReaderBolt, que procesará el tweet y emitirá todos los hashtags disponibles. Usos de HashtagReaderBoltgetHashTagEntitiesmétodo proporcionado por twitter4j. getHashTagEntities lee el tweet y devuelve la lista de hashtag. El código completo del programa es el siguiente:

Codificación: 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;
   }
	
}

Perno contador de hashtag

El hashtag emitido se reenviará a HashtagCounterBolt. Este perno procesará todos los hashtags y guardará todos y cada uno de los hashtag y su recuento en la memoria usando el objeto Java Map. El código completo del programa se proporciona a continuación.

Codificación: 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;
   }
	
}

Envío de una topología

El envío de una topología es la aplicación principal. La topología de Twitter consta deTwitterSampleSpout, HashtagReaderBolty HashtagCounterBolt. El siguiente código de programa muestra cómo enviar una topología.

Codificación: 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();
   }
}

Crear y ejecutar la aplicación

La aplicación completa tiene cuatro códigos Java. Son los siguientes:

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

Puede compilar la aplicación usando el siguiente comando:

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

Ejecute la aplicación usando los siguientes comandos:

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

Salida

La aplicación imprimirá el hashtag disponible actual y su recuento. La salida debe ser similar a la siguiente:

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! Finance es el sitio web líder en Internet de noticias comerciales y datos financieros. Es parte de Yahoo! y brinda información sobre noticias financieras, estadísticas del mercado, datos del mercado internacional y otra información sobre recursos financieros a los que cualquiera puede acceder.

Si está registrado en Yahoo! usuario, entonces puede personalizar Yahoo! Finance para aprovechar sus determinadas ofertas. Yahoo! La API de finanzas se utiliza para consultar datos financieros de Yahoo!

Esta API muestra datos con un retraso de 15 minutos desde el tiempo real y actualiza su base de datos cada 1 minuto para acceder a la información actual relacionada con las acciones. Ahora tomemos un escenario en tiempo real de una empresa y veamos cómo generar una alerta cuando el valor de sus acciones desciende por debajo de 100.

Creación de pico

El propósito de spout es obtener los detalles de la empresa y emitir los precios a los pernos. Puede utilizar el siguiente código de programa para crear un pico.

Codificación: 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;
   }
	
}

Creación de pernos

Aquí el propósito de bolt es procesar los precios de la empresa dada cuando los precios caen por debajo de 100. Utiliza el objeto Java Map para establecer la alerta de límite de precio de corte como truecuando los precios de las acciones caen por debajo de 100; de lo contrario falso. El código completo del programa es el siguiente:

Codificación: 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;
   }
	
}

Envío de una topología

Esta es la aplicación principal en la que YahooFinanceSpout.java y PriceCutOffBolt.java están conectados y producen una topología. El siguiente código de programa muestra cómo enviar una topología.

Codificación: 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();
   }
}

Crear y ejecutar la aplicación

La aplicación completa tiene tres códigos Java. Son los siguientes:

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

La aplicación se puede construir usando el siguiente comando:

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

La aplicación se puede ejecutar usando el siguiente comando:

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

Salida

La salida será similar a la siguiente:

GOOGL : false
AAPL : false
INTC : true

El framework Apache Storm es compatible con muchas de las mejores aplicaciones industriales de la actualidad. Proporcionaremos una descripción muy breve de algunas de las aplicaciones más notables de Storm en este capítulo.

Klout

Klout es una aplicación que utiliza análisis de redes sociales para clasificar a sus usuarios en función de la influencia social en línea a través de Klout Score, que es un valor numérico entre 1 y 100. Klout utiliza la abstracción Trident incorporada de Apache Storm para crear topologías complejas que transmiten datos.

El canal del clima

El canal meteorológico utiliza topologías de tormenta para ingerir datos meteorológicos. Se ha asociado con Twitter para permitir publicidad informada sobre el clima en Twitter y aplicaciones móviles.OpenSignal es una empresa que se especializa en mapeo de cobertura inalámbrica. StormTag y WeatherSignalson proyectos basados ​​en el clima creados por OpenSignal. StormTag es una estación meteorológica Bluetooth que se conecta a un llavero. Los datos meteorológicos recopilados por el dispositivo se envían a la aplicación WeatherSignal y a los servidores OpenSignal.

Industria de las telecomunicaciones

Los proveedores de telecomunicaciones procesan millones de llamadas telefónicas por segundo. Realizan análisis forenses en llamadas caídas y mala calidad de sonido. Los registros de detalles de llamadas fluyen a una velocidad de millones por segundo y Apache Storm los procesa en tiempo real e identifica cualquier patrón problemático. El análisis de tormentas se puede utilizar para mejorar continuamente la calidad de las llamadas.