 <!--
 ▄▄▄       ██▓███   ▄▄▄       ▄████▄   ██░ ██ ▓█████     ██▓  ▄████  ███▄    █  ██▓▄▄▄█████▓▓█████
▒████▄    ▓██░  ██▒▒████▄    ▒██▀ ▀█  ▓██░ ██▒▓█   ▀    ▓██▒ ██▒ ▀█▒ ██ ▀█   █ ▓██▒▓  ██▒ ▓▒▓█   ▀
▒██  ▀█▄  ▓██░ ██▓▒▒██  ▀█▄  ▒▓█    ▄ ▒██▀▀██░▒███      ▒██▒▒██░▄▄▄░▓██  ▀█ ██▒▒██▒▒ ▓██░ ▒░▒███
░██▄▄▄▄██ ▒██▄█▓▒ ▒░██▄▄▄▄██ ▒▓▓▄ ▄██▒░▓█ ░██ ▒▓█  ▄    ░██░░▓█  ██▓▓██▒  ▐▌██▒░██░░ ▓██▓ ░ ▒▓█  ▄
 ▓█   ▓██▒▒██▒ ░  ░ ▓█   ▓██▒▒ ▓███▀ ░░▓█▒░██▓░▒████▒   ░██░░▒▓███▀▒▒██░   ▓██░░██░  ▒██▒ ░ ░▒████▒
 ▒▒   ▓▒█░▒▓▒░ ░  ░ ▒▒   ▓▒█░░ ░▒ ▒  ░ ▒ ░░▒░▒░░ ▒░ ░   ░▓   ░▒   ▒ ░ ▒░   ▒ ▒ ░▓    ▒ ░░   ░░ ▒░ ░
  ▒   ▒▒ ░░▒ ░       ▒   ▒▒ ░  ░  ▒    ▒ ░▒░ ░ ░ ░  ░    ▒ ░  ░   ░ ░ ░░   ░ ▒░ ▒ ░    ░     ░ ░  ░
  ░   ▒   ░░         ░   ▒   ░         ░  ░░ ░   ░       ▒ ░░ ░   ░    ░   ░ ░  ▒ ░  ░         ░
      ░  ░               ░  ░░ ░       ░  ░  ░   ░  ░    ░        ░          ░  ░              ░  ░
-->

<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
-->

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="Cache-Control" content="no-cache, no-store, must-revalidate" />
    <meta http-equiv="Pragma" content="no-cache" />
    <meta http-equiv="Expires" content="0" />
    <title>Apache Ignite - Streaming and CEP</title>
    <link media="all" rel="stylesheet" href="/css/all.css">
    <link href="https://netdna.bootstrapcdn.com/font-awesome/4.2.0/css/font-awesome.css" rel="stylesheet">
    <link href='https://fonts.googleapis.com/css?family=Open+Sans:400,300,300italic,400italic,600,600italic,700,700italic,800,800italic' rel='stylesheet' type='text/css'>

    <!--#include virtual="/includes/sh.html" -->
</head>
<body>
<div id="wrapper">
    <!--#include virtual="/includes/header.html" -->

    <main id="main" role="main" class="container">
        <section id="streaming" class="page-section">
            <div class="col-sm-12 col-md-12 col-xs-12" style="padding:0 0 20px 0;">
                <div class="col-sm-6 col-md-7 col-xs-12" style="padding-left:0; padding-right:0">
                    <h2 class="first">Streaming & CEP</h2>
                    <p>
                        Ignite streaming allows to process continuous never-ending streams of data in scalable
                        and fault-tolerant fashion. The rates at which data can be injected into Ignite can be very
                        high and easily exceed millions of events per second on a moderately sized cluster.
                    </p>
                    <p>
                    <div class="page-heading">How it Works:</div>
                    <ol class="page-list">
                        <li>Clients inject streams of data into Ignite.</li>
                        <li>Data is automatically partitioned between Ignite data nodes.</li>
                        <li>Data is concurrently processed in <code>sliding windows</code>.</li>
                        <li>Clients perform concurrent <code>SQL queries</code> on the streamed data.</li>
                        <li>Clients subscribe to <code>continuous queries</code> as data changes.</li>
                    </ol>
                    </p>
                </div>

                <div class="col-sm-6 col-md-5 col-xs-12" style="padding-right:0">
                    <img class="first img-responsive" src="/images/ignite-stream-query.png" width="382px" style="float:right;"/>
                    <br/>
                    <img class="img-responsive" src="/images/in_memory_streaming.png" width="280px" style="margin: auto;">
                </div>
            </div>
            <div class="code-examples">
                <div class="page-heading">Code Examples:</div>
                <!-- Nav tabs -->
                <ul id="datagrid-examples" class="nav nav-tabs">
                    <li class="active"><a href="#streaming-data" role="tab" data-toggle="tab">Stream Data</a></li>
                    <li><a href="#streaming-transformer" role="tab" data-toggle="tab">Transform Data</a></li>
                    <li><a href="#streaming-query" role="tab" data-toggle="tab">Query Sliding Windows</a></li>
                </ul>

                <!-- Tab panes -->
                <div class="tab-content">
                    <div role="tabpanel" class="tab-pane active" id="streaming-data">
                        <pre class="brush:java">
                            // Get the data streamer reference and stream data.
                            try (IgniteDataStreamer&lt;Integer, String&gt; stmr = ignite.dataStreamer("myStreamCache")) {
                                // Stream entries.
                                for (int i = 0; i &lt; 100000; i++)
                                    stmr.addData(i, Integer.toString(i));
                            }
                        </pre>
                    </div>
                    <div role="tabpanel" class="tab-pane" id="streaming-transformer">
                        <pre class="brush:java">
                            CacheConfiguration cfg = new CacheConfiguration("wordCountCache");

                            IgniteCache&lt;Integer, Long&gt; stmCache = ignite.getOrCreateCache(cfg);

                            try (IgniteDataStreamer&lt;String, Long&gt; stmr = ignite.dataStreamer(stmCache.getName())) {
                              // Allow data updates.
                              stmr.allowOverwrite(true);

                              // Configure data transformation to count instances of the same word.
                              stmr.receiver(StreamTransformer.from((e, arg) -> {
                                // Get current count.
                                Long val = e.getValue();

                                // Increment count by 1.
                                e.setValue(val == null ? 1L : val + 1);

                                return null;
                              }));

                              // Stream words into the streamer cache.
                              for (String word : text)
                                stmr.addData(word, 1L);
                            }
                        </pre>
                    </div>
                    <div role="tabpanel" class="tab-pane" id="streaming-query">
                        <pre class="brush:java">
                            CacheConfiguration&lt;String, Instrument&gt; cfg = new CacheConfiguration<>("instCache");

                            // LRU window holding 1,000,000 entries.
                            cfg.setEvictionPolicyFactory(new LruEvictionPolicy(1_000_000));

                            // Index some fields for querying portfolio positions.
                            cfg.setIndexedTypes(String.class, Instrument.class);

                            // Get a handle on the cache (create it if necessary).
                            IgniteCache&lt;String, Instrument&lt; instCache = ignite.getOrCreateCache(cfg);

                            // Select top 3 best performing instruments from the sliding window.
                            SqlFieldsQuery top3qry = new SqlFieldsQuery(
                              "select symbol, (latestPrice - openPrice) as change "
                                + "from Instrument "
                                + "order by change "
                                + "desc limit 3"
                            );

                            // List of rows. Every row is represented as a List as well.
                            List&lt;List&lt;?&gt;&gt; top3 = instCache.query(top3qry).getAll();                        </pre>
                    </div>
                </div>
            </div>
            <div class="page-heading">GitHub Examples:</div>
            <p>
                Also see <a href="https://github.com/apache/incubator-ignite/blob/master/examples/src/main/java/org/apache/ignite/examples/datagrid/CacheContinuousQueryExample.java" target="github">continuous queries</a>,
                <a href="https://github.com/apache/incubator-ignite/tree/master/examples/src/main/java/org/apache/ignite/examples/streaming/wordcount" target="github">word count</a>,
                and <a href="https://github.com/apache/incubator-ignite/tree/master/examples/src/main/java/org/apache/ignite/examples/streaming" target="github">other streaming examples</a> available on GitHub.
            </p>
        </section>
        <section id="key-features" class="page-section">
            <h2>Streaming Features</h2>
            <table class="formatted" name="Streaming Features">
                <thead>
                    <tr>
                        <th width="35%" class="left">Feature</th>
                        <th>Description</th>
                    </tr>
                </thead>
                <tbody>
                    <tr>
                        <td class="left">Data Streamers</td>
                        <td>
                            <p>
                                Data streamers are defined by <code>IgniteDataStreamer</code> API and are built to inject large
                                amounts of continuous streams of data into Ignite stream caches. Data streamers are
                                built in a scalable and fault-tolerant fashion and provide <b>at-least-once-guarantee</b>
                                semantics for all the data streamed into Ignite.
                            </p>
                            <div class="page-links">
                                <a href="http://apacheignite.readme.io/docs/data-streamers" target="docs">Docs for this Feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Collocated Processing</td>
                        <td>
                            <p>
                                For cases when you need to execute some custom logic instead of just adding new data,
                                you can take advantage of <code>StreamReceiver</code> API.
                            </p>
                            <p>
                                Stream receivers allow you to react to the streamed data in collocated fashion,
                                directly on the nodes where it will be cached. You can change the data or add any
                                custom pre-processing logic to it, before putting the data into cache.
                            </p>
                            <div class="page-links">
                                <a href="http://apacheignite.readme.io/docs/data-streamers#streamreceiver" target="docs">Docs for this Feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Sliding Windows</td>
                        <td>
                            <p>
                                Ignite streaming functionality also allows to query into sliding windows of data.
                                Sliding windows are configured as Ignite cache eviction policies, and can be
                                <code>time-based</code>, <code>size-based</code>, or <code>batch-based</code>. You can configure one sliding-window per cache. However,
                                you can easily define more than one cache if you need different sliding windows for the
                                same data.
                            </p>
                            <div class="page-links">
                                <a href="http://apacheignite.readme.io/docs/sliding-windows" target="docs">Docs for this Feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Querying Sliding Windows</td>
                        <td>
                            <p>
                                You can use full set of Ignite data indexing capabilities, together with
                                Ignite SQL, TEXT, and Predicate based cache queries, to query into the streaming data.
                            </p>
                            <div class="page-links">
                                <a href="http://apacheignite.readme.io/docs/cache-queries" target="docs">Docs for this Feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Continuous Queries</td>
                        <td>
                            <p>
                                Continuous queries are useful for cases when you want to execute a query and then
                                continue to get notified about the data changes that fall into your query filter.
                            </p>
                            <div class="page-links">
                                <a href="http://apacheignite.readme.io/docs/continuous-queries" target="docs">Docs for this Feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">JMS Data Streamer</td>
                        <td>
                            <p>
                                Ignite JMS Data Streamer consumes messages from JMS brokers and inserts them into Ignite caches.
                            </p>
                            <div  class="page-links">
                                <a href="https://apacheignite.readme.io/docs/jms-data-streamer" target="docs">Docs for this feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Apache Flume Sink</td>
                        <td>
                            <p>
                                IgniteSink is a Flume sink that extracts events from an associated Flume channel and injects into an Ignite cache.
                            </p>
                            <div  class="page-links">
                                <a href="https://apacheignite.readme.io/docs/flume-data-streamer" target="docs">Docs for this feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">MQTT Streamer</td>
                        <td>
                            <p>
                                Ignite MQTT Streamer consumes messages from a MQTT topic and feeds transformed key-value pairs into an Ignite cache.
                            </p>
                            <div  class="page-links">
                                <a href="https://apacheignite.readme.io/docs/mqtt-streamer" target="docs">Docs for this feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Twitter Streamer</td>
                        <td>
                            <p>
                                Ignite Twitter Streamer consumes messages from a Twitter Streaming API and inserts them into an Ignite cache.
                            </p>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Apache Kafka Streamer</td>
                        <td>
                            <p>
                                Ignite Kafka Data Streamer consumes messages for a given Kafka Topic from Kafka Broker and inserts them into an Ignite cache.
                            </p>
                            <div  class="page-links">
                                <a href="https://apacheignite.readme.io/docs/kafka-streamer" target="docs">Docs for this feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Apache Camel streamer</td>
                        <td>
                            <p>
                                Ignite Camel streamer consumes messages from an Apache Camel consumer endpoint and feeds them into an Ignite cache.
                            </p>
                            <div  class="page-links">
                                <a href="https://apacheignite.readme.io/docs/camel-streamer" target="docs">Docs for this feature <i class="fa fa-angle-double-right"></i></a>
                            </div>
                        </td>
                    </tr>
                    <tr>
                        <td class="left">Apache Storm Streamer</td>
                        <td>
                            <p>
                                Ignite Storm Streamer consumes messages from an Apache Storm consumer endpoint and feeds them into an Ignite cache.
                            </p>
                        </td>
                    </tr>
                </tbody>
            </table>
        </section>
    </main>

    <!--#include virtual="/includes/footer.html" -->
</div>
<!--#include virtual="/includes/scripts.html" -->
</body>
</html>