{"id":60919,"date":"2016-10-18T07:00:40","date_gmt":"2016-10-18T04:00:40","guid":{"rendered":"https:\/\/www.javacodegeeks.com\/?p=60919"},"modified":"2016-10-17T15:30:07","modified_gmt":"2016-10-17T12:30:07","slug":"complex-event-processing-cep-apache-storm-apache-ignite","status":"publish","type":"post","link":"https:\/\/www.javacodegeeks.com\/2016\/10\/complex-event-processing-cep-apache-storm-apache-ignite.html","title":{"rendered":"Complex event processing (CEP) with Apache Storm and Apache Ignite"},"content":{"rendered":"

In this article, author of the book\u00a0“High Performance in-memory computing with Apache Ignite”<\/a>, will discuss the complex event processing with Apache Strom and the Apache Ignite. Part of this article taken from the complex event processing chapter of the
\nbook<\/a>.<\/p>\n

There is no broadly or highly accepted definition of the term Complex Event Processing or CEP. What Complex Event Processing is may be briefly described as the following quote from the Wikipedia:<\/p>\n

“Complex Event Processing, or CEP, is primarily an event processing concept that deals with the task of processing multiple events with the goal of identifying the meaningful events within the event cloud. CEP employs techniques such as detection of complex patterns of many events, event correlation and abstraction, event hierarchies, and relationships between events such as causality, membership, and timing, and event-driven processes.”<\/p>\n

For simplicity, Complex Event Processing (CEP) is a technology for low-latency filtering, aggregating and computing on real-world never ending or streaming event data. The quantity and speed of both raw infrastructure and business events are exponentially growing in IT environments. In addition, the explosion of mobile devices and the ubiquity of high-speed connectivity add to the explosion of mobile data. At the same time, demand for business process agility and execution has only grown. These two trends have put pressure on organizations to increase their capability to support event-driven architecture patterns of implementation. Real-time event processing requires both the infrastructure and the application development environment to execute on event processing requirements. These requirements often include the need to scale from everyday use cases to extremely high velocities or varieties of data and event throughput, potentially with latencies measured in microseconds rather than seconds of response time.<\/p>\n

Apache Ignite allows processing continuous never-ending streams of data in scalable and fault-tolerant fashion in in-memory, rather than analyzing data after it’s reached the database. Not only does this enable you to correlate relationships and detect meaningful patterns from significantly more data, you can do it faster and much more efficiently. Event history can live in memory for any length of time (critical for long-running event sequences) or be recorded as transactions in a stored database.<\/p>\n

Apache Ignite CEP can be used in a wealth of industries area, the following are some first class use cases:<\/p>\n

    \n
  1. Financial services: the ability to perform real-time risk analysis, monitoring and reporting of financial trading and fraud detection.<\/li>\n
  2. Telecommunication: ability to perform real time call detail record and SMS monitoring and DDoS attack.<\/li>\n
  3. IT systems and infrastructure: the ability to detect failed or unavailable application or servers in real time.<\/li>\n
  4. Logistics: ability to track shipments and order processing in real-time and reports on potential delays on arrival.<\/li>\n<\/ol>\n

    There are a few more industrials or functional areas, where you can use Apache Ignite to process streams event data such as Insurance, transportation and Public sector. Complex event processing or CEP contains three main parts of its process:<\/p>\n

      \n
    1. Event Capture or data ingesting.<\/li>\n
    2. Compute or calculation of these data.<\/li>\n
    3. Response or action.<\/li>\n<\/ol>\n

      \"chap6-1\"<\/a><\/p>\n

      As shown in the above figure, data are ingesting from difference sources. Sources can be any sensors (IoT), web application or industry applications. Stream data can be concurrently processed directly on the Ignite cluster in collecting fashion. In addition, data can be enriched from other sources or filter out. After computing the data, computed or aggregated data can be exported to other systems for visualizing or taking an action.<\/p>\n

      Apache Ignite Storm Streamer module provides streaming via Storm to Ignite cache. Before start using the Ignite streamer lets take a look at the Apache Storm to get a few basics about apache Storm.<\/p>\n

      Apache storm is a distributed fault-tolerant real-time computing system. In a short time, Apache Storm became a standard for distributed real-time processing system that allows you to process a large amount of data. Apache Storm project is open source and written in Java and Clojure. It became a first choose for real-time analytics. Apache Ignite Storm streamer module provides a convenience way to streaming data via Storm to Ignite cache.<\/p>\n

      Key concepts:<\/b><\/p>\n

      Apache Storm reads raw stream of data from the one end and passes it through a sequence of small processing units and output the processed information at the other end. Let\u2019s have a detailed look at the main components of Apache Storm \u2013<\/p>\n

      Tuples<\/i> \u2013 It is the main data structure of the Storm. It\u2019s an ordered list of elements. Generally, tuple supports all primitives data types.<\/p>\n

      \"chap6-19\"<\/a><\/p>\n

      Streams<\/i> \u2013 It\u2019s an unbound and un-ordered sequence of tuples.<\/p>\n

      \"chap6-20\"<\/a><\/p>\n

      Spouts<\/i> – Source of streams, in simple terms, a spout reads the data from a source for use in topology. A spout can reliable or unreliable. A spout can talk with Queues, Web logs, event data etc.<\/p>\n

      \"chap6-21\"<\/a><\/p>\n

      Bolts<\/i> – Bolts are logical processing units, it is responsible for processing data and creating new streams. Bolts can perform the operations of filtering, aggregation, joining, interacting with files\/database and so on. Bolts receive data from the spout and emit to one or more bolts.<\/p>\n

      \"chap6-22\"<\/a><\/p>\n

      Topology<\/i> \u2013 A topology is a directed graph of Spouts and Bolts, each node of this graph contains the data processing logic (bolts) while connecting edges define the flow of the data (streams).<\/p>\n

      Unlike Hadoop, Storm keeps the topology running forever until you kill it. A simple topology starts with spouts, emit stream from the sources to bolt for processing data. Apache Storm main job is to run the topology and will run any number of topology at given time.<\/p>\n

      \"chap6-23\"<\/a>

      <\/div> <\/div><\/p>\n

      Ignite out of the box provides an implementation of Storm Bolt (StormStreamer) to streaming the computed data into Ignite cache. On the other hand, you can write down your custom Strom Bolt to ingest stream data into Ignite. To develop a custom Storm Bolt, you just have to implement *BaseBasicBolt* or *IRichBolt* Storm interface. However, if you decide to use StormStreamer, you have to configure a few properties to work the Ignite Bolt correctly. All mandatory properties are shown below:<\/p>\n\n\n\n\n\n\n\n\n
      No<\/th>\nProperty Name<\/th>\nDescription<\/th>\n<\/tr>\n
      1<\/td>\nCacheName<\/td>\nCache name of the Ignite cache, in which the data will,be store.<\/td>\n<\/tr>\n
      2<\/td>\nIgniteTupleField<\/td>\nNames the Ignite Tuple field, by which tuple data is,obtained in topology. By default the value is ignite.<\/td>\n<\/tr>\n
      3<\/td>\nIgniteConfigFile<\/td>\nThis property will set the Ignite spring configuration
      \nfile. Allows you to send and consume message to and
      \nfrom Ignite topics.<\/td>\n<\/tr>\n
      4<\/td>\nAllowOverwrite<\/td>\nIt will enabling overwriting existing values in the,cache, default value is false.<\/td>\n<\/tr>\n
      5<\/td>\nAutoFlushFrequency<\/td>\nAutomatic flush frequency in milliseconds. Essentially,,this is the time after which the streamer will make an
      \nattempt to submit all data added so far to remote
      \nnodes. Default is 10 sec.<\/td>\n<\/tr>\n<\/tbody>\n<\/table>\n

      Now that we have got the basics, let\u2019s build something useful to check how the Ignite\u00a0StormStreamer<\/i> works. The basic idea behind the application is to design one topology of spout and bolt that can process a huge amount of data from a traffic log files and trigger an alert when a specific value crosses a predefined threshold. Using a topology, the log file is read line by line and the topology is designed to monitor the incoming data. In our case, the log file will contain data, such as vehicle registration number, speed and the highway name from highway traffic camera. If the vehicle crosses the speed limit (for example 120km\/h), Storm topology will send the data to Ignite cache.<\/p>\n

      Next listing will show a CSV file of the type we are going to use in our example, which contain vehicle data information such as vehicle registration number, the speed at which the vehicle is traveling and the location of the highway.<\/p>\n

      AB 123, 160, North city\r\nBC 123, 170, South city\r\nCD 234, 40, South city\r\nDE 123, 40, East city\r\nEF 123, 190, South city\r\nGH 123, 150, West city\r\nXY 123, 110, North city\r\nGF 123, 100, South city\r\nPO 234, 140, South city\r\nXX 123, 110, East city\r\nYY 123, 120, South city\r\nZQ 123, 100, West city<\/pre>\n
      The idea of the above example is taken from the Dr. Dobbs journal. Since this book is not for studying Apache Storm, I am going to keep the example simple as possible. Also, I have added the famous word count example of Storm, which ingests the word count value into Ignite cache through StormStreamer module. If you are curious about the code, it’s available at
      \nchapter-cep\/storm<\/a>. The above CSV file will be the source for the Storm topology.<\/p>\n
      \"chap6-24\"<\/a><\/p>\n
      As shown in above figure, the\u00a0FileSourceSpout<\/i> accepts the input CSV log file, reads the data line by line and emits the data to the SpeedLimitBolt for further threshold processing. Once the processing is done and found any car with exceeding the speed limit, the data is emitted to the Ignite StormStreamer bolt, where it is ingested into the cache. Let’s dive into the detailed explanation of our Storm topology.<\/p>\n
      Step 1:<\/b><\/p>\n
      Because this is a Storm topology, you must add the Storm and the Ignite StormStreamer dependency in the maven project.<\/p>\n
      <dependency>\r\n  <groupId>org.apache.ignite<\/groupId>\r\n  <artifactId>ignite-storm<\/artifactId>\r\n  <version>1.6.0<\/version>\r\n<\/dependency>\r\n<dependency>\r\n  <groupId>org.apache.ignite<\/groupId>\r\n  <artifactId>ignite-core<\/artifactId>\r\n  <version>1.6.0<\/version>\r\n<\/dependency>\r\n<dependency>\r\n  <groupId>org.apache.ignite<\/groupId>\r\n  <artifactId>ignite-spring<\/artifactId>\r\n  <version>1.6.0<\/version>\r\n<\/dependency>\r\n<dependency>\r\n  <groupId>org.apache.storm<\/groupId>\r\n  <artifactId>storm-core<\/artifactId>\r\n  <version>0.10.0<\/version>\r\n  <exclusions>\r\n  <exclusion>\r\n  <groupId>log4j<\/groupId>\r\n  <artifactId>log4j<\/artifactId>\r\n  <\/exclusion>\r\n  <exclusion>\r\n  <groupId>org.slf4j<\/groupId>\r\n  <artifactId>slf4j-log4j12<\/artifactId>\r\n  <\/exclusion>\r\n  <exclusion>\r\n  <groupId>commons-logging<\/groupId>\r\n  <artifactId>commons-logging<\/artifactId>\r\n  <\/exclusion>\r\n  <exclusion>\r\n  <groupId>org.slf4j<\/groupId>\r\n  <artifactId>slf4j-simple<\/artifactId>\r\n  <\/exclusion>\r\n  <exclusion>\r\n  <groupId>org.slf4j<\/groupId>\r\n  <artifactId>log4j-over-slf4j<\/artifactId>\r\n  <\/exclusion>\r\n  <exclusion>\r\n  <groupId>org.apache.zookeeper<\/groupId>\r\n  <artifactId>zookeeper<\/artifactId>\r\n  <\/exclusion>\r\n  <\/exclusions>\r\n<\/dependency><\/pre>\n
      At the time of writing this book, Apache Storm version 0.10.0 is only supported. Note that, You do not need any\u00a0Kafka module<\/b> to run or execute this example as describe in the Ignite documentation.<\/p>\n
      Step 2:<\/b><\/p>\n
      Create an Ignite configuration file (see example-ignite.xml file in\u00a0\/chapter-cep\/storm\/src\/resources\/example-ignite.xml<\/a>) and make sure that it is available from the classpath. The content of the Ignite configuration is identical from the previous section of this chapter.<\/p>\n
      <beans xmlns=\"http:\/\/www.springframework.org\/schema\/beans\"\r\n  xmlns:xsi=\"http:\/\/www.w3.org\/2001\/XMLSchema-instance\"\r\n  xmlns:util=\"http:\/\/www.springframework.org\/schema\/util\"\r\n  xsi:schemaLocation=\"\r\n  http:\/\/www.springframework.org\/schema\/beans\r\n  http:\/\/www.springframework.org\/schema\/beans\/spring-beans.xsd\r\n  http:\/\/www.springframework.org\/schema\/util\r\n  http:\/\/www.springframework.org\/schema\/util\/spring-util.xsd\">\r\n  <bean id=\"ignite.cfg\" class=\"org.apache.ignite.configuration.IgniteConfiguration\">\r\n  <!-- Enable client mode. -->\r\n  <property name=\"clientMode\" value=\"true\"\/>\r\n  <!-- Cache accessed from IgniteSink. -->\r\n  <property name=\"cacheConfiguration\">\r\n  <list>\r\n  <!-- Partitioned cache example configuration with configurations adjusted to server nodes'. -->\r\n  <bean class=\"org.apache.ignite.configuration.CacheConfiguration\">\r\n  <property name=\"atomicityMode\" value=\"ATOMIC\"\/>\r\n\r\n  <property name=\"name\" value=\"testCache\"\/>\r\n  <\/bean>\r\n  <\/list>\r\n  <\/property>\r\n  <!-- Enable cache events. -->\r\n  <property name=\"includeEventTypes\">\r\n  <list>\r\n  <!-- Cache events (only EVT_CACHE_OBJECT_PUT for tests). -->\r\n  <util:constant static-field=\"org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT\"\/>\r\n  <\/list>\r\n  <\/property>\r\n  <!-- Explicitly configure TCP discovery SPI to provide list of initial nodes. -->\r\n  <property name=\"discoverySpi\">\r\n  <bean class=\"org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi\">\r\n  <property name=\"ipFinder\">\r\n  <bean class=\"org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder\">\r\n  <property name=\"addresses\">\r\n  <list>\r\n  <value>127.0.0.1:47500<\/value>\r\n  <\/list>\r\n  <\/property>\r\n  <\/bean>\r\n  <\/property>\r\n  <\/bean>\r\n  <\/property>\r\n  <\/bean>\r\n<\/beans><\/pre>\n
      Step 3:<\/b><\/p>\n
      Create an\u00a0ignite-storm.properties<\/i> file to add the cache name, tuple name and the name of the Ignite configuration as shown below.<\/p>\n
      cache.name=testCache\r\ntuple.name=ignite\r\nignite.spring.xml=example-ignite.xml<\/pre>\n
      Step 4:<\/b><\/p>\n
      Next, create FileSourceSpout Java class as shown below,<\/p>\n
      public class FileSourceSpout extends BaseRichSpout {\r\n  private static final Logger LOGGER = LogManager.getLogger(FileSourceSpout.class);\r\n  private SpoutOutputCollector outputCollector;\r\n  @Override\r\n  public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {\r\n  this.outputCollector = spoutOutputCollector;\r\n  }\r\n@Override\r\n  public void nextTuple() {\r\n  try {\r\n  Path filePath = Paths.get(this.getClass().getClassLoader().getResource(\"source.csv\").toURI());\r\n  try(Stream<String> lines = Files.lines(filePath)){\r\n  lines.forEach(line ->{\r\n  outputCollector.emit(new Values(line));\r\n  });\r\n  } catch(IOException e){\r\n  LOGGER.error(e.getMessage());\r\n  }\r\n  } catch (URISyntaxException e) {\r\n  LOGGER.error(e.getMessage());\r\n  }\r\n  }\r\n  @Override\r\n  public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {\r\n  outputFieldsDeclarer.declare(new Fields(\"trafficLog\"));\r\n  }\r\n}<\/pre>\n
      The\u00a0FileSourceSpout<\/i> code has three important methods<\/p>\n