Updating apex-3.3 docs with latest mkdocs processor which fixes search results bug
diff --git a/docs/apex-3.3/__init__.py b/docs/apex-3.3/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/docs/apex-3.3/__init__.py
diff --git a/docs/apex-3.3/__init__.pyc b/docs/apex-3.3/__init__.pyc
new file mode 100644
index 0000000..f478a23
--- /dev/null
+++ b/docs/apex-3.3/__init__.pyc
Binary files differ
diff --git a/docs/apex-3.3/apex_development_setup/index.html b/docs/apex-3.3/apex_development_setup/index.html
index 074fc50..60350b2 100644
--- a/docs/apex-3.3/apex_development_setup/index.html
+++ b/docs/apex-3.3/apex_development_setup/index.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>Development Setup - Apache Apex Documentation</title>
@@ -22,6 +25,8 @@
<script>
// Current page data
var mkdocs_page_name = "Development Setup";
+ var mkdocs_page_input_path = "apex_development_setup.md";
+ var mkdocs_page_url = "/apex_development_setup/";
</script>
<script src="../js/jquery-2.1.1.min.js"></script>
@@ -284,7 +289,7 @@
<div class="rst-footer-buttons" role="navigation" aria-label="footer navigation">
- <a href="../application_development/" class="btn btn-neutral float-right" title="Applications"/>Next <span class="icon icon-circle-arrow-right"></span></a>
+ <a href="../application_development/" class="btn btn-neutral float-right" title="Applications">Next <span class="icon icon-circle-arrow-right"></span></a>
<a href=".." class="btn btn-neutral" title="Apache Apex"><span class="icon icon-circle-arrow-left"></span> Previous</a>
diff --git a/docs/apex-3.3/application_development/index.html b/docs/apex-3.3/application_development/index.html
index 8b36471..5297074 100644
--- a/docs/apex-3.3/application_development/index.html
+++ b/docs/apex-3.3/application_development/index.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>Applications - Apache Apex Documentation</title>
@@ -22,6 +25,8 @@
<script>
// Current page data
var mkdocs_page_name = "Applications";
+ var mkdocs_page_input_path = "application_development.md";
+ var mkdocs_page_url = "/application_development/";
</script>
<script src="../js/jquery-2.1.1.min.js"></script>
@@ -2504,7 +2509,7 @@
<div class="rst-footer-buttons" role="navigation" aria-label="footer navigation">
- <a href="../application_packages/" class="btn btn-neutral float-right" title="Packages"/>Next <span class="icon icon-circle-arrow-right"></span></a>
+ <a href="../application_packages/" class="btn btn-neutral float-right" title="Packages">Next <span class="icon icon-circle-arrow-right"></span></a>
<a href="../apex_development_setup/" class="btn btn-neutral" title="Development Setup"><span class="icon icon-circle-arrow-left"></span> Previous</a>
diff --git a/docs/apex-3.3/application_packages/index.html b/docs/apex-3.3/application_packages/index.html
index d212cbe..41e9a29 100644
--- a/docs/apex-3.3/application_packages/index.html
+++ b/docs/apex-3.3/application_packages/index.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>Packages - Apache Apex Documentation</title>
@@ -22,6 +25,8 @@
<script>
// Current page data
var mkdocs_page_name = "Packages";
+ var mkdocs_page_input_path = "application_packages.md";
+ var mkdocs_page_url = "/application_packages/";
</script>
<script src="../js/jquery-2.1.1.min.js"></script>
@@ -733,7 +738,7 @@
<div class="rst-footer-buttons" role="navigation" aria-label="footer navigation">
- <a href="../operator_development/" class="btn btn-neutral float-right" title="Operators"/>Next <span class="icon icon-circle-arrow-right"></span></a>
+ <a href="../operator_development/" class="btn btn-neutral float-right" title="Operators">Next <span class="icon icon-circle-arrow-right"></span></a>
<a href="../application_development/" class="btn btn-neutral" title="Applications"><span class="icon icon-circle-arrow-left"></span> Previous</a>
diff --git a/docs/apex-3.3/autometrics/index.html b/docs/apex-3.3/autometrics/index.html
index 54e0ec5..6fd3fd8 100644
--- a/docs/apex-3.3/autometrics/index.html
+++ b/docs/apex-3.3/autometrics/index.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>AutoMetric API - Apache Apex Documentation</title>
@@ -22,6 +25,8 @@
<script>
// Current page data
var mkdocs_page_name = "AutoMetric API";
+ var mkdocs_page_input_path = "autometrics.md";
+ var mkdocs_page_url = "/autometrics/";
</script>
<script src="../js/jquery-2.1.1.min.js"></script>
@@ -338,7 +343,7 @@
<div class="rst-footer-buttons" role="navigation" aria-label="footer navigation">
- <a href="../dtcli/" class="btn btn-neutral float-right" title="dtCli"/>Next <span class="icon icon-circle-arrow-right"></span></a>
+ <a href="../dtcli/" class="btn btn-neutral float-right" title="dtCli">Next <span class="icon icon-circle-arrow-right"></span></a>
<a href="../operator_development/" class="btn btn-neutral" title="Operators"><span class="icon icon-circle-arrow-left"></span> Previous</a>
diff --git a/docs/apex-3.3/base.html b/docs/apex-3.3/base.html
new file mode 100644
index 0000000..87c9f8a
--- /dev/null
+++ b/docs/apex-3.3/base.html
@@ -0,0 +1,118 @@
+<!DOCTYPE html>
+<!--[if IE 8]><html class="no-js lt-ie9" lang="en" > <![endif]-->
+<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
+<head>
+ <meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
+ <meta name="viewport" content="width=device-width, initial-scale=1.0">
+ {% if page_description %}<meta name="description" content="{{ page_description }}">{% endif %}
+ {% if site_author %}<meta name="author" content="{{ site_author }}">{% endif %}
+ {% block htmltitle %}
+ <title>{% if page_title %}{{ page_title }} - {% endif %}{{ site_name }}</title>
+ {% endblock %}
+
+ {% if favicon %}<link rel="shortcut icon" href="{{ favicon }}">
+ {% else %}<link rel="shortcut icon" href="{{ base_url }}/img/favicon.ico">{% endif %}
+
+ {# CSS #}
+ <link href='https://fonts.googleapis.com/css?family=Lato:400,700|Roboto+Slab:400,700|Inconsolata:400,700' rel='stylesheet' type='text/css'>
+
+ <link rel="stylesheet" href="{{ base_url }}/css/theme.css" type="text/css" />
+ <link rel="stylesheet" href="{{ base_url }}/css/theme_extra.css" type="text/css" />
+ <link rel="stylesheet" href="{{ base_url }}/css/highlight.css">
+ {%- for path in extra_css %}
+ <link href="{{ path }}" rel="stylesheet">
+ {%- endfor %}
+
+ {% if current_page %}
+ <script>
+ // Current page data
+ var mkdocs_page_name = {{ page_title|tojson|safe }};
+ var mkdocs_page_input_path = {{ current_page.input_path|tojson|safe }};
+ var mkdocs_page_url = {{ current_page.abs_url|tojson|safe }};
+ </script>
+ {% endif %}
+ <script src="{{ base_url }}/js/jquery-2.1.1.min.js"></script>
+ <script src="{{ base_url }}/js/modernizr-2.8.3.min.js"></script>
+ <script type="text/javascript" src="{{ base_url }}/js/highlight.pack.js"></script>
+ <script src="{{ base_url }}/js/theme.js"></script>
+
+ {%- block extrahead %} {% endblock %}
+
+ {%- for path in extra_javascript %}
+ <script src="{{ path }}"></script>
+ {%- endfor %}
+
+ {% if google_analytics %}
+ <script>
+ (function(i,s,o,g,r,a,m){i['GoogleAnalyticsObject']=r;i[r]=i[r]||function(){
+ (i[r].q=i[r].q||[]).push(arguments)},i[r].l=1*new Date();a=s.createElement(o),
+ m=s.getElementsByTagName(o)[0];a.async=1;a.src=g;m.parentNode.insertBefore(a,m)
+ })(window,document,'script','//www.google-analytics.com/analytics.js','ga');
+
+ ga('create', '{{ google_analytics[0] }}', '{{ google_analytics[1] }}');
+ ga('send', 'pageview');
+ </script>
+ {% endif %}
+</head>
+
+<body class="wy-body-for-nav" role="document">
+
+ <div class="wy-grid-for-nav">
+
+ {# SIDE NAV, TOGGLES ON MOBILE #}
+ <nav data-toggle="wy-nav-shift" class="wy-nav-side stickynav">
+ <div class="wy-side-nav-search">
+ <a href="{{ homepage_url }}" class="icon icon-home"> {{ site_name }}</a>
+ {% include "searchbox.html" %}
+ </div>
+
+ <div class="wy-menu wy-menu-vertical" data-spy="affix" role="navigation" aria-label="main navigation">
+ <ul class="current">
+ {% for nav_item in nav %}
+ <li>{% include "toc.html" %}<li>
+ {% endfor %}
+ </ul>
+ </div>
+
+ </nav>
+
+ <section data-toggle="wy-nav-shift" class="wy-nav-content-wrap">
+
+ {# MOBILE NAV, TRIGGLES SIDE NAV ON TOGGLE #}
+ <nav class="wy-nav-top" role="navigation" aria-label="top navigation">
+ <i data-toggle="wy-nav-top" class="fa fa-bars"></i>
+ <a href="{{ homepage_url }}">{{ site_name }}</a>
+ </nav>
+
+ {# PAGE CONTENT #}
+ <div class="wy-nav-content">
+ <div class="rst-content">
+ {% include "breadcrumbs.html" %}
+ <div role="main">
+ <div class="section">
+ {% block content %}
+ {{ content }}
+ {% endblock %}
+ </div>
+ </div>
+ {%- block footer %}
+ {% include "footer.html" %}
+ {% endblock %}
+ </div>
+ </div>
+
+ </section>
+
+ </div>
+
+{% include "versions.html" %}
+
+</body>
+</html>
+{% if current_page and current_page.is_homepage %}
+<!--
+MkDocs version : {{ mkdocs_version }}
+Build Date UTC : {{ build_date_utc }}
+-->
+{% endif %}
diff --git a/docs/apex-3.3/breadcrumbs.html b/docs/apex-3.3/breadcrumbs.html
new file mode 100644
index 0000000..01960e6
--- /dev/null
+++ b/docs/apex-3.3/breadcrumbs.html
@@ -0,0 +1,25 @@
+<div role="navigation" aria-label="breadcrumbs navigation">
+ <ul class="wy-breadcrumbs">
+ <li><a href="{{ homepage_url }}">Docs</a> »</li>
+ {% if current_page %}
+ {% for doc in current_page.ancestors %}
+ {% if doc.link %}
+ <li><a href="{{ doc.link|e }}">{{ doc.title }}</a> »</li>
+ {% else %}
+ <li>{{ doc.title }} »</li>
+ {% endif %}
+ {% endfor %}
+ {% endif %}
+ {% if current_page %}<li>{{ current_page.title }}</li>{% endif %}
+ <li class="wy-breadcrumbs-aside">
+ {% if repo_url %}
+ {% if repo_name == 'GitHub' %}
+ <a href="{{ repo_url }}" class="icon icon-github"> Edit on GitHub</a>
+ {% elif repo_name == 'Bitbucket' %}
+ <a href="{{ repo_url }}" class="icon icon-bitbucket"> Edit on BitBucket</a>
+ {% endif %}
+ {% endif %}
+ </li>
+ </ul>
+ <hr/>
+</div>
diff --git a/docs/apex-3.3/css/highlight.css b/docs/apex-3.3/css/highlight.css
index 0375453..0ae40a7 100644
--- a/docs/apex-3.3/css/highlight.css
+++ b/docs/apex-3.3/css/highlight.css
@@ -8,7 +8,6 @@
.hljs {
display: block;
overflow-x: auto;
- padding: 0.5em;
color: #333;
-webkit-text-size-adjust: none;
}
diff --git a/docs/apex-3.3/css/theme_extra.css b/docs/apex-3.3/css/theme_extra.css
index ccb384e..9845d00 100644
--- a/docs/apex-3.3/css/theme_extra.css
+++ b/docs/apex-3.3/css/theme_extra.css
@@ -1,16 +1,4 @@
/*
- * Tweak the overal size to better match RTD.
- */
-body {
- font-size: 90%;
-}
-
-h3, h4, h5, h6 {
- color: #2980b9;
- font-weight: 300
-}
-
-/*
* Sphinx doesn't have support for section dividers like we do in
* MkDocs, this styles the section titles in the nav
*
@@ -57,6 +45,7 @@
*/
code {
white-space: pre;
+ padding: 2px 5px;
}
/*
@@ -69,6 +58,33 @@
word-wrap: break-word;
}
+/**
+ * Make code blocks display as blocks and give them the appropriate
+ * font size and padding.
+ *
+ * https://github.com/mkdocs/mkdocs/issues/855
+ */
+pre code {
+ display: block;
+ padding: 12px;
+ font-size: 12px;
+}
+
+/*
+ * Fix link colors when the link text is inline code.
+ *
+ * https://github.com/mkdocs/mkdocs/issues/718
+ */
+a code {
+ color: #2980B9;
+}
+a:hover code {
+ color: #3091d1;
+}
+a:visited code {
+ color: #9B59B6;
+}
+
/*
* The CSS classes from highlight.js seem to clash with the
* ReadTheDocs theme causing some code to be incorrectly made
@@ -124,3 +140,15 @@
.wy-menu-vertical li ul.subnav ul.subnav{
padding-left: 1em;
}
+
+
+/*
+ * Improve inline code blocks within admonitions.
+ *
+ * https://github.com/mkdocs/mkdocs/issues/656
+ */
+ div.admonition code {
+ color: #404040;
+ border: 1px solid rgba(0, 0, 0, 0.2);
+ background: rgba(255, 255, 255, 0.7);
+}
diff --git a/docs/apex-3.3/dtcli/index.html b/docs/apex-3.3/dtcli/index.html
index 4d8fddd..fc33b6f 100644
--- a/docs/apex-3.3/dtcli/index.html
+++ b/docs/apex-3.3/dtcli/index.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>dtCli - Apache Apex Documentation</title>
@@ -22,6 +25,8 @@
<script>
// Current page data
var mkdocs_page_name = "dtCli";
+ var mkdocs_page_input_path = "dtcli.md";
+ var mkdocs_page_url = "/dtcli/";
</script>
<script src="../js/jquery-2.1.1.min.js"></script>
diff --git a/docs/apex-3.3/footer.html b/docs/apex-3.3/footer.html
new file mode 100644
index 0000000..1103193
--- /dev/null
+++ b/docs/apex-3.3/footer.html
@@ -0,0 +1,23 @@
+<footer>
+ {% if next_page or previous_page %}
+ <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation">
+ {% if next_page %}
+ <a href="{{ next_page.url }}" class="btn btn-neutral float-right" title="{{ next_page.title }}">Next <span class="icon icon-circle-arrow-right"></span></a>
+ {% endif %}
+ {% if previous_page %}
+ <a href="{{ previous_page.url }}" class="btn btn-neutral" title="{{ previous_page.title }}"><span class="icon icon-circle-arrow-left"></span> Previous</a>
+ {% endif %}
+ </div>
+ {% endif %}
+
+ <hr/>
+
+ <div role="contentinfo">
+ <!-- Copyright etc -->
+ {% if copyright %}
+ <p>{{ copyright }}</p>
+ {% endif %}
+ </div>
+
+ Built with <a href="http://www.mkdocs.org">MkDocs</a> using a <a href="https://github.com/snide/sphinx_rtd_theme">theme</a> provided by <a href="https://readthedocs.org">Read the Docs</a>.
+</footer>
diff --git a/docs/apex-3.3/index.html b/docs/apex-3.3/index.html
index 17caf9d..95952ca 100644
--- a/docs/apex-3.3/index.html
+++ b/docs/apex-3.3/index.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>Apache Apex Documentation</title>
@@ -21,7 +24,9 @@
<script>
// Current page data
- var mkdocs_page_name = "None";
+ var mkdocs_page_name = null;
+ var mkdocs_page_input_path = "index.md";
+ var mkdocs_page_url = "/";
</script>
<script src="./js/jquery-2.1.1.min.js"></script>
@@ -175,7 +180,7 @@
<div class="rst-footer-buttons" role="navigation" aria-label="footer navigation">
- <a href="apex_development_setup/" class="btn btn-neutral float-right" title="Development Setup"/>Next <span class="icon icon-circle-arrow-right"></span></a>
+ <a href="apex_development_setup/" class="btn btn-neutral float-right" title="Development Setup">Next <span class="icon icon-circle-arrow-right"></span></a>
</div>
@@ -212,6 +217,6 @@
</html>
<!--
-MkDocs version : 0.14.0
-Build Date UTC : 2016-03-05 00:37:23.777568
+MkDocs version : 0.15.3
+Build Date UTC : 2016-03-10 00:39:42.856605
-->
diff --git a/docs/apex-3.3/mkdocs/js/lunr-0.5.7.min.js b/docs/apex-3.3/mkdocs/js/lunr-0.5.7.min.js
deleted file mode 100644
index b72449a..0000000
--- a/docs/apex-3.3/mkdocs/js/lunr-0.5.7.min.js
+++ /dev/null
@@ -1,7 +0,0 @@
-/**
- * lunr - http://lunrjs.com - A bit like Solr, but much smaller and not as bright - 0.5.7
- * Copyright (C) 2014 Oliver Nightingale
- * MIT Licensed
- * @license
- */
-!function(){var t=function(e){var n=new t.Index;return n.pipeline.add(t.trimmer,t.stopWordFilter,t.stemmer),e&&e.call(n,n),n};t.version="0.5.7",t.utils={},t.utils.warn=function(t){return function(e){t.console&&console.warn&&console.warn(e)}}(this),t.EventEmitter=function(){this.events={}},t.EventEmitter.prototype.addListener=function(){var t=Array.prototype.slice.call(arguments),e=t.pop(),n=t;if("function"!=typeof e)throw new TypeError("last argument must be a function");n.forEach(function(t){this.hasHandler(t)||(this.events[t]=[]),this.events[t].push(e)},this)},t.EventEmitter.prototype.removeListener=function(t,e){if(this.hasHandler(t)){var n=this.events[t].indexOf(e);this.events[t].splice(n,1),this.events[t].length||delete this.events[t]}},t.EventEmitter.prototype.emit=function(t){if(this.hasHandler(t)){var e=Array.prototype.slice.call(arguments,1);this.events[t].forEach(function(t){t.apply(void 0,e)})}},t.EventEmitter.prototype.hasHandler=function(t){return t in this.events},t.tokenizer=function(t){if(!arguments.length||null==t||void 0==t)return[];if(Array.isArray(t))return t.map(function(t){return t.toLowerCase()});for(var e=t.toString().replace(/^\s+/,""),n=e.length-1;n>=0;n--)if(/\S/.test(e.charAt(n))){e=e.substring(0,n+1);break}return e.split(/(?:\s+|\-)/).filter(function(t){return!!t}).map(function(t){return t.toLowerCase()})},t.Pipeline=function(){this._stack=[]},t.Pipeline.registeredFunctions={},t.Pipeline.registerFunction=function(e,n){n in this.registeredFunctions&&t.utils.warn("Overwriting existing registered function: "+n),e.label=n,t.Pipeline.registeredFunctions[e.label]=e},t.Pipeline.warnIfFunctionNotRegistered=function(e){var n=e.label&&e.label in this.registeredFunctions;n||t.utils.warn("Function is not registered with pipeline. This may cause problems when serialising the index.\n",e)},t.Pipeline.load=function(e){var n=new t.Pipeline;return e.forEach(function(e){var i=t.Pipeline.registeredFunctions[e];if(!i)throw new Error("Cannot load un-registered function: "+e);n.add(i)}),n},t.Pipeline.prototype.add=function(){var e=Array.prototype.slice.call(arguments);e.forEach(function(e){t.Pipeline.warnIfFunctionNotRegistered(e),this._stack.push(e)},this)},t.Pipeline.prototype.after=function(e,n){t.Pipeline.warnIfFunctionNotRegistered(n);var i=this._stack.indexOf(e)+1;this._stack.splice(i,0,n)},t.Pipeline.prototype.before=function(e,n){t.Pipeline.warnIfFunctionNotRegistered(n);var i=this._stack.indexOf(e);this._stack.splice(i,0,n)},t.Pipeline.prototype.remove=function(t){var e=this._stack.indexOf(t);this._stack.splice(e,1)},t.Pipeline.prototype.run=function(t){for(var e=[],n=t.length,i=this._stack.length,o=0;n>o;o++){for(var r=t[o],s=0;i>s&&(r=this._stack[s](r,o,t),void 0!==r);s++);void 0!==r&&e.push(r)}return e},t.Pipeline.prototype.reset=function(){this._stack=[]},t.Pipeline.prototype.toJSON=function(){return this._stack.map(function(e){return t.Pipeline.warnIfFunctionNotRegistered(e),e.label})},t.Vector=function(){this._magnitude=null,this.list=void 0,this.length=0},t.Vector.Node=function(t,e,n){this.idx=t,this.val=e,this.next=n},t.Vector.prototype.insert=function(e,n){var i=this.list;if(!i)return this.list=new t.Vector.Node(e,n,i),this.length++;for(var o=i,r=i.next;void 0!=r;){if(e<r.idx)return o.next=new t.Vector.Node(e,n,r),this.length++;o=r,r=r.next}return o.next=new t.Vector.Node(e,n,r),this.length++},t.Vector.prototype.magnitude=function(){if(this._magniture)return this._magnitude;for(var t,e=this.list,n=0;e;)t=e.val,n+=t*t,e=e.next;return this._magnitude=Math.sqrt(n)},t.Vector.prototype.dot=function(t){for(var e=this.list,n=t.list,i=0;e&&n;)e.idx<n.idx?e=e.next:e.idx>n.idx?n=n.next:(i+=e.val*n.val,e=e.next,n=n.next);return i},t.Vector.prototype.similarity=function(t){return this.dot(t)/(this.magnitude()*t.magnitude())},t.SortedSet=function(){this.length=0,this.elements=[]},t.SortedSet.load=function(t){var e=new this;return e.elements=t,e.length=t.length,e},t.SortedSet.prototype.add=function(){Array.prototype.slice.call(arguments).forEach(function(t){~this.indexOf(t)||this.elements.splice(this.locationFor(t),0,t)},this),this.length=this.elements.length},t.SortedSet.prototype.toArray=function(){return this.elements.slice()},t.SortedSet.prototype.map=function(t,e){return this.elements.map(t,e)},t.SortedSet.prototype.forEach=function(t,e){return this.elements.forEach(t,e)},t.SortedSet.prototype.indexOf=function(t,e,n){var e=e||0,n=n||this.elements.length,i=n-e,o=e+Math.floor(i/2),r=this.elements[o];return 1>=i?r===t?o:-1:t>r?this.indexOf(t,o,n):r>t?this.indexOf(t,e,o):r===t?o:void 0},t.SortedSet.prototype.locationFor=function(t,e,n){var e=e||0,n=n||this.elements.length,i=n-e,o=e+Math.floor(i/2),r=this.elements[o];if(1>=i){if(r>t)return o;if(t>r)return o+1}return t>r?this.locationFor(t,o,n):r>t?this.locationFor(t,e,o):void 0},t.SortedSet.prototype.intersect=function(e){for(var n=new t.SortedSet,i=0,o=0,r=this.length,s=e.length,a=this.elements,h=e.elements;;){if(i>r-1||o>s-1)break;a[i]!==h[o]?a[i]<h[o]?i++:a[i]>h[o]&&o++:(n.add(a[i]),i++,o++)}return n},t.SortedSet.prototype.clone=function(){var e=new t.SortedSet;return e.elements=this.toArray(),e.length=e.elements.length,e},t.SortedSet.prototype.union=function(t){var e,n,i;return this.length>=t.length?(e=this,n=t):(e=t,n=this),i=e.clone(),i.add.apply(i,n.toArray()),i},t.SortedSet.prototype.toJSON=function(){return this.toArray()},t.Index=function(){this._fields=[],this._ref="id",this.pipeline=new t.Pipeline,this.documentStore=new t.Store,this.tokenStore=new t.TokenStore,this.corpusTokens=new t.SortedSet,this.eventEmitter=new t.EventEmitter,this._idfCache={},this.on("add","remove","update",function(){this._idfCache={}}.bind(this))},t.Index.prototype.on=function(){var t=Array.prototype.slice.call(arguments);return this.eventEmitter.addListener.apply(this.eventEmitter,t)},t.Index.prototype.off=function(t,e){return this.eventEmitter.removeListener(t,e)},t.Index.load=function(e){e.version!==t.version&&t.utils.warn("version mismatch: current "+t.version+" importing "+e.version);var n=new this;return n._fields=e.fields,n._ref=e.ref,n.documentStore=t.Store.load(e.documentStore),n.tokenStore=t.TokenStore.load(e.tokenStore),n.corpusTokens=t.SortedSet.load(e.corpusTokens),n.pipeline=t.Pipeline.load(e.pipeline),n},t.Index.prototype.field=function(t,e){var e=e||{},n={name:t,boost:e.boost||1};return this._fields.push(n),this},t.Index.prototype.ref=function(t){return this._ref=t,this},t.Index.prototype.add=function(e,n){var i={},o=new t.SortedSet,r=e[this._ref],n=void 0===n?!0:n;this._fields.forEach(function(n){var r=this.pipeline.run(t.tokenizer(e[n.name]));i[n.name]=r,t.SortedSet.prototype.add.apply(o,r)},this),this.documentStore.set(r,o),t.SortedSet.prototype.add.apply(this.corpusTokens,o.toArray());for(var s=0;s<o.length;s++){var a=o.elements[s],h=this._fields.reduce(function(t,e){var n=i[e.name].length;if(!n)return t;var o=i[e.name].filter(function(t){return t===a}).length;return t+o/n*e.boost},0);this.tokenStore.add(a,{ref:r,tf:h})}n&&this.eventEmitter.emit("add",e,this)},t.Index.prototype.remove=function(t,e){var n=t[this._ref],e=void 0===e?!0:e;if(this.documentStore.has(n)){var i=this.documentStore.get(n);this.documentStore.remove(n),i.forEach(function(t){this.tokenStore.remove(t,n)},this),e&&this.eventEmitter.emit("remove",t,this)}},t.Index.prototype.update=function(t,e){var e=void 0===e?!0:e;this.remove(t,!1),this.add(t,!1),e&&this.eventEmitter.emit("update",t,this)},t.Index.prototype.idf=function(t){var e="@"+t;if(Object.prototype.hasOwnProperty.call(this._idfCache,e))return this._idfCache[e];var n=this.tokenStore.count(t),i=1;return n>0&&(i=1+Math.log(this.tokenStore.length/n)),this._idfCache[e]=i},t.Index.prototype.search=function(e){var n=this.pipeline.run(t.tokenizer(e)),i=new t.Vector,o=[],r=this._fields.reduce(function(t,e){return t+e.boost},0),s=n.some(function(t){return this.tokenStore.has(t)},this);if(!s)return[];n.forEach(function(e,n,s){var a=1/s.length*this._fields.length*r,h=this,u=this.tokenStore.expand(e).reduce(function(n,o){var r=h.corpusTokens.indexOf(o),s=h.idf(o),u=1,l=new t.SortedSet;if(o!==e){var c=Math.max(3,o.length-e.length);u=1/Math.log(c)}return r>-1&&i.insert(r,a*s*u),Object.keys(h.tokenStore.get(o)).forEach(function(t){l.add(t)}),n.union(l)},new t.SortedSet);o.push(u)},this);var a=o.reduce(function(t,e){return t.intersect(e)});return a.map(function(t){return{ref:t,score:i.similarity(this.documentVector(t))}},this).sort(function(t,e){return e.score-t.score})},t.Index.prototype.documentVector=function(e){for(var n=this.documentStore.get(e),i=n.length,o=new t.Vector,r=0;i>r;r++){var s=n.elements[r],a=this.tokenStore.get(s)[e].tf,h=this.idf(s);o.insert(this.corpusTokens.indexOf(s),a*h)}return o},t.Index.prototype.toJSON=function(){return{version:t.version,fields:this._fields,ref:this._ref,documentStore:this.documentStore.toJSON(),tokenStore:this.tokenStore.toJSON(),corpusTokens:this.corpusTokens.toJSON(),pipeline:this.pipeline.toJSON()}},t.Index.prototype.use=function(t){var e=Array.prototype.slice.call(arguments,1);e.unshift(this),t.apply(this,e)},t.Store=function(){this.store={},this.length=0},t.Store.load=function(e){var n=new this;return n.length=e.length,n.store=Object.keys(e.store).reduce(function(n,i){return n[i]=t.SortedSet.load(e.store[i]),n},{}),n},t.Store.prototype.set=function(t,e){this.has(t)||this.length++,this.store[t]=e},t.Store.prototype.get=function(t){return this.store[t]},t.Store.prototype.has=function(t){return t in this.store},t.Store.prototype.remove=function(t){this.has(t)&&(delete this.store[t],this.length--)},t.Store.prototype.toJSON=function(){return{store:this.store,length:this.length}},t.stemmer=function(){var t={ational:"ate",tional:"tion",enci:"ence",anci:"ance",izer:"ize",bli:"ble",alli:"al",entli:"ent",eli:"e",ousli:"ous",ization:"ize",ation:"ate",ator:"ate",alism:"al",iveness:"ive",fulness:"ful",ousness:"ous",aliti:"al",iviti:"ive",biliti:"ble",logi:"log"},e={icate:"ic",ative:"",alize:"al",iciti:"ic",ical:"ic",ful:"",ness:""},n="[^aeiou]",i="[aeiouy]",o=n+"[^aeiouy]*",r=i+"[aeiou]*",s="^("+o+")?"+r+o,a="^("+o+")?"+r+o+"("+r+")?$",h="^("+o+")?"+r+o+r+o,u="^("+o+")?"+i,l=new RegExp(s),c=new RegExp(h),p=new RegExp(a),f=new RegExp(u),d=/^(.+?)(ss|i)es$/,v=/^(.+?)([^s])s$/,m=/^(.+?)eed$/,g=/^(.+?)(ed|ing)$/,y=/.$/,S=/(at|bl|iz)$/,w=new RegExp("([^aeiouylsz])\\1$"),x=new RegExp("^"+o+i+"[^aeiouwxy]$"),k=/^(.+?[^aeiou])y$/,b=/^(.+?)(ational|tional|enci|anci|izer|bli|alli|entli|eli|ousli|ization|ation|ator|alism|iveness|fulness|ousness|aliti|iviti|biliti|logi)$/,E=/^(.+?)(icate|ative|alize|iciti|ical|ful|ness)$/,_=/^(.+?)(al|ance|ence|er|ic|able|ible|ant|ement|ment|ent|ou|ism|ate|iti|ous|ive|ize)$/,O=/^(.+?)(s|t)(ion)$/,F=/^(.+?)e$/,P=/ll$/,T=new RegExp("^"+o+i+"[^aeiouwxy]$"),$=function(n){var i,o,r,s,a,h,u;if(n.length<3)return n;if(r=n.substr(0,1),"y"==r&&(n=r.toUpperCase()+n.substr(1)),s=d,a=v,s.test(n)?n=n.replace(s,"$1$2"):a.test(n)&&(n=n.replace(a,"$1$2")),s=m,a=g,s.test(n)){var $=s.exec(n);s=l,s.test($[1])&&(s=y,n=n.replace(s,""))}else if(a.test(n)){var $=a.exec(n);i=$[1],a=f,a.test(i)&&(n=i,a=S,h=w,u=x,a.test(n)?n+="e":h.test(n)?(s=y,n=n.replace(s,"")):u.test(n)&&(n+="e"))}if(s=k,s.test(n)){var $=s.exec(n);i=$[1],n=i+"i"}if(s=b,s.test(n)){var $=s.exec(n);i=$[1],o=$[2],s=l,s.test(i)&&(n=i+t[o])}if(s=E,s.test(n)){var $=s.exec(n);i=$[1],o=$[2],s=l,s.test(i)&&(n=i+e[o])}if(s=_,a=O,s.test(n)){var $=s.exec(n);i=$[1],s=c,s.test(i)&&(n=i)}else if(a.test(n)){var $=a.exec(n);i=$[1]+$[2],a=c,a.test(i)&&(n=i)}if(s=F,s.test(n)){var $=s.exec(n);i=$[1],s=c,a=p,h=T,(s.test(i)||a.test(i)&&!h.test(i))&&(n=i)}return s=P,a=c,s.test(n)&&a.test(n)&&(s=y,n=n.replace(s,"")),"y"==r&&(n=r.toLowerCase()+n.substr(1)),n};return $}(),t.Pipeline.registerFunction(t.stemmer,"stemmer"),t.stopWordFilter=function(e){return-1===t.stopWordFilter.stopWords.indexOf(e)?e:void 0},t.stopWordFilter.stopWords=new t.SortedSet,t.stopWordFilter.stopWords.length=119,t.stopWordFilter.stopWords.elements=["","a","able","about","across","after","all","almost","also","am","among","an","and","any","are","as","at","be","because","been","but","by","can","cannot","could","dear","did","do","does","either","else","ever","every","for","from","get","got","had","has","have","he","her","hers","him","his","how","however","i","if","in","into","is","it","its","just","least","let","like","likely","may","me","might","most","must","my","neither","no","nor","not","of","off","often","on","only","or","other","our","own","rather","said","say","says","she","should","since","so","some","than","that","the","their","them","then","there","these","they","this","tis","to","too","twas","us","wants","was","we","were","what","when","where","which","while","who","whom","why","will","with","would","yet","you","your"],t.Pipeline.registerFunction(t.stopWordFilter,"stopWordFilter"),t.trimmer=function(t){return t.replace(/^\W+/,"").replace(/\W+$/,"")},t.Pipeline.registerFunction(t.trimmer,"trimmer"),t.TokenStore=function(){this.root={docs:{}},this.length=0},t.TokenStore.load=function(t){var e=new this;return e.root=t.root,e.length=t.length,e},t.TokenStore.prototype.add=function(t,e,n){var n=n||this.root,i=t[0],o=t.slice(1);return i in n||(n[i]={docs:{}}),0===o.length?(n[i].docs[e.ref]=e,void(this.length+=1)):this.add(o,e,n[i])},t.TokenStore.prototype.has=function(t){if(!t)return!1;for(var e=this.root,n=0;n<t.length;n++){if(!e[t[n]])return!1;e=e[t[n]]}return!0},t.TokenStore.prototype.getNode=function(t){if(!t)return{};for(var e=this.root,n=0;n<t.length;n++){if(!e[t[n]])return{};e=e[t[n]]}return e},t.TokenStore.prototype.get=function(t,e){return this.getNode(t,e).docs||{}},t.TokenStore.prototype.count=function(t,e){return Object.keys(this.get(t,e)).length},t.TokenStore.prototype.remove=function(t,e){if(t){for(var n=this.root,i=0;i<t.length;i++){if(!(t[i]in n))return;n=n[t[i]]}delete n.docs[e]}},t.TokenStore.prototype.expand=function(t,e){var n=this.getNode(t),i=n.docs||{},e=e||[];return Object.keys(i).length&&e.push(t),Object.keys(n).forEach(function(n){"docs"!==n&&e.concat(this.expand(t+n,e))},this),e},t.TokenStore.prototype.toJSON=function(){return{root:this.root,length:this.length}},function(t,e){"function"==typeof define&&define.amd?define(e):"object"==typeof exports?module.exports=e():t.lunr=e()}(this,function(){return t})}();
diff --git a/docs/apex-3.3/mkdocs/js/lunr.min.js b/docs/apex-3.3/mkdocs/js/lunr.min.js
new file mode 100644
index 0000000..b0198df
--- /dev/null
+++ b/docs/apex-3.3/mkdocs/js/lunr.min.js
@@ -0,0 +1,7 @@
+/**
+ * lunr - http://lunrjs.com - A bit like Solr, but much smaller and not as bright - 0.7.0
+ * Copyright (C) 2016 Oliver Nightingale
+ * MIT Licensed
+ * @license
+ */
+!function(){var t=function(e){var n=new t.Index;return n.pipeline.add(t.trimmer,t.stopWordFilter,t.stemmer),e&&e.call(n,n),n};t.version="0.7.0",t.utils={},t.utils.warn=function(t){return function(e){t.console&&console.warn&&console.warn(e)}}(this),t.utils.asString=function(t){return void 0===t||null===t?"":t.toString()},t.EventEmitter=function(){this.events={}},t.EventEmitter.prototype.addListener=function(){var t=Array.prototype.slice.call(arguments),e=t.pop(),n=t;if("function"!=typeof e)throw new TypeError("last argument must be a function");n.forEach(function(t){this.hasHandler(t)||(this.events[t]=[]),this.events[t].push(e)},this)},t.EventEmitter.prototype.removeListener=function(t,e){if(this.hasHandler(t)){var n=this.events[t].indexOf(e);this.events[t].splice(n,1),this.events[t].length||delete this.events[t]}},t.EventEmitter.prototype.emit=function(t){if(this.hasHandler(t)){var e=Array.prototype.slice.call(arguments,1);this.events[t].forEach(function(t){t.apply(void 0,e)})}},t.EventEmitter.prototype.hasHandler=function(t){return t in this.events},t.tokenizer=function(e){return arguments.length&&null!=e&&void 0!=e?Array.isArray(e)?e.map(function(e){return t.utils.asString(e).toLowerCase()}):e.toString().trim().toLowerCase().split(t.tokenizer.seperator):[]},t.tokenizer.seperator=/[\s\-]+/,t.tokenizer.load=function(t){var e=this.registeredFunctions[t];if(!e)throw new Error("Cannot load un-registered function: "+t);return e},t.tokenizer.label="default",t.tokenizer.registeredFunctions={"default":t.tokenizer},t.tokenizer.registerFunction=function(e,n){n in this.registeredFunctions&&t.utils.warn("Overwriting existing tokenizer: "+n),e.label=n,this.registeredFunctions[n]=e},t.Pipeline=function(){this._stack=[]},t.Pipeline.registeredFunctions={},t.Pipeline.registerFunction=function(e,n){n in this.registeredFunctions&&t.utils.warn("Overwriting existing registered function: "+n),e.label=n,t.Pipeline.registeredFunctions[e.label]=e},t.Pipeline.warnIfFunctionNotRegistered=function(e){var n=e.label&&e.label in this.registeredFunctions;n||t.utils.warn("Function is not registered with pipeline. This may cause problems when serialising the index.\n",e)},t.Pipeline.load=function(e){var n=new t.Pipeline;return e.forEach(function(e){var i=t.Pipeline.registeredFunctions[e];if(!i)throw new Error("Cannot load un-registered function: "+e);n.add(i)}),n},t.Pipeline.prototype.add=function(){var e=Array.prototype.slice.call(arguments);e.forEach(function(e){t.Pipeline.warnIfFunctionNotRegistered(e),this._stack.push(e)},this)},t.Pipeline.prototype.after=function(e,n){t.Pipeline.warnIfFunctionNotRegistered(n);var i=this._stack.indexOf(e);if(-1==i)throw new Error("Cannot find existingFn");i+=1,this._stack.splice(i,0,n)},t.Pipeline.prototype.before=function(e,n){t.Pipeline.warnIfFunctionNotRegistered(n);var i=this._stack.indexOf(e);if(-1==i)throw new Error("Cannot find existingFn");this._stack.splice(i,0,n)},t.Pipeline.prototype.remove=function(t){var e=this._stack.indexOf(t);-1!=e&&this._stack.splice(e,1)},t.Pipeline.prototype.run=function(t){for(var e=[],n=t.length,i=this._stack.length,r=0;n>r;r++){for(var o=t[r],s=0;i>s&&(o=this._stack[s](o,r,t),void 0!==o&&""!==o);s++);void 0!==o&&""!==o&&e.push(o)}return e},t.Pipeline.prototype.reset=function(){this._stack=[]},t.Pipeline.prototype.toJSON=function(){return this._stack.map(function(e){return t.Pipeline.warnIfFunctionNotRegistered(e),e.label})},t.Vector=function(){this._magnitude=null,this.list=void 0,this.length=0},t.Vector.Node=function(t,e,n){this.idx=t,this.val=e,this.next=n},t.Vector.prototype.insert=function(e,n){this._magnitude=void 0;var i=this.list;if(!i)return this.list=new t.Vector.Node(e,n,i),this.length++;if(e<i.idx)return this.list=new t.Vector.Node(e,n,i),this.length++;for(var r=i,o=i.next;void 0!=o;){if(e<o.idx)return r.next=new t.Vector.Node(e,n,o),this.length++;r=o,o=o.next}return r.next=new t.Vector.Node(e,n,o),this.length++},t.Vector.prototype.magnitude=function(){if(this._magnitude)return this._magnitude;for(var t,e=this.list,n=0;e;)t=e.val,n+=t*t,e=e.next;return this._magnitude=Math.sqrt(n)},t.Vector.prototype.dot=function(t){for(var e=this.list,n=t.list,i=0;e&&n;)e.idx<n.idx?e=e.next:e.idx>n.idx?n=n.next:(i+=e.val*n.val,e=e.next,n=n.next);return i},t.Vector.prototype.similarity=function(t){return this.dot(t)/(this.magnitude()*t.magnitude())},t.SortedSet=function(){this.length=0,this.elements=[]},t.SortedSet.load=function(t){var e=new this;return e.elements=t,e.length=t.length,e},t.SortedSet.prototype.add=function(){var t,e;for(t=0;t<arguments.length;t++)e=arguments[t],~this.indexOf(e)||this.elements.splice(this.locationFor(e),0,e);this.length=this.elements.length},t.SortedSet.prototype.toArray=function(){return this.elements.slice()},t.SortedSet.prototype.map=function(t,e){return this.elements.map(t,e)},t.SortedSet.prototype.forEach=function(t,e){return this.elements.forEach(t,e)},t.SortedSet.prototype.indexOf=function(t){for(var e=0,n=this.elements.length,i=n-e,r=e+Math.floor(i/2),o=this.elements[r];i>1;){if(o===t)return r;t>o&&(e=r),o>t&&(n=r),i=n-e,r=e+Math.floor(i/2),o=this.elements[r]}return o===t?r:-1},t.SortedSet.prototype.locationFor=function(t){for(var e=0,n=this.elements.length,i=n-e,r=e+Math.floor(i/2),o=this.elements[r];i>1;)t>o&&(e=r),o>t&&(n=r),i=n-e,r=e+Math.floor(i/2),o=this.elements[r];return o>t?r:t>o?r+1:void 0},t.SortedSet.prototype.intersect=function(e){for(var n=new t.SortedSet,i=0,r=0,o=this.length,s=e.length,a=this.elements,h=e.elements;;){if(i>o-1||r>s-1)break;a[i]!==h[r]?a[i]<h[r]?i++:a[i]>h[r]&&r++:(n.add(a[i]),i++,r++)}return n},t.SortedSet.prototype.clone=function(){var e=new t.SortedSet;return e.elements=this.toArray(),e.length=e.elements.length,e},t.SortedSet.prototype.union=function(t){var e,n,i;this.length>=t.length?(e=this,n=t):(e=t,n=this),i=e.clone();for(var r=0,o=n.toArray();r<o.length;r++)i.add(o[r]);return i},t.SortedSet.prototype.toJSON=function(){return this.toArray()},t.Index=function(){this._fields=[],this._ref="id",this.pipeline=new t.Pipeline,this.documentStore=new t.Store,this.tokenStore=new t.TokenStore,this.corpusTokens=new t.SortedSet,this.eventEmitter=new t.EventEmitter,this.tokenizerFn=t.tokenizer,this._idfCache={},this.on("add","remove","update",function(){this._idfCache={}}.bind(this))},t.Index.prototype.on=function(){var t=Array.prototype.slice.call(arguments);return this.eventEmitter.addListener.apply(this.eventEmitter,t)},t.Index.prototype.off=function(t,e){return this.eventEmitter.removeListener(t,e)},t.Index.load=function(e){e.version!==t.version&&t.utils.warn("version mismatch: current "+t.version+" importing "+e.version);var n=new this;return n._fields=e.fields,n._ref=e.ref,n.tokenizer=t.tokenizer.load(e.tokenizer),n.documentStore=t.Store.load(e.documentStore),n.tokenStore=t.TokenStore.load(e.tokenStore),n.corpusTokens=t.SortedSet.load(e.corpusTokens),n.pipeline=t.Pipeline.load(e.pipeline),n},t.Index.prototype.field=function(t,e){var e=e||{},n={name:t,boost:e.boost||1};return this._fields.push(n),this},t.Index.prototype.ref=function(t){return this._ref=t,this},t.Index.prototype.tokenizer=function(e){var n=e.label&&e.label in t.tokenizer.registeredFunctions;return n||t.utils.warn("Function is not a registered tokenizer. This may cause problems when serialising the index"),this.tokenizerFn=e,this},t.Index.prototype.add=function(e,n){var i={},r=new t.SortedSet,o=e[this._ref],n=void 0===n?!0:n;this._fields.forEach(function(t){var n=this.pipeline.run(this.tokenizerFn(e[t.name]));i[t.name]=n;for(var o=0;o<n.length;o++){var s=n[o];r.add(s),this.corpusTokens.add(s)}},this),this.documentStore.set(o,r);for(var s=0;s<r.length;s++){for(var a=r.elements[s],h=0,u=0;u<this._fields.length;u++){var l=this._fields[u],c=i[l.name],f=c.length;if(f){for(var d=0,p=0;f>p;p++)c[p]===a&&d++;h+=d/f*l.boost}}this.tokenStore.add(a,{ref:o,tf:h})}n&&this.eventEmitter.emit("add",e,this)},t.Index.prototype.remove=function(t,e){var n=t[this._ref],e=void 0===e?!0:e;if(this.documentStore.has(n)){var i=this.documentStore.get(n);this.documentStore.remove(n),i.forEach(function(t){this.tokenStore.remove(t,n)},this),e&&this.eventEmitter.emit("remove",t,this)}},t.Index.prototype.update=function(t,e){var e=void 0===e?!0:e;this.remove(t,!1),this.add(t,!1),e&&this.eventEmitter.emit("update",t,this)},t.Index.prototype.idf=function(t){var e="@"+t;if(Object.prototype.hasOwnProperty.call(this._idfCache,e))return this._idfCache[e];var n=this.tokenStore.count(t),i=1;return n>0&&(i=1+Math.log(this.documentStore.length/n)),this._idfCache[e]=i},t.Index.prototype.search=function(e){var n=this.pipeline.run(this.tokenizerFn(e)),i=new t.Vector,r=[],o=this._fields.reduce(function(t,e){return t+e.boost},0),s=n.some(function(t){return this.tokenStore.has(t)},this);if(!s)return[];n.forEach(function(e,n,s){var a=1/s.length*this._fields.length*o,h=this,u=this.tokenStore.expand(e).reduce(function(n,r){var o=h.corpusTokens.indexOf(r),s=h.idf(r),u=1,l=new t.SortedSet;if(r!==e){var c=Math.max(3,r.length-e.length);u=1/Math.log(c)}o>-1&&i.insert(o,a*s*u);for(var f=h.tokenStore.get(r),d=Object.keys(f),p=d.length,v=0;p>v;v++)l.add(f[d[v]].ref);return n.union(l)},new t.SortedSet);r.push(u)},this);var a=r.reduce(function(t,e){return t.intersect(e)});return a.map(function(t){return{ref:t,score:i.similarity(this.documentVector(t))}},this).sort(function(t,e){return e.score-t.score})},t.Index.prototype.documentVector=function(e){for(var n=this.documentStore.get(e),i=n.length,r=new t.Vector,o=0;i>o;o++){var s=n.elements[o],a=this.tokenStore.get(s)[e].tf,h=this.idf(s);r.insert(this.corpusTokens.indexOf(s),a*h)}return r},t.Index.prototype.toJSON=function(){return{version:t.version,fields:this._fields,ref:this._ref,tokenizer:this.tokenizerFn.label,documentStore:this.documentStore.toJSON(),tokenStore:this.tokenStore.toJSON(),corpusTokens:this.corpusTokens.toJSON(),pipeline:this.pipeline.toJSON()}},t.Index.prototype.use=function(t){var e=Array.prototype.slice.call(arguments,1);e.unshift(this),t.apply(this,e)},t.Store=function(){this.store={},this.length=0},t.Store.load=function(e){var n=new this;return n.length=e.length,n.store=Object.keys(e.store).reduce(function(n,i){return n[i]=t.SortedSet.load(e.store[i]),n},{}),n},t.Store.prototype.set=function(t,e){this.has(t)||this.length++,this.store[t]=e},t.Store.prototype.get=function(t){return this.store[t]},t.Store.prototype.has=function(t){return t in this.store},t.Store.prototype.remove=function(t){this.has(t)&&(delete this.store[t],this.length--)},t.Store.prototype.toJSON=function(){return{store:this.store,length:this.length}},t.stemmer=function(){var t={ational:"ate",tional:"tion",enci:"ence",anci:"ance",izer:"ize",bli:"ble",alli:"al",entli:"ent",eli:"e",ousli:"ous",ization:"ize",ation:"ate",ator:"ate",alism:"al",iveness:"ive",fulness:"ful",ousness:"ous",aliti:"al",iviti:"ive",biliti:"ble",logi:"log"},e={icate:"ic",ative:"",alize:"al",iciti:"ic",ical:"ic",ful:"",ness:""},n="[^aeiou]",i="[aeiouy]",r=n+"[^aeiouy]*",o=i+"[aeiou]*",s="^("+r+")?"+o+r,a="^("+r+")?"+o+r+"("+o+")?$",h="^("+r+")?"+o+r+o+r,u="^("+r+")?"+i,l=new RegExp(s),c=new RegExp(h),f=new RegExp(a),d=new RegExp(u),p=/^(.+?)(ss|i)es$/,v=/^(.+?)([^s])s$/,g=/^(.+?)eed$/,m=/^(.+?)(ed|ing)$/,y=/.$/,S=/(at|bl|iz)$/,w=new RegExp("([^aeiouylsz])\\1$"),k=new RegExp("^"+r+i+"[^aeiouwxy]$"),x=/^(.+?[^aeiou])y$/,b=/^(.+?)(ational|tional|enci|anci|izer|bli|alli|entli|eli|ousli|ization|ation|ator|alism|iveness|fulness|ousness|aliti|iviti|biliti|logi)$/,E=/^(.+?)(icate|ative|alize|iciti|ical|ful|ness)$/,F=/^(.+?)(al|ance|ence|er|ic|able|ible|ant|ement|ment|ent|ou|ism|ate|iti|ous|ive|ize)$/,_=/^(.+?)(s|t)(ion)$/,z=/^(.+?)e$/,O=/ll$/,P=new RegExp("^"+r+i+"[^aeiouwxy]$"),T=function(n){var i,r,o,s,a,h,u;if(n.length<3)return n;if(o=n.substr(0,1),"y"==o&&(n=o.toUpperCase()+n.substr(1)),s=p,a=v,s.test(n)?n=n.replace(s,"$1$2"):a.test(n)&&(n=n.replace(a,"$1$2")),s=g,a=m,s.test(n)){var T=s.exec(n);s=l,s.test(T[1])&&(s=y,n=n.replace(s,""))}else if(a.test(n)){var T=a.exec(n);i=T[1],a=d,a.test(i)&&(n=i,a=S,h=w,u=k,a.test(n)?n+="e":h.test(n)?(s=y,n=n.replace(s,"")):u.test(n)&&(n+="e"))}if(s=x,s.test(n)){var T=s.exec(n);i=T[1],n=i+"i"}if(s=b,s.test(n)){var T=s.exec(n);i=T[1],r=T[2],s=l,s.test(i)&&(n=i+t[r])}if(s=E,s.test(n)){var T=s.exec(n);i=T[1],r=T[2],s=l,s.test(i)&&(n=i+e[r])}if(s=F,a=_,s.test(n)){var T=s.exec(n);i=T[1],s=c,s.test(i)&&(n=i)}else if(a.test(n)){var T=a.exec(n);i=T[1]+T[2],a=c,a.test(i)&&(n=i)}if(s=z,s.test(n)){var T=s.exec(n);i=T[1],s=c,a=f,h=P,(s.test(i)||a.test(i)&&!h.test(i))&&(n=i)}return s=O,a=c,s.test(n)&&a.test(n)&&(s=y,n=n.replace(s,"")),"y"==o&&(n=o.toLowerCase()+n.substr(1)),n};return T}(),t.Pipeline.registerFunction(t.stemmer,"stemmer"),t.generateStopWordFilter=function(t){var e=t.reduce(function(t,e){return t[e]=e,t},{});return function(t){return t&&e[t]!==t?t:void 0}},t.stopWordFilter=t.generateStopWordFilter(["a","able","about","across","after","all","almost","also","am","among","an","and","any","are","as","at","be","because","been","but","by","can","cannot","could","dear","did","do","does","either","else","ever","every","for","from","get","got","had","has","have","he","her","hers","him","his","how","however","i","if","in","into","is","it","its","just","least","let","like","likely","may","me","might","most","must","my","neither","no","nor","not","of","off","often","on","only","or","other","our","own","rather","said","say","says","she","should","since","so","some","than","that","the","their","them","then","there","these","they","this","tis","to","too","twas","us","wants","was","we","were","what","when","where","which","while","who","whom","why","will","with","would","yet","you","your"]),t.Pipeline.registerFunction(t.stopWordFilter,"stopWordFilter"),t.trimmer=function(t){return t.replace(/^\W+/,"").replace(/\W+$/,"")},t.Pipeline.registerFunction(t.trimmer,"trimmer"),t.TokenStore=function(){this.root={docs:{}},this.length=0},t.TokenStore.load=function(t){var e=new this;return e.root=t.root,e.length=t.length,e},t.TokenStore.prototype.add=function(t,e,n){var n=n||this.root,i=t.charAt(0),r=t.slice(1);return i in n||(n[i]={docs:{}}),0===r.length?(n[i].docs[e.ref]=e,void(this.length+=1)):this.add(r,e,n[i])},t.TokenStore.prototype.has=function(t){if(!t)return!1;for(var e=this.root,n=0;n<t.length;n++){if(!e[t.charAt(n)])return!1;e=e[t.charAt(n)]}return!0},t.TokenStore.prototype.getNode=function(t){if(!t)return{};for(var e=this.root,n=0;n<t.length;n++){if(!e[t.charAt(n)])return{};e=e[t.charAt(n)]}return e},t.TokenStore.prototype.get=function(t,e){return this.getNode(t,e).docs||{}},t.TokenStore.prototype.count=function(t,e){return Object.keys(this.get(t,e)).length},t.TokenStore.prototype.remove=function(t,e){if(t){for(var n=this.root,i=0;i<t.length;i++){if(!(t.charAt(i)in n))return;n=n[t.charAt(i)]}delete n.docs[e]}},t.TokenStore.prototype.expand=function(t,e){var n=this.getNode(t),i=n.docs||{},e=e||[];return Object.keys(i).length&&e.push(t),Object.keys(n).forEach(function(n){"docs"!==n&&e.concat(this.expand(t+n,e))},this),e},t.TokenStore.prototype.toJSON=function(){return{root:this.root,length:this.length}},function(t,e){"function"==typeof define&&define.amd?define(e):"object"==typeof exports?module.exports=e():t.lunr=e()}(this,function(){return t})}();
diff --git a/docs/apex-3.3/mkdocs/js/search.js b/docs/apex-3.3/mkdocs/js/search.js
index 0bdb5b0..88d563a 100644
--- a/docs/apex-3.3/mkdocs/js/search.js
+++ b/docs/apex-3.3/mkdocs/js/search.js
@@ -1,6 +1,6 @@
require([
base_url + '/mkdocs/js/mustache.min.js',
- base_url + '/mkdocs/js/lunr-0.5.7.min.js',
+ base_url + '/mkdocs/js/lunr.min.js',
'text!search-results-template.mustache',
'text!../search_index.json',
], function (Mustache, lunr, results_template, data) {
diff --git a/docs/apex-3.3/mkdocs/search_index.json b/docs/apex-3.3/mkdocs/search_index.json
index 85c9b89..34dd42f 100644
--- a/docs/apex-3.3/mkdocs/search_index.json
+++ b/docs/apex-3.3/mkdocs/search_index.json
@@ -72,10 +72,20 @@
},
{
"location": "/application_development/#hadoop-cluster",
- "text": "In this section we discuss various Hadoop cluster setups. Single Node Cluster In a single node Hadoop cluster all services are deployed on a\nsingle server (a developer can use his/her development machine as a\nsingle node cluster). The platform does not distinguish between a single\nor multi-node setup and behaves exactly the same in both cases. In this mode, the resource manager, name node, data node, and node\nmanager occupy one process each. This is an example of running a\nstreaming application as a multi-process\u00a0application on the same server.\nWith prevalence of fast, multi-core systems, this mode is effective for\ndebugging, fine tuning, and generic analysis before submitting the job\nto a larger Hadoop cluster. In this mode, execution uses the Hadoop\nservices and hence is likely to identify issues that are related to the\nHadoop environment (such issues will not be uncovered in local mode).\nThe throughput will obviously not be as high as on a multi-node Hadoop\ncluster. Additionally, since each container (i.e. Java process) requires\na significant amount of memory, you will be able to run a much smaller\nnumber of containers than on a multi-node cluster. Multi-Node Cluster In a multi-node Hadoop cluster all the services of Hadoop are\ntypically distributed across multiple nodes in a production or\nproduction-level test environment. Upon launch the application is\nsubmitted to the Hadoop cluster and executes as a multi-processapplication on\u00a0multiple nodes. Before you start deploying, testing and troubleshooting your\napplication on a cluster, you should ensure that Hadoop (version 2.2.0\nor later)\u00a0is properly installed and\nyou have basic skills for working with it.",
+ "text": "In this section we discuss various Hadoop cluster setups.",
"title": "Hadoop Cluster"
},
{
+ "location": "/application_development/#single-node-cluster",
+ "text": "In a single node Hadoop cluster all services are deployed on a\nsingle server (a developer can use his/her development machine as a\nsingle node cluster). The platform does not distinguish between a single\nor multi-node setup and behaves exactly the same in both cases. In this mode, the resource manager, name node, data node, and node\nmanager occupy one process each. This is an example of running a\nstreaming application as a multi-process\u00a0application on the same server.\nWith prevalence of fast, multi-core systems, this mode is effective for\ndebugging, fine tuning, and generic analysis before submitting the job\nto a larger Hadoop cluster. In this mode, execution uses the Hadoop\nservices and hence is likely to identify issues that are related to the\nHadoop environment (such issues will not be uncovered in local mode).\nThe throughput will obviously not be as high as on a multi-node Hadoop\ncluster. Additionally, since each container (i.e. Java process) requires\na significant amount of memory, you will be able to run a much smaller\nnumber of containers than on a multi-node cluster.",
+ "title": "Single Node Cluster"
+ },
+ {
+ "location": "/application_development/#multi-node-cluster",
+ "text": "In a multi-node Hadoop cluster all the services of Hadoop are\ntypically distributed across multiple nodes in a production or\nproduction-level test environment. Upon launch the application is\nsubmitted to the Hadoop cluster and executes as a multi-processapplication on\u00a0multiple nodes. Before you start deploying, testing and troubleshooting your\napplication on a cluster, you should ensure that Hadoop (version 2.2.0\nor later)\u00a0is properly installed and\nyou have basic skills for working with it.",
+ "title": "Multi-Node Cluster"
+ },
+ {
"location": "/application_development/#apache-apex-platform-overview",
"text": "",
"title": "Apache Apex Platform Overview"
@@ -92,40 +102,190 @@
},
{
"location": "/application_development/#hadoop-components",
- "text": "In this section we cover some aspects of Hadoop that your\nstreaming application interacts with. This section is not meant to\neducate the reader on Hadoop, but just get the reader acquainted with\nthe terms. We strongly advise readers to learn Hadoop from other\nsources. A streaming application runs as a native Hadoop 2.2 application.\nHadoop 2.2 does not differentiate between a map-reduce job and other\napplications, and hence as far as Hadoop is concerned, the streaming\napplication is just another job. This means that your application\nleverages all the bells and whistles Hadoop provides and is fully\nsupported within Hadoop technology stack. The platform is responsible\nfor properly integrating itself with the relevant components of Hadoop\nthat exist today and those that may emerge in the future All investments that leverage multi-tenancy (for example quotas\nand queues), security (for example kerberos), data flow integration (for\nexample copying data in-out of HDFS), monitoring, metrics collections,\netc. will require no changes when streaming applications run on\nHadoop. YARN YARN is\nthe core library of Hadoop 2.2 that is tasked with resource management\nand works as a distributed application framework. In this section we\nwill walk through Yarn's components. In Hadoop 2.2, the old jobTracker\nhas been replaced by a combination of ResourceManager (RM) and\nApplicationMaster (AM). Resource Manager (RM) ResourceManager (RM)\nmanages all the distributed resources. It allocates and arbitrates all\nthe slots and the resources (cpu, memory, network) of these slots. It\nworks with per-node NodeManagers (NMs) and per-application\nApplicationMasters (AMs). Currently memory usage is monitored by RM; in\nupcoming releases it will have CPU as well as network management. RM is\nshared by map-reduce and streaming applications. Running streaming\napplications requires no changes in the RM. Application Master (AM) The AM is the watchdog or monitoring process for your application\nand has the responsibility of negotiating resources with RM and\ninteracting with NodeManagers to get the allocated containers started.\nThe AM is the starting point of your application and is considered user\ncode (not system Hadoop code). The AM itself runs in one container. All\nresource management within the application are managed by the AM. This\nis a critical feature for Hadoop 2.2 where tasks done by jobTracker in\nHadoop 1.0 have been distributed allowing Hadoop 2.2 to scale much\nbeyond Hadoop 1.0. STRAM is a native YARN ApplicationManager. Node Managers (NM) There is one NodeManager (NM)\nper node in the cluster. All the containers (i.e. processes) on that\nnode are monitored by the NM. It takes instructions from RM and manages\nresources of that node as per RM instructions. NMs interactions are same\nfor map-reduce and for streaming applications. Running streaming\napplications requires no changes in the NM. RPC Protocol Communication among RM, AM, and NM is done via the Hadoop RPC\nprotocol. Streaming applications use the same protocol to send their\ndata. No changes are needed in RPC support provided by Hadoop to enable\ncommunication done by components of your application. HDFS Hadoop includes a highly fault tolerant, high throughput\ndistributed file system ( HDFS ).\nIt runs on commodity hardware, and your streaming application will, by\ndefault, use it. There is no difference between files created by a\nstreaming application and those created by map-reduce.",
+ "text": "In this section we cover some aspects of Hadoop that your\nstreaming application interacts with. This section is not meant to\neducate the reader on Hadoop, but just get the reader acquainted with\nthe terms. We strongly advise readers to learn Hadoop from other\nsources. A streaming application runs as a native Hadoop 2.2 application.\nHadoop 2.2 does not differentiate between a map-reduce job and other\napplications, and hence as far as Hadoop is concerned, the streaming\napplication is just another job. This means that your application\nleverages all the bells and whistles Hadoop provides and is fully\nsupported within Hadoop technology stack. The platform is responsible\nfor properly integrating itself with the relevant components of Hadoop\nthat exist today and those that may emerge in the future All investments that leverage multi-tenancy (for example quotas\nand queues), security (for example kerberos), data flow integration (for\nexample copying data in-out of HDFS), monitoring, metrics collections,\netc. will require no changes when streaming applications run on\nHadoop.",
"title": "Hadoop Components"
},
{
+ "location": "/application_development/#yarn",
+ "text": "YARN is\nthe core library of Hadoop 2.2 that is tasked with resource management\nand works as a distributed application framework. In this section we\nwill walk through Yarn's components. In Hadoop 2.2, the old jobTracker\nhas been replaced by a combination of ResourceManager (RM) and\nApplicationMaster (AM).",
+ "title": "YARN"
+ },
+ {
+ "location": "/application_development/#resource-manager-rm",
+ "text": "ResourceManager (RM)\nmanages all the distributed resources. It allocates and arbitrates all\nthe slots and the resources (cpu, memory, network) of these slots. It\nworks with per-node NodeManagers (NMs) and per-application\nApplicationMasters (AMs). Currently memory usage is monitored by RM; in\nupcoming releases it will have CPU as well as network management. RM is\nshared by map-reduce and streaming applications. Running streaming\napplications requires no changes in the RM.",
+ "title": "Resource Manager (RM)"
+ },
+ {
+ "location": "/application_development/#application-master-am",
+ "text": "The AM is the watchdog or monitoring process for your application\nand has the responsibility of negotiating resources with RM and\ninteracting with NodeManagers to get the allocated containers started.\nThe AM is the starting point of your application and is considered user\ncode (not system Hadoop code). The AM itself runs in one container. All\nresource management within the application are managed by the AM. This\nis a critical feature for Hadoop 2.2 where tasks done by jobTracker in\nHadoop 1.0 have been distributed allowing Hadoop 2.2 to scale much\nbeyond Hadoop 1.0. STRAM is a native YARN ApplicationManager.",
+ "title": "Application Master (AM)"
+ },
+ {
+ "location": "/application_development/#node-managers-nm",
+ "text": "There is one NodeManager (NM)\nper node in the cluster. All the containers (i.e. processes) on that\nnode are monitored by the NM. It takes instructions from RM and manages\nresources of that node as per RM instructions. NMs interactions are same\nfor map-reduce and for streaming applications. Running streaming\napplications requires no changes in the NM.",
+ "title": "Node Managers (NM)"
+ },
+ {
+ "location": "/application_development/#rpc-protocol",
+ "text": "Communication among RM, AM, and NM is done via the Hadoop RPC\nprotocol. Streaming applications use the same protocol to send their\ndata. No changes are needed in RPC support provided by Hadoop to enable\ncommunication done by components of your application.",
+ "title": "RPC Protocol"
+ },
+ {
+ "location": "/application_development/#hdfs",
+ "text": "Hadoop includes a highly fault tolerant, high throughput\ndistributed file system ( HDFS ).\nIt runs on commodity hardware, and your streaming application will, by\ndefault, use it. There is no difference between files created by a\nstreaming application and those created by map-reduce.",
+ "title": "HDFS"
+ },
+ {
"location": "/application_development/#developing-an-application",
"text": "In this chapter we describe the methodology to develop an\napplication using the Realtime Streaming Platform. The platform was\ndesigned to make it easy to build and launch sophisticated streaming\napplications with the developer having to deal only with the\napplication/business logic. The platform deals with details of where to\nrun what operators on which servers and how to correctly route streams\nof data among them.",
"title": "Developing An Application"
},
{
"location": "/application_development/#development-process",
- "text": "While the platform does not mandate a specific methodology or set\nof development tools, we have recommendations to maximize productivity\nfor the different phases of application development. Design Identify common, reusable operators. Use a library\n if possible. Identify scalability and performance requirements before\n designing the DAG. Leverage attributes that the platform supports for scalability\n and performance. Use operators that are benchmarked and tested so that later\n surprises are minimized. If you have glue code, create appropriate\n unit tests for it. Use THREAD_LOCAL locality for high throughput streams. If all\n the operators on that stream cannot fit in one container,\n try\u00a0NODE_LOCAL\u00a0locality. Both THREAD_LOCAL and\n NODE_LOCAL streams avoid the Network Interface Card (NIC)\n completly. The former uses intra-process communication to also avoid\n serialization-deserialization overhead. The overall throughput and latencies are are not necessarily\n correlated to the number of operators in a simple way -- the\n relationship is more nuanced. A lot depends on how much work\n individual operators are doing, how many are able to operate in\n parallel, and how much data is flowing through the arcs of the DAG.\n It is, at times, better to break a computation down into its\n constituent simple parts and then stitch them together via streams\n to better utilize the compute resources of the cluster. Decide on a\n per application basis the fine line between complexity of each\n operator vs too many streams. Doing multiple computations in one\n operator does save network I/O, while operators that are too complex\n are hard to maintain. Do not use operators that depend on the order of two streams\n as far as possible. In such cases behavior is not idempotent. Persist key information to HDFS if possible; it may be useful\n for debugging later. Decide on an appropriate fault tolerance mechanism. If some\n data loss is acceptable, use the at-most-once mechanism as it has\n fastest recovery. Creating New Project Please refer to the Apex Application Packages \u00a0for\nthe basic steps for creating a new project. Writing the application code Preferably use an IDE (Eclipse, Netbeans etc.) that allows you to\nmanage dependencies and assists with the Java coding. Specific benefits\ninclude ease of managing operator library jar files, individual operator\nclasses, ports and properties. It will also highlight and assist to\nrectify issues such as type mismatches when adding streams while\ntyping. Testing Write test cases with JUnit or similar test framework so that code\nis tested as it is written. For such testing, the DAG can run in local\nmode within the IDE. Doing this may involve writing mock input or output\noperators for the integration points with external systems. For example,\ninstead of reading from a live data stream, the application in test mode\ncan read from and write to files. This can be done with a single\napplication DAG by instrumenting a test mode using settings in the\nconfiguration that is passed to the application factory\ninterface. Good test coverage will not only eliminate basic validation errors\nsuch as missing port connections or property constraint violations, but\nalso validate the correct processing of the data. The same tests can be\nre-run whenever the application or its dependencies change (operator\nlibraries, version of the platform etc.) Running an application The platform provides a commandline tool called dtcli\u00a0for managing applications (launching,\nkilling, viewing, etc.). This tool was already discussed above briefly\nin the section entitled Running the Test Application. It will introspect\nthe jar file specified with the launch command for applications (classes\nthat implement ApplicationFactory) or property files that define\napplications. It will also deploy the dependency jar files from the\napplication package to the cluster. Dtcli can run the application in local mode (i.e. outside a\ncluster). It is recommended to first run the application in local mode\nin the development environment before launching on the Hadoop cluster.\nThis way some of the external system integration and correct\nfunctionality of the application can be verified in an easier to debug\nenvironment before testing distributed mode. For more details on CLI please refer to the dtCli Guide .",
+ "text": "While the platform does not mandate a specific methodology or set\nof development tools, we have recommendations to maximize productivity\nfor the different phases of application development.",
"title": "Development Process"
},
{
+ "location": "/application_development/#design",
+ "text": "Identify common, reusable operators. Use a library\n if possible. Identify scalability and performance requirements before\n designing the DAG. Leverage attributes that the platform supports for scalability\n and performance. Use operators that are benchmarked and tested so that later\n surprises are minimized. If you have glue code, create appropriate\n unit tests for it. Use THREAD_LOCAL locality for high throughput streams. If all\n the operators on that stream cannot fit in one container,\n try\u00a0NODE_LOCAL\u00a0locality. Both THREAD_LOCAL and\n NODE_LOCAL streams avoid the Network Interface Card (NIC)\n completly. The former uses intra-process communication to also avoid\n serialization-deserialization overhead. The overall throughput and latencies are are not necessarily\n correlated to the number of operators in a simple way -- the\n relationship is more nuanced. A lot depends on how much work\n individual operators are doing, how many are able to operate in\n parallel, and how much data is flowing through the arcs of the DAG.\n It is, at times, better to break a computation down into its\n constituent simple parts and then stitch them together via streams\n to better utilize the compute resources of the cluster. Decide on a\n per application basis the fine line between complexity of each\n operator vs too many streams. Doing multiple computations in one\n operator does save network I/O, while operators that are too complex\n are hard to maintain. Do not use operators that depend on the order of two streams\n as far as possible. In such cases behavior is not idempotent. Persist key information to HDFS if possible; it may be useful\n for debugging later. Decide on an appropriate fault tolerance mechanism. If some\n data loss is acceptable, use the at-most-once mechanism as it has\n fastest recovery.",
+ "title": "Design"
+ },
+ {
+ "location": "/application_development/#creating-new-project",
+ "text": "Please refer to the Apex Application Packages \u00a0for\nthe basic steps for creating a new project.",
+ "title": "Creating New Project"
+ },
+ {
+ "location": "/application_development/#writing-the-application-code",
+ "text": "Preferably use an IDE (Eclipse, Netbeans etc.) that allows you to\nmanage dependencies and assists with the Java coding. Specific benefits\ninclude ease of managing operator library jar files, individual operator\nclasses, ports and properties. It will also highlight and assist to\nrectify issues such as type mismatches when adding streams while\ntyping.",
+ "title": "Writing the application code"
+ },
+ {
+ "location": "/application_development/#testing",
+ "text": "Write test cases with JUnit or similar test framework so that code\nis tested as it is written. For such testing, the DAG can run in local\nmode within the IDE. Doing this may involve writing mock input or output\noperators for the integration points with external systems. For example,\ninstead of reading from a live data stream, the application in test mode\ncan read from and write to files. This can be done with a single\napplication DAG by instrumenting a test mode using settings in the\nconfiguration that is passed to the application factory\ninterface. Good test coverage will not only eliminate basic validation errors\nsuch as missing port connections or property constraint violations, but\nalso validate the correct processing of the data. The same tests can be\nre-run whenever the application or its dependencies change (operator\nlibraries, version of the platform etc.)",
+ "title": "Testing"
+ },
+ {
+ "location": "/application_development/#running-an-application",
+ "text": "The platform provides a commandline tool called dtcli\u00a0for managing applications (launching,\nkilling, viewing, etc.). This tool was already discussed above briefly\nin the section entitled Running the Test Application. It will introspect\nthe jar file specified with the launch command for applications (classes\nthat implement ApplicationFactory) or property files that define\napplications. It will also deploy the dependency jar files from the\napplication package to the cluster. Dtcli can run the application in local mode (i.e. outside a\ncluster). It is recommended to first run the application in local mode\nin the development environment before launching on the Hadoop cluster.\nThis way some of the external system integration and correct\nfunctionality of the application can be verified in an easier to debug\nenvironment before testing distributed mode. For more details on CLI please refer to the dtCli Guide .",
+ "title": "Running an application"
+ },
+ {
"location": "/application_development/#application-api",
- "text": "This section introduces the API to write a streaming application.\nThe work involves connecting operators via streams to form the logical\nDAG. The steps are Instantiate an application (DAG) (Optional) Set Attributes Assign application name Set any other attributes as per application requirements Create/re-use and instantiate operators Assign operator name that is unique within the application Declare schema upfront for each operator (and thereby its ports) (Optional) Set properties\u00a0 and attributes on the dag as per specification Connect ports of operators via streams Each stream connects one output port of an operator to one or more input ports of other operators. (Optional) Set attributes on the streams Test the application. There are two methods to create an application, namely Java, and\nProperties file. Java API is for applications being developed by humans,\nand properties file (Hadoop like) is more suited for DAGs generated by\ntools. Java API The Java API is the most common way to create a streaming\napplication. It is meant for application developers who prefer to\nleverage the features of Java, and the ease of use and enhanced\nproductivity provided by IDEs like NetBeans or Eclipse. Using Java to\nspecify the application provides extra validation abilities of Java\ncompiler, such as compile time checks for type safety at the time of\nwriting the code. Later in this chapter you can read more about\nvalidation support in the platform. The developer specifies the streaming application by implementing\nthe ApplicationFactory interface, which is how platform tools (CLI etc.)\nrecognize and instantiate applications. Here we show how to create a\nYahoo! Finance application that streams the last trade price of a ticker\nand computes the high and low price in every 1 min window. Run above\n test application\u00a0to execute the\nDAG in local mode within the IDE. Let us revisit how the Yahoo! Finance test application constructs the DAG: public class Application implements StreamingApplication\n{\n\n ...\n\n @Override\n public void populateDAG(DAG dag, Configuration conf)\n {\n dag.getAttributes().attr(DAG.STRAM_WINDOW_SIZE_MILLIS).set(streamingWindowSizeMilliSeconds);\n\n StockTickInput tick = getStockTickInputOperator( StockTickInput , dag);\n SumKeyVal String, Long dailyVolume = getDailyVolumeOperator( DailyVolume , dag);\n ConsolidatorKeyVal String,Double,Long,String,?,? quoteOperator = getQuoteOperator( Quote , dag);\n\n RangeKeyVal String, Double highlow = getHighLowOperator( HighLow , dag, appWindowCountMinute);\n SumKeyVal String, Long minuteVolume = getMinuteVolumeOperator( MinuteVolume , dag, appWindowCountMinute);\n ConsolidatorKeyVal String,HighLow,Long,?,?,? chartOperator = getChartOperator( Chart , dag);\n\n SimpleMovingAverage String, Double priceSMA = getPriceSimpleMovingAverageOperator( PriceSMA , dag, appWindowCountSMA);\n\n dag.addStream( price , tick.price, quoteOperator.in1, highlow.data, priceSMA.data);\n dag.addStream( vol , tick.volume, dailyVolume.data, minuteVolume.data);\n dag.addStream( time , tick.time, quoteOperator.in3);\n dag.addStream( daily_vol , dailyVolume.sum, quoteOperator.in2);\n\n dag.addStream( quote_data , quoteOperator.out, getConsole( quoteConsole , dag, QUOTE ));\n\n dag.addStream( high_low , highlow.range, chartOperator.in1);\n dag.addStream( vol_1min , minuteVolume.sum, chartOperator.in2);\n dag.addStream( chart_data , chartOperator.out, getConsole( chartConsole , dag, CHART ));\n\n dag.addStream( sma_price , priceSMA.doubleSMA, getConsole( priceSMAConsole , dag, Price SMA ));\n\n return dag;\n }\n} Property File API The platform also supports specification of a DAG via a property\nfile. The aim here to make it easy for tools to create and run an\napplication. This method of specification does not have the Java\ncompiler support of compile time check, but since these applications\nwould be created by software, they should be correct by construction.\nThe syntax is derived from Hadoop properties and should be easy for\nfolks who are used to creating software that integrated with\nHadoop. Create an application (DAG): myApplication.properties # input operator that reads from a file\ndt.operator.inputOp.classname=com.acme.SampleInputOperator\ndt.operator.inputOp.fileName=somefile.txt\n\n# output operator that writes to the console\ndt.operator.outputOp.classname=com.acme.ConsoleOutputOperator\n\n# stream connecting both operators\ndt.stream.inputStream.source=inputOp.outputPort\ndt.stream.inputStream.sinks=outputOp.inputPort Above snippet is intended to convey the basic idea of specifying\nthe DAG without using Java. Operators would come from a predefined\nlibrary and referenced in the specification by class name and port names\n(obtained from the library providers documentation or runtime\nintrospection by tools). For those interested in details, see later\nsections and refer to the Operation and\nInstallation Guide\u00a0mentioned above. Attributes Attributes impact the runtime behavior of the application. They do\nnot impact the functionality. An example of an attribute is application\nname. Setting it changes the application name. Another example is\nstreaming window size. Setting it changes the streaming window size from\nthe default value to the specified value. Users cannot add new\nattributes, they can only choose from the ones that come packaged and\npre-supported by the platform. Details of attributes are covered in the\n Operation and Installation\nGuide.",
+ "text": "This section introduces the API to write a streaming application.\nThe work involves connecting operators via streams to form the logical\nDAG. The steps are Instantiate an application (DAG) (Optional) Set Attributes Assign application name Set any other attributes as per application requirements Create/re-use and instantiate operators Assign operator name that is unique within the application Declare schema upfront for each operator (and thereby its ports) (Optional) Set properties\u00a0 and attributes on the dag as per specification Connect ports of operators via streams Each stream connects one output port of an operator to one or more input ports of other operators. (Optional) Set attributes on the streams Test the application. There are two methods to create an application, namely Java, and\nProperties file. Java API is for applications being developed by humans,\nand properties file (Hadoop like) is more suited for DAGs generated by\ntools.",
"title": "Application API"
},
{
+ "location": "/application_development/#java-api",
+ "text": "The Java API is the most common way to create a streaming\napplication. It is meant for application developers who prefer to\nleverage the features of Java, and the ease of use and enhanced\nproductivity provided by IDEs like NetBeans or Eclipse. Using Java to\nspecify the application provides extra validation abilities of Java\ncompiler, such as compile time checks for type safety at the time of\nwriting the code. Later in this chapter you can read more about\nvalidation support in the platform. The developer specifies the streaming application by implementing\nthe ApplicationFactory interface, which is how platform tools (CLI etc.)\nrecognize and instantiate applications. Here we show how to create a\nYahoo! Finance application that streams the last trade price of a ticker\nand computes the high and low price in every 1 min window. Run above\n test application\u00a0to execute the\nDAG in local mode within the IDE. Let us revisit how the Yahoo! Finance test application constructs the DAG: public class Application implements StreamingApplication\n{\n\n ...\n\n @Override\n public void populateDAG(DAG dag, Configuration conf)\n {\n dag.getAttributes().attr(DAG.STRAM_WINDOW_SIZE_MILLIS).set(streamingWindowSizeMilliSeconds);\n\n StockTickInput tick = getStockTickInputOperator( StockTickInput , dag);\n SumKeyVal String, Long dailyVolume = getDailyVolumeOperator( DailyVolume , dag);\n ConsolidatorKeyVal String,Double,Long,String,?,? quoteOperator = getQuoteOperator( Quote , dag);\n\n RangeKeyVal String, Double highlow = getHighLowOperator( HighLow , dag, appWindowCountMinute);\n SumKeyVal String, Long minuteVolume = getMinuteVolumeOperator( MinuteVolume , dag, appWindowCountMinute);\n ConsolidatorKeyVal String,HighLow,Long,?,?,? chartOperator = getChartOperator( Chart , dag);\n\n SimpleMovingAverage String, Double priceSMA = getPriceSimpleMovingAverageOperator( PriceSMA , dag, appWindowCountSMA);\n\n dag.addStream( price , tick.price, quoteOperator.in1, highlow.data, priceSMA.data);\n dag.addStream( vol , tick.volume, dailyVolume.data, minuteVolume.data);\n dag.addStream( time , tick.time, quoteOperator.in3);\n dag.addStream( daily_vol , dailyVolume.sum, quoteOperator.in2);\n\n dag.addStream( quote_data , quoteOperator.out, getConsole( quoteConsole , dag, QUOTE ));\n\n dag.addStream( high_low , highlow.range, chartOperator.in1);\n dag.addStream( vol_1min , minuteVolume.sum, chartOperator.in2);\n dag.addStream( chart_data , chartOperator.out, getConsole( chartConsole , dag, CHART ));\n\n dag.addStream( sma_price , priceSMA.doubleSMA, getConsole( priceSMAConsole , dag, Price SMA ));\n\n return dag;\n }\n}",
+ "title": "Java API"
+ },
+ {
+ "location": "/application_development/#property-file-api",
+ "text": "The platform also supports specification of a DAG via a property\nfile. The aim here to make it easy for tools to create and run an\napplication. This method of specification does not have the Java\ncompiler support of compile time check, but since these applications\nwould be created by software, they should be correct by construction.\nThe syntax is derived from Hadoop properties and should be easy for\nfolks who are used to creating software that integrated with\nHadoop. Create an application (DAG): myApplication.properties # input operator that reads from a file\ndt.operator.inputOp.classname=com.acme.SampleInputOperator\ndt.operator.inputOp.fileName=somefile.txt\n\n# output operator that writes to the console\ndt.operator.outputOp.classname=com.acme.ConsoleOutputOperator\n\n# stream connecting both operators\ndt.stream.inputStream.source=inputOp.outputPort\ndt.stream.inputStream.sinks=outputOp.inputPort Above snippet is intended to convey the basic idea of specifying\nthe DAG without using Java. Operators would come from a predefined\nlibrary and referenced in the specification by class name and port names\n(obtained from the library providers documentation or runtime\nintrospection by tools). For those interested in details, see later\nsections and refer to the Operation and\nInstallation Guide\u00a0mentioned above.",
+ "title": "Property File API"
+ },
+ {
+ "location": "/application_development/#attributes",
+ "text": "Attributes impact the runtime behavior of the application. They do\nnot impact the functionality. An example of an attribute is application\nname. Setting it changes the application name. Another example is\nstreaming window size. Setting it changes the streaming window size from\nthe default value to the specified value. Users cannot add new\nattributes, they can only choose from the ones that come packaged and\npre-supported by the platform. Details of attributes are covered in the\n Operation and Installation\nGuide.",
+ "title": "Attributes"
+ },
+ {
"location": "/application_development/#operators",
- "text": "Operators\u00a0are basic compute units.\nOperators process each incoming tuple and emit zero or more tuples on\noutput ports as per the business logic. The data flow, connectivity,\nfault tolerance (node outage), etc. is taken care of by the platform. As\nan operator developer, all that is needed is to figure out what to do\nwith the incoming tuple and when (and which output port) to send out a\nparticular output tuple. Correctly designed operators will most likely\nget reused. Operator design needs care and foresight. For details, refer\nto the Operator Developer Guide . As an application developer you need to connect operators\nin a way that it implements your business logic. You may also require\noperator customization for functionality and use attributes for\nperformance/scalability etc. All operators process tuples asynchronously in a distributed\ncluster. An operator cannot assume or predict the exact time a tuple\nthat it emitted will get consumed by a downstream operator. An operator\nalso cannot predict the exact time when a tuple arrives from an upstream\noperator. The only guarantee is that the upstream operators are\nprocessing the current or a future window, i.e. the windowId of upstream\noperator is equals or exceeds its own windowId. Conversely the windowId\nof a downstream operator is less than or equals its own windowId. The\nend of a window operation, i.e. the API call to endWindow on an operator\nrequires that all upstream operators have finished processing this\nwindow. This means that completion of processing a window propagates in\na blocking fashion through an operator. Later sections provides more\ndetails on streams and data flow of tuples. Each operator has a unique name within the DAG as provided by the\nuser. This is the name of the operator in the logical plan. The name of\nthe operator in the physical plan is an integer assigned to it by STRAM.\nThese integers are use the sequence from 1 to N, where N is total number\nof physically unique operators in the DAG. \u00a0Following the same rule,\neach partitioned instance of a logical operator has its own integer as\nan id. This id along with the Hadoop container name uniquely identifies\nthe operator in the execution plan of the DAG. The logical names and the\nphysical names are required for web service support. Operators can be\naccessed via both names. These same names are used while interacting\nwith dtcli\u00a0to access an operator.\nIdeally these names should be self-descriptive. For example in Figure 1,\nthe node named \u201cDaily volume\u201d has a physical identifier of 2. Operator Interface Operator interface in a DAG consists of ports,\u00a0properties,\u00a0and attributes.\nOperators interact with other components of the DAG via ports. Functional behavior of the operators\ncan be customized via parameters. Run time performance and physical\ninstantiation is controlled by attributes. Ports and parameters are\nfields (variables) of the Operator class/object, while attributes are\nmeta information that is attached to the operator object via an\nAttributeMap. An operator must have at least one port. Properties are\noptional. Attributes are provided by the platform and always have a\ndefault value that enables normal functioning of operators. Ports Ports are connection points by which an operator receives and\nemits tuples. These should be transient objects instantiated in the\noperator object, that implement particular interfaces. Ports should be\ntransient as they contain no state. They have a pre-defined schema and\ncan only be connected to other ports with the same schema. An input port\nneeds to implement the interface Operator.InputPort\u00a0and\ninterface Sink. A default\nimplementation of these is provided by the abstract class DefaultInputPort. An output port needs to\nimplement the interface Operator.OutputPort. A default implementation\nof this is provided by the concrete class DefaultOutputPort. These two are a quick way to\nimplement the above interfaces, but operator developers have the option\nof providing their own implementations. Here are examples of an input and an output port from the operator\nSum. @InputPortFieldAnnotation(name = data )\npublic final transient DefaultInputPort V data = new DefaultInputPort V () {\n @Override\n public void process(V tuple)\n {\n ...\n }\n}\n@OutputPortFieldAnnotation(optional=true)\npublic final transient DefaultOutputPort V sum = new DefaultOutputPort V (){ \u2026 }; The process call is in the Sink interface. An emit on an output\nport is done via emit(tuple) call. For the above example it would be\nsum.emit(t), where the type of t is the generic parameter V. There is no limit on how many ports an operator can have. However\nany operator must have at least one port. An operator with only one port\nis called an Input Adapter if it has no input port and an Output Adapter\nif it has no output port. These are special operators needed to get/read\ndata from outside system/source into the application, or push/write data\ninto an outside system/sink. These could be in Hadoop or outside of\nHadoop. These two operators are in essence gateways for the streaming\napplication to communicate with systems outside the application. Port connectivity can be validated during compile time by adding\nPortFieldAnnotations shown above. By default all ports have to be\nconnected, to allow a port to go unconnected, you need to add\n\u201coptional=true\u201d to the annotation. Attributes can be specified for ports that affect the runtime\nbehavior. An example of an attribute is parallel partition that specifes\na parallel computation flow per partition. It is described in detail in\nthe Parallel Partitions section. Another example is queue capacity that specifies the buffer size for the\nport. Details of attributes are covered in Operation and Installation Guide. Properties Properties are the abstractions by which functional behavior of an\noperator can be customized. They should be non-transient objects\ninstantiated in the operator object. They need to be non-transient since\nthey are part of the operator state and re-construction of the operator\nobject from its checkpointed state must restore the operator to the\ndesired state. Properties are optional, i.e. an operator may or may not\nhave properties; they are part of user code and their values are not\ninterpreted by the platform in any way. All non-serializable objects should be declared transient.\nExamples include sockets, session information, etc. These objects should\nbe initialized during setup call, which is called every time the\noperator is initialized. Attributes Attributes are values assigned to the operators that impact\nrun-time. This includes things like the number of partitions, at most\nonce or at least once or exactly once recovery modes, etc. Attributes do\nnot impact functionality of the operator. Users can change certain\nattributes in runtime. Users cannot add attributes to operators; they\nare pre-defined by the platform. They are interpreted by the platform\nand thus cannot be defined in user created code (like properties).\nDetails of attributes are covered in Configuration Guide . Operator State The state of an operator is defined as the data that it transfers\nfrom one window to a future window. Since the computing model of the\nplatform is to treat windows like micro-batches, the operator state can\nbe checkpointed every Nth window, or every T units of time, where T is significantly greater\nthan the streaming window. \u00a0When an operator is checkpointed, the entire\nobject is written to HDFS. \u00a0The larger the amount of state in an\noperator, the longer it takes to recover from a failure. A stateless\noperator can recover much quicker than a stateful one. The needed\nwindows are preserved by the upstream buffer server and are used to\nrecompute the lost windows, and also rebuild the buffer server in the\ncurrent container. The distinction between Stateless and Stateful is based solely on\nthe need to transfer data in the operator from one window to the next.\nThe state of an operator is independent of the number of ports. Stateless A Stateless operator is defined as one where no data is needed to\nbe kept at the end of every window. This means that all the computations\nof a window can be derived from all the tuples the operator receives\nwithin that window. This guarantees that the output of any window can be\nreconstructed by simply replaying the tuples that arrived in that\nwindow. Stateless operators are more efficient in terms of fault\ntolerance, and cost to achieve SLA. Stateful A Stateful operator is defined as one where data is needed to be\nstored at the end of a window for computations occurring in later\nwindow; a common example is the computation of a sum of values in the\ninput tuples. Operator API The Operator API consists of methods that operator developers may\nneed to override. In this section we will discuss the Operator APIs from\nthe point of view of an application developer. Knowledge of how an\noperator works internally is critical for writing an application. Those\ninterested in the details should refer to Malhar Operator Developer Guide. The APIs are available in three modes, namely Single Streaming\nWindow, Sliding Application Window, and Aggregate Application Window.\nThese are not mutually exclusive, i.e. an operator can use single\nstreaming window as well as sliding application window. A physical\ninstance of an operator is always processing tuples from a single\nwindow. The processing of tuples is guaranteed to be sequential, no\nmatter which input port the tuples arrive on. In the later part of this section we will evaluate three common\nuses of streaming windows by applications. They have different\ncharacteristics and implications on optimization and recovery mechanisms\n(i.e. algorithm used to recover a node after outage) as discussed later\nin the section. Streaming Window Streaming window is atomic micro-batch computation period. The API\nmethods relating to a streaming window are as follows public void process( tuple_type tuple) // Called on the input port on which the tuple arrives\npublic void beginWindow(long windowId) // Called at the start of the window as soon as the first begin_window tuple arrives\npublic void endWindow() // Called at the end of the window after end_window tuples arrive on all input ports\npublic void setup(OperatorContext context) // Called once during initialization of the operator\npublic void teardown() // Called once when the operator is being shutdown A tuple can be emitted in any of the three streaming run-time\ncalls, namely beginWindow, process, and endWindow but not in setup or\nteardown. Aggregate Application Window An operator with an aggregate window is stateful within the\napplication window timeframe and possibly stateless at the end of that\napplication window. An size of an aggregate application window is an\noperator attribute and is defined as a multiple of the streaming window\nsize. The platform recognizes this attribute and optimizes the operator.\nThe beginWindow, and endWindow calls are not invoked for those streaming\nwindows that do not align with the application window. For example in\ncase of streaming window of 0.5 second and application window of 5\nminute, an application window spans 600 streaming windows (5*60*2 =\n600). At the start of the sequence of these 600 atomic streaming\nwindows, a beginWindow gets invoked, and at the end of these 600\nstreaming windows an endWindow gets invoked. All the intermediate\nstreaming windows do not invoke beginWindow or endWindow. Bookkeeping,\nnode recovery, stats, UI, etc. continue to work off streaming windows.\nFor example if operators are being checkpointed say on an average every\n30th window, then the above application window would have about 20\ncheckpoints. Sliding Application Window A sliding window is computations that requires previous N\nstreaming windows. After each streaming window the Nth past window is\ndropped and the new window is added to the computation. An operator with\nsliding window is a stateful operator at end of any window. The sliding\nwindow period is an attribute and is a multiple of streaming window. The\nplatform recognizes this attribute and leverages it during bookkeeping.\nA sliding aggregate window with tolerance to data loss does not have a\nvery high bookkeeping cost. The cost of all three recovery mechanisms,\n at most once\u00a0(data loss tolerant),\nat least once\u00a0(data loss\nintolerant), and exactly once\u00a0(data\nloss intolerant and no extra computations) is same as recovery\nmechanisms based on streaming window. STRAM is not able to leverage this\noperator for any extra optimization. Single vs Multi-Input Operator A single-input operator by definition has a single upstream\noperator, since there can only be one writing port for a stream. \u00a0If an\noperator has a single upstream operator, then the beginWindow on the\nupstream also blocks the beginWindow of the single-input operator. For\nan operator to start processing any window at least one upstream\noperator has to start processing that window. A multi-input operator\nreads from more than one upstream ports. Such an operator would start\nprocessing as soon as the first begin_window event arrives. However the\nwindow would not close (i.e. invoke endWindow) till all ports receive\nend_window events for that windowId. Thus the end of a window is a\nblocking event. As we saw earlier, a multi-input operator is also the\npoint in the DAG where windows of all upstream operators are\nsynchronized. The windows (atomic micro-batches) from a faster (or just\nahead in processing) upstream operators are queued up till the slower\nupstream operator catches up. STRAM monitors such bottlenecks and takes\ncorrective actions. The platform ensures minimal delay, i.e processing\nstarts as long as at least one upstream operator has started\nprocessing. Recovery Mechanisms Application developers can set any of the recovery mechanisms\nbelow to deal with node outage. In general, the cost of recovery depends\non the state of the operator, while data integrity is dependant on the\napplication. The mechanisms are per window as the platform treats\nwindows as atomic compute units. Three recovery mechanisms are\nsupported, namely At-least-once: All atomic batches are processed at least once.\n No data loss occurs. At-most-once: All atomic batches are processed at most once.\n Data loss is possible; this is the most efficient setting. Exactly-once: All atomic batches are processed exactly once.\n No data loss occurs; this is the least efficient setting since\n additional work is needed to ensure proper semantics. At-least-once is the default. During a recovery event, the\noperator connects to the upstream buffer server and asks for windows to\nbe replayed. At-least-once and exactly-once mechanisms start from its\ncheckpointed state. At-most-once starts from the next begin-window\nevent. Recovery mechanisms can be specified per Operator while writing\nthe application as shown below. Operator o = dag.addOperator(\u201coperator\u201d, \u2026);\ndag.setAttribute(o, OperatorContext.PROCESSING_MODE, ProcessingMode.AT_MOST_ONCE); Also note that once an operator is attributed to AT_MOST_ONCE,\nall the operators downstream to it have to be AT_MOST_ONCE. The client\nwill give appropriate warnings or errors if that\u2019s not the case. Details are explained in the chapter on Fault Tolerance below.",
+ "text": "Operators\u00a0are basic compute units.\nOperators process each incoming tuple and emit zero or more tuples on\noutput ports as per the business logic. The data flow, connectivity,\nfault tolerance (node outage), etc. is taken care of by the platform. As\nan operator developer, all that is needed is to figure out what to do\nwith the incoming tuple and when (and which output port) to send out a\nparticular output tuple. Correctly designed operators will most likely\nget reused. Operator design needs care and foresight. For details, refer\nto the Operator Developer Guide . As an application developer you need to connect operators\nin a way that it implements your business logic. You may also require\noperator customization for functionality and use attributes for\nperformance/scalability etc. All operators process tuples asynchronously in a distributed\ncluster. An operator cannot assume or predict the exact time a tuple\nthat it emitted will get consumed by a downstream operator. An operator\nalso cannot predict the exact time when a tuple arrives from an upstream\noperator. The only guarantee is that the upstream operators are\nprocessing the current or a future window, i.e. the windowId of upstream\noperator is equals or exceeds its own windowId. Conversely the windowId\nof a downstream operator is less than or equals its own windowId. The\nend of a window operation, i.e. the API call to endWindow on an operator\nrequires that all upstream operators have finished processing this\nwindow. This means that completion of processing a window propagates in\na blocking fashion through an operator. Later sections provides more\ndetails on streams and data flow of tuples. Each operator has a unique name within the DAG as provided by the\nuser. This is the name of the operator in the logical plan. The name of\nthe operator in the physical plan is an integer assigned to it by STRAM.\nThese integers are use the sequence from 1 to N, where N is total number\nof physically unique operators in the DAG. \u00a0Following the same rule,\neach partitioned instance of a logical operator has its own integer as\nan id. This id along with the Hadoop container name uniquely identifies\nthe operator in the execution plan of the DAG. The logical names and the\nphysical names are required for web service support. Operators can be\naccessed via both names. These same names are used while interacting\nwith dtcli\u00a0to access an operator.\nIdeally these names should be self-descriptive. For example in Figure 1,\nthe node named \u201cDaily volume\u201d has a physical identifier of 2.",
"title": "Operators"
},
{
+ "location": "/application_development/#operator-interface",
+ "text": "Operator interface in a DAG consists of ports,\u00a0properties,\u00a0and attributes.\nOperators interact with other components of the DAG via ports. Functional behavior of the operators\ncan be customized via parameters. Run time performance and physical\ninstantiation is controlled by attributes. Ports and parameters are\nfields (variables) of the Operator class/object, while attributes are\nmeta information that is attached to the operator object via an\nAttributeMap. An operator must have at least one port. Properties are\noptional. Attributes are provided by the platform and always have a\ndefault value that enables normal functioning of operators.",
+ "title": "Operator Interface"
+ },
+ {
+ "location": "/application_development/#ports",
+ "text": "Ports are connection points by which an operator receives and\nemits tuples. These should be transient objects instantiated in the\noperator object, that implement particular interfaces. Ports should be\ntransient as they contain no state. They have a pre-defined schema and\ncan only be connected to other ports with the same schema. An input port\nneeds to implement the interface Operator.InputPort\u00a0and\ninterface Sink. A default\nimplementation of these is provided by the abstract class DefaultInputPort. An output port needs to\nimplement the interface Operator.OutputPort. A default implementation\nof this is provided by the concrete class DefaultOutputPort. These two are a quick way to\nimplement the above interfaces, but operator developers have the option\nof providing their own implementations. Here are examples of an input and an output port from the operator\nSum. @InputPortFieldAnnotation(name = data )\npublic final transient DefaultInputPort V data = new DefaultInputPort V () {\n @Override\n public void process(V tuple)\n {\n ...\n }\n}\n@OutputPortFieldAnnotation(optional=true)\npublic final transient DefaultOutputPort V sum = new DefaultOutputPort V (){ \u2026 }; The process call is in the Sink interface. An emit on an output\nport is done via emit(tuple) call. For the above example it would be\nsum.emit(t), where the type of t is the generic parameter V. There is no limit on how many ports an operator can have. However\nany operator must have at least one port. An operator with only one port\nis called an Input Adapter if it has no input port and an Output Adapter\nif it has no output port. These are special operators needed to get/read\ndata from outside system/source into the application, or push/write data\ninto an outside system/sink. These could be in Hadoop or outside of\nHadoop. These two operators are in essence gateways for the streaming\napplication to communicate with systems outside the application. Port connectivity can be validated during compile time by adding\nPortFieldAnnotations shown above. By default all ports have to be\nconnected, to allow a port to go unconnected, you need to add\n\u201coptional=true\u201d to the annotation. Attributes can be specified for ports that affect the runtime\nbehavior. An example of an attribute is parallel partition that specifes\na parallel computation flow per partition. It is described in detail in\nthe Parallel Partitions section. Another example is queue capacity that specifies the buffer size for the\nport. Details of attributes are covered in Operation and Installation Guide.",
+ "title": "Ports"
+ },
+ {
+ "location": "/application_development/#properties",
+ "text": "Properties are the abstractions by which functional behavior of an\noperator can be customized. They should be non-transient objects\ninstantiated in the operator object. They need to be non-transient since\nthey are part of the operator state and re-construction of the operator\nobject from its checkpointed state must restore the operator to the\ndesired state. Properties are optional, i.e. an operator may or may not\nhave properties; they are part of user code and their values are not\ninterpreted by the platform in any way. All non-serializable objects should be declared transient.\nExamples include sockets, session information, etc. These objects should\nbe initialized during setup call, which is called every time the\noperator is initialized.",
+ "title": "Properties"
+ },
+ {
+ "location": "/application_development/#attributes_1",
+ "text": "Attributes are values assigned to the operators that impact\nrun-time. This includes things like the number of partitions, at most\nonce or at least once or exactly once recovery modes, etc. Attributes do\nnot impact functionality of the operator. Users can change certain\nattributes in runtime. Users cannot add attributes to operators; they\nare pre-defined by the platform. They are interpreted by the platform\nand thus cannot be defined in user created code (like properties).\nDetails of attributes are covered in Configuration Guide .",
+ "title": "Attributes"
+ },
+ {
+ "location": "/application_development/#operator-state",
+ "text": "The state of an operator is defined as the data that it transfers\nfrom one window to a future window. Since the computing model of the\nplatform is to treat windows like micro-batches, the operator state can\nbe checkpointed every Nth window, or every T units of time, where T is significantly greater\nthan the streaming window. \u00a0When an operator is checkpointed, the entire\nobject is written to HDFS. \u00a0The larger the amount of state in an\noperator, the longer it takes to recover from a failure. A stateless\noperator can recover much quicker than a stateful one. The needed\nwindows are preserved by the upstream buffer server and are used to\nrecompute the lost windows, and also rebuild the buffer server in the\ncurrent container. The distinction between Stateless and Stateful is based solely on\nthe need to transfer data in the operator from one window to the next.\nThe state of an operator is independent of the number of ports.",
+ "title": "Operator State"
+ },
+ {
+ "location": "/application_development/#stateless",
+ "text": "A Stateless operator is defined as one where no data is needed to\nbe kept at the end of every window. This means that all the computations\nof a window can be derived from all the tuples the operator receives\nwithin that window. This guarantees that the output of any window can be\nreconstructed by simply replaying the tuples that arrived in that\nwindow. Stateless operators are more efficient in terms of fault\ntolerance, and cost to achieve SLA.",
+ "title": "Stateless"
+ },
+ {
+ "location": "/application_development/#stateful",
+ "text": "A Stateful operator is defined as one where data is needed to be\nstored at the end of a window for computations occurring in later\nwindow; a common example is the computation of a sum of values in the\ninput tuples.",
+ "title": "Stateful"
+ },
+ {
+ "location": "/application_development/#operator-api",
+ "text": "The Operator API consists of methods that operator developers may\nneed to override. In this section we will discuss the Operator APIs from\nthe point of view of an application developer. Knowledge of how an\noperator works internally is critical for writing an application. Those\ninterested in the details should refer to Malhar Operator Developer Guide. The APIs are available in three modes, namely Single Streaming\nWindow, Sliding Application Window, and Aggregate Application Window.\nThese are not mutually exclusive, i.e. an operator can use single\nstreaming window as well as sliding application window. A physical\ninstance of an operator is always processing tuples from a single\nwindow. The processing of tuples is guaranteed to be sequential, no\nmatter which input port the tuples arrive on. In the later part of this section we will evaluate three common\nuses of streaming windows by applications. They have different\ncharacteristics and implications on optimization and recovery mechanisms\n(i.e. algorithm used to recover a node after outage) as discussed later\nin the section.",
+ "title": "Operator API"
+ },
+ {
+ "location": "/application_development/#streaming-window",
+ "text": "Streaming window is atomic micro-batch computation period. The API\nmethods relating to a streaming window are as follows public void process( tuple_type tuple) // Called on the input port on which the tuple arrives\npublic void beginWindow(long windowId) // Called at the start of the window as soon as the first begin_window tuple arrives\npublic void endWindow() // Called at the end of the window after end_window tuples arrive on all input ports\npublic void setup(OperatorContext context) // Called once during initialization of the operator\npublic void teardown() // Called once when the operator is being shutdown A tuple can be emitted in any of the three streaming run-time\ncalls, namely beginWindow, process, and endWindow but not in setup or\nteardown.",
+ "title": "Streaming Window"
+ },
+ {
+ "location": "/application_development/#aggregate-application-window",
+ "text": "An operator with an aggregate window is stateful within the\napplication window timeframe and possibly stateless at the end of that\napplication window. An size of an aggregate application window is an\noperator attribute and is defined as a multiple of the streaming window\nsize. The platform recognizes this attribute and optimizes the operator.\nThe beginWindow, and endWindow calls are not invoked for those streaming\nwindows that do not align with the application window. For example in\ncase of streaming window of 0.5 second and application window of 5\nminute, an application window spans 600 streaming windows (5*60*2 =\n600). At the start of the sequence of these 600 atomic streaming\nwindows, a beginWindow gets invoked, and at the end of these 600\nstreaming windows an endWindow gets invoked. All the intermediate\nstreaming windows do not invoke beginWindow or endWindow. Bookkeeping,\nnode recovery, stats, UI, etc. continue to work off streaming windows.\nFor example if operators are being checkpointed say on an average every\n30th window, then the above application window would have about 20\ncheckpoints.",
+ "title": "Aggregate Application Window"
+ },
+ {
+ "location": "/application_development/#sliding-application-window",
+ "text": "A sliding window is computations that requires previous N\nstreaming windows. After each streaming window the Nth past window is\ndropped and the new window is added to the computation. An operator with\nsliding window is a stateful operator at end of any window. The sliding\nwindow period is an attribute and is a multiple of streaming window. The\nplatform recognizes this attribute and leverages it during bookkeeping.\nA sliding aggregate window with tolerance to data loss does not have a\nvery high bookkeeping cost. The cost of all three recovery mechanisms,\n at most once\u00a0(data loss tolerant),\nat least once\u00a0(data loss\nintolerant), and exactly once\u00a0(data\nloss intolerant and no extra computations) is same as recovery\nmechanisms based on streaming window. STRAM is not able to leverage this\noperator for any extra optimization.",
+ "title": "Sliding Application Window"
+ },
+ {
+ "location": "/application_development/#single-vs-multi-input-operator",
+ "text": "A single-input operator by definition has a single upstream\noperator, since there can only be one writing port for a stream. \u00a0If an\noperator has a single upstream operator, then the beginWindow on the\nupstream also blocks the beginWindow of the single-input operator. For\nan operator to start processing any window at least one upstream\noperator has to start processing that window. A multi-input operator\nreads from more than one upstream ports. Such an operator would start\nprocessing as soon as the first begin_window event arrives. However the\nwindow would not close (i.e. invoke endWindow) till all ports receive\nend_window events for that windowId. Thus the end of a window is a\nblocking event. As we saw earlier, a multi-input operator is also the\npoint in the DAG where windows of all upstream operators are\nsynchronized. The windows (atomic micro-batches) from a faster (or just\nahead in processing) upstream operators are queued up till the slower\nupstream operator catches up. STRAM monitors such bottlenecks and takes\ncorrective actions. The platform ensures minimal delay, i.e processing\nstarts as long as at least one upstream operator has started\nprocessing.",
+ "title": "Single vs Multi-Input Operator"
+ },
+ {
+ "location": "/application_development/#recovery-mechanisms",
+ "text": "Application developers can set any of the recovery mechanisms\nbelow to deal with node outage. In general, the cost of recovery depends\non the state of the operator, while data integrity is dependant on the\napplication. The mechanisms are per window as the platform treats\nwindows as atomic compute units. Three recovery mechanisms are\nsupported, namely At-least-once: All atomic batches are processed at least once.\n No data loss occurs. At-most-once: All atomic batches are processed at most once.\n Data loss is possible; this is the most efficient setting. Exactly-once: All atomic batches are processed exactly once.\n No data loss occurs; this is the least efficient setting since\n additional work is needed to ensure proper semantics. At-least-once is the default. During a recovery event, the\noperator connects to the upstream buffer server and asks for windows to\nbe replayed. At-least-once and exactly-once mechanisms start from its\ncheckpointed state. At-most-once starts from the next begin-window\nevent. Recovery mechanisms can be specified per Operator while writing\nthe application as shown below. Operator o = dag.addOperator(\u201coperator\u201d, \u2026);\ndag.setAttribute(o, OperatorContext.PROCESSING_MODE, ProcessingMode.AT_MOST_ONCE); Also note that once an operator is attributed to AT_MOST_ONCE,\nall the operators downstream to it have to be AT_MOST_ONCE. The client\nwill give appropriate warnings or errors if that\u2019s not the case. Details are explained in the chapter on Fault Tolerance below.",
+ "title": "Recovery Mechanisms"
+ },
+ {
"location": "/application_development/#streams",
"text": "A stream\u00a0is a connector\n(edge) abstraction, and is a fundamental building block of the platform.\nA stream consists of tuples that flow from one port (called the\noutput\u00a0port) to one or more ports\non other operators (called input\u00a0ports) another -- so note a potentially\nconfusing aspect of this terminology: tuples enter a stream through its\noutput port and leave via one or more input ports. A stream has the\nfollowing characteristics Tuples are always delivered in the same order in which they\n were emitted. Consists of a sequence of windows one after another. Each\n window being a collection of in-order tuples. A stream that connects two containers passes through a\n buffer server. All streams can be persisted (by default in HDFS). Exactly one output port writes to the stream. Can be read by one or more input ports. Connects operators within an application, not outside\n an application. Has an unique name within an application. Has attributes which act as hints to STRAM. Streams have four modes, namely in-line, in-node, in-rack,\n and other. Modes may be overruled (for example due to lack\n of containers). They are defined as follows: THREAD_LOCAL: In the same thread, uses thread\n stack (intra-thread). This mode can only be used for a downstream\n operator which has only one input port connected; also called\n in-line. CONTAINER_LOCAL: In the same container (intra-process); also\n called in-container. NODE_LOCAL: In the same Hadoop node (inter processes, skips\n NIC); also called in-node. RACK_LOCAL: On nodes in the same rack; also called\n in-rack. unspecified: No guarantee. Could be anywhere within the\n cluster An example of a stream declaration is given below DAG dag = new DAG();\n \u2026\ndag.addStream( views , viewAggregate.sum, cost.data).setLocality(CONTAINER_LOCAL); // A container local stream\ndag.addStream(\u201cclicks\u201d, clickAggregate.sum, rev.data); // An example of unspecified locality The platform guarantees in-order delivery of tuples in a stream.\nSTRAM views each stream as collection of ordered windows. Since no tuple\ncan exist outside a window, a replay of a stream consists of replay of a\nset of windows. When multiple input ports read the same stream, the\nexecution plan of a stream ensures that each input port is logically not\nblocked by the reading of another input port. The schema of a stream is\nsame as the schema of the tuple. In a stream all tuples emitted by an operator in a window belong\nto that window. A replay of this window would consists of an in-order\nreplay of all the tuples. Thus the tuple order within a stream is\nguaranteed. However since an operator may receive multiple streams (for\nexample an operator with two input ports), the order of arrival of two\ntuples belonging to different streams is not guaranteed. In general in\nan asynchronous distributed architecture this is expected. Thus the\noperator (specially one with multiple input ports) should not depend on\nthe tuple order from two streams. One way to cope with this\nindeterminate order, if necessary, is to wait to get all the tuples of a\nwindow and emit results in endWindow call. All operator templates\nprovided as part of Malhar operator library follow these principles. A logical stream gets partitioned into physical streams each\nconnecting the partition to the upstream operator. If two different\nattributes are needed on the same stream, it should be split using\nStreamDuplicator\u00a0operator. Modes of the streams are critical for performance. An in-line\nstream is the most optimal as it simply delivers the tuple as-is without\nserialization-deserialization. Streams should be marked\ncontainer_local, specially in case where there is a large tuple volume\nbetween two operators which then on drops significantly. Since the\nsetLocality call merely provides a hint, STRAM may ignore it. An In-node\nstream is not as efficient as an in-line one, but it is clearly better\nthan going off-node since it still avoids the potential bottleneck of\nthe network card. THREAD_LOCAL and CONTAINER_LOCAL streams do not use a buffer\nserver as this stream is in a single process. The other two do.",
"title": "Streams"
},
{
"location": "/application_development/#validating-an-application",
- "text": "The platform provides various ways of validating the application\nspecification and data input. An understanding of these checks is very\nimportant for an application developer since it affects productivity.\nValidation of an application is done in three phases, namely Compile Time: Caught during application development, and is\n most cost effective. These checks are mainly done on declarative\n objects and leverages the Java compiler. An example is checking that\n the schemas specified on all ports of a stream are\n mutually compatible. Initialization Time: When the application is being\n initialized, before submitting to Hadoop. These checks are related\n to configuration/context of an application, and are done by the\n logical DAG builder implementation. An example is the checking that\n all non-optional ports are connected to other ports. Run Time: Validations done when the application is running.\n This is the costliest of all checks. These are checks that can only\n be done at runtime as they involve data. For example divide by 0\n check as part of business logic. Compile Time Compile time validations apply when an application is specified in\nJava code and include all checks that can be done by Java compiler in\nthe development environment (including IDEs like NetBeans or Eclipse).\nExamples include Schema Validation: The tuples on ports are POJO (plain old\n java objects) and compiler checks to ensure that all the ports on a\n stream have the same schema. Stream Check: Single Output Port and at least one Input port\n per stream. A stream can only have one output port writer. This is\n part of the addStream api. This\n check ensures that developers only connect one output port to\n a stream. The same signature also ensures that there is at least one\n input port for a stream Naming: Compile time checks ensures that applications\n components operators, streams are named Initialization/Instantiation Time Initialization time validations include various checks that are\ndone post compile, and before the application starts running in a\ncluster (or local mode). These are mainly configuration/contextual in\nnature. These checks are as critical to proper functionality of the\napplication as the compile time validations. Examples include JavaBeans Validation :\n Examples include @Max(): Value must be less than or equal to the number @Min(): Value must be greater than or equal to the\n number @NotNull: The value of the field or property must not be\n null @Pattern(regexp = \u201c....\u201d): Value must match the regular\n expression Input port connectivity: By default, every non-optional input\n port must be connected. A port can be declared optional by using an\n annotation: \u00a0 \u00a0 @InputPortFieldAnnotation(name = \"...\", optional\n = true) Output Port Connectivity: Similar. The annotation here is: \u00a0 \u00a0\n @OutputPortFieldAnnotation(name = \"...\", optional = true) Unique names in application scope: Operators, streams, must have\n unique names. Cycles in the dag: DAG cannot have a cycle. Unique names in operator scope: Ports, properties, annotations\n must have unique names. One stream per port: A port can connect to only one stream.\n This check applies to input as well as output ports even though an\n output port can technically write to two streams. If you must have\n two streams originating from a single output port, use \u00a0a\u00a0streamDuplicator operator. Application Window Period: Has to be an integral multiple the\n streaming window period. Run Time Run time checks are those that are done when the application is\nrunning. The real-time streaming platform provides rich run time error\nhandling mechanisms. The checks are exclusively done by the application\nbusiness logic, but the platform allows applications to count and audit\nthese. Some of these features are in the process of development (backend\nand UI) and this section will be updated as they are developed. Upon\ncompletion examples will be added to demos to illustrate these. Error ports are output ports with error annotations. Since they\nare normal ports, they can be monitored and tuples counted, persisted\nand counts shown in the UI.",
+ "text": "The platform provides various ways of validating the application\nspecification and data input. An understanding of these checks is very\nimportant for an application developer since it affects productivity.\nValidation of an application is done in three phases, namely Compile Time: Caught during application development, and is\n most cost effective. These checks are mainly done on declarative\n objects and leverages the Java compiler. An example is checking that\n the schemas specified on all ports of a stream are\n mutually compatible. Initialization Time: When the application is being\n initialized, before submitting to Hadoop. These checks are related\n to configuration/context of an application, and are done by the\n logical DAG builder implementation. An example is the checking that\n all non-optional ports are connected to other ports. Run Time: Validations done when the application is running.\n This is the costliest of all checks. These are checks that can only\n be done at runtime as they involve data. For example divide by 0\n check as part of business logic.",
"title": "Validating an Application"
},
{
+ "location": "/application_development/#compile-time",
+ "text": "Compile time validations apply when an application is specified in\nJava code and include all checks that can be done by Java compiler in\nthe development environment (including IDEs like NetBeans or Eclipse).\nExamples include Schema Validation: The tuples on ports are POJO (plain old\n java objects) and compiler checks to ensure that all the ports on a\n stream have the same schema. Stream Check: Single Output Port and at least one Input port\n per stream. A stream can only have one output port writer. This is\n part of the addStream api. This\n check ensures that developers only connect one output port to\n a stream. The same signature also ensures that there is at least one\n input port for a stream Naming: Compile time checks ensures that applications\n components operators, streams are named",
+ "title": "Compile Time"
+ },
+ {
+ "location": "/application_development/#initializationinstantiation-time",
+ "text": "Initialization time validations include various checks that are\ndone post compile, and before the application starts running in a\ncluster (or local mode). These are mainly configuration/contextual in\nnature. These checks are as critical to proper functionality of the\napplication as the compile time validations. Examples include JavaBeans Validation :\n Examples include @Max(): Value must be less than or equal to the number @Min(): Value must be greater than or equal to the\n number @NotNull: The value of the field or property must not be\n null @Pattern(regexp = \u201c....\u201d): Value must match the regular\n expression Input port connectivity: By default, every non-optional input\n port must be connected. A port can be declared optional by using an\n annotation: \u00a0 \u00a0 @InputPortFieldAnnotation(name = \"...\", optional\n = true) Output Port Connectivity: Similar. The annotation here is: \u00a0 \u00a0\n @OutputPortFieldAnnotation(name = \"...\", optional = true) Unique names in application scope: Operators, streams, must have\n unique names. Cycles in the dag: DAG cannot have a cycle. Unique names in operator scope: Ports, properties, annotations\n must have unique names. One stream per port: A port can connect to only one stream.\n This check applies to input as well as output ports even though an\n output port can technically write to two streams. If you must have\n two streams originating from a single output port, use \u00a0a\u00a0streamDuplicator operator. Application Window Period: Has to be an integral multiple the\n streaming window period.",
+ "title": "Initialization/Instantiation Time"
+ },
+ {
+ "location": "/application_development/#run-time",
+ "text": "Run time checks are those that are done when the application is\nrunning. The real-time streaming platform provides rich run time error\nhandling mechanisms. The checks are exclusively done by the application\nbusiness logic, but the platform allows applications to count and audit\nthese. Some of these features are in the process of development (backend\nand UI) and this section will be updated as they are developed. Upon\ncompletion examples will be added to demos to illustrate these. Error ports are output ports with error annotations. Since they\nare normal ports, they can be monitored and tuples counted, persisted\nand counts shown in the UI.",
+ "title": "Run Time"
+ },
+ {
"location": "/application_development/#multi-tenancy-and-security",
"text": "Hadoop is a multi-tenant distributed operating system. Security is\nan intrinsic element of multi-tenancy as without it a cluster cannot be\nreasonably be shared among enterprise applications. Streaming\napplications follow all multi-tenancy security models used in Hadoop as\nthey are native Hadoop applications.",
"title": "Multi-Tenancy and Security"
@@ -147,10 +307,40 @@
},
{
"location": "/application_development/#partitioning",
- "text": "If all tuples sent through the stream(s) that are connected to the\ninput port(s) of an operator in the DAG are received by a single\nphysical instance of that operator, that operator can become a\nperformance bottleneck. This leads to scalability issues when\nthroughput, memory, or CPU needs exceed the processing capacity of that\nsingle instance. To address the problem, the platform offers the capability to\npartition the inflow of data so that it is divided across multiple\nphysical instances of a logical operator in the DAG. There are two\nfunctional ways to partition Load balance: Incoming load is simply partitioned\n into stream(s) that go to separate instances of physical operators\n and scalability is achieved via adding more physical operators. Each\n tuple is sent to physical operator (partition) based on a\n round-robin or other similar algorithm. This scheme scales linearly.\n A lot of key based computations can load balance in the platform due\n to the ability to insert Unifiers. For many computations, the\n endWindow and Unifier setup is similar to the combiner and reducer\n mechanism in a Map-Reduce computation. Sticky Key: The key assertion is that distribution of tuples\n are sticky, i.e the data with\n same key will always be processed by the same physical operator, no\n matter how many times it is sent through the stream. This stickiness\n will continue even if the number of partitions grows dynamically and\n can eventually be leveraged for advanced features like\n bucket testing. How this is accomplished and what is required to\n develop compliant operators will be explained below. We plan to add more partitioning mechanisms proactively to the\nplatform over time as needed by emerging usage patterns. The aim is to\nallow enterprises to be able to focus on their business logic, and\nsignificantly reduce the cost of operability. As an enabling technology\nfor managing high loads, this platform provides enterprises with a\nsignificant innovative edge. Scalability and Partitioning is a\nfoundational building block for this platform. Sticky Partition vs Round Robin As noted above, partitioning via sticky key is data aware but\nround-robin partitioning is not. An example for non-sticky load\nbalancing would be round robin distribution over multiple instances,\nwhere for example a tuple stream of A, A,\nA with 3 physical operator\ninstances would result in processing of a single A by each of the instances, In contrast, sticky\npartitioning means that exactly one instance of the operators will\nprocess all of the Atuples if they\nfall into the same bucket, while B\nmay be processed by another operator. Data aware mapping of\ntuples to partitions (similar to distributed hash table) is accomplished\nvia Stream Codecs. In later sections we would show how these two\napproaches can be used in combination. Stream Codec The platform does not make assumptions about the tuple\ntype, it could be any Java object. The operator developer knows what\ntuple type an input port expects and is capable of processing. Each\ninput port has a stream codec \u00a0associated thatdefines how data is serialized when transmitted over a socket\nstream; it also defines another\nfunction that computes the partition hash key for the tuple. The engine\nuses that key to determine which physical instance(s) \u00a0(for a\npartitioned operator) receive that \u00a0tuple. For this to work, consistent hashing is required.\nThe default codec uses the Java Object#hashCode function, which is\nsufficient for basic types such as Integer, String etc. It will also\nwork with custom tuple classes as long as they implement hashCode\nappropriately. Reliance on hashCode may not work when generic containers\nare used that do not hash the actual data, such as standard collection\nclasses (HashMap etc.), in which case a custom stream codec must be\nassigned to the input port. Static Partitioning DAG designers can specify at design time how they would like\ncertain operators to be partitioned. STRAM then instantiates the DAG\nwith the physical plan which adheres to the partitioning scheme defined\nby the design. This plan is the initial partition of the application. In\nother words, Static Partitioning is used to tell STRAM to compute the\nphysical DAG from a logical DAG once, without taking into consideration\nruntime states or loads of various operators. Dynamic Partitioning In streaming applications the load changes during the day, thus\ncreating situations where the number of partitioned operator instances\nneeds to adjust dynamically. The load can be measured in terms of\nprocessing within the DAG based on throughput, or latency, or\nconsiderations in external system components (time based etc.) that the\nplatform may not be aware of. Whatever the trigger, the resource\nrequirement for the current processing needs to be adjusted at run-time.\nThe platform may detect that operator instances are over or under\nutilized and may need to dynamically adjust the number of instances on\nthe fly. More instances of a logical operator may be required (partition\nsplit) or underutilized operator instances may need decommissioning\n(partition merge). We refer to either of the changes as dynamic\npartitioning. The default partitioning scheme supports split and merge\nof partitions, but without state transfer. The contract of the\nPartitioner\u00a0interface allows the operator\ndeveloper to implement split/merge and the associated state transfer, if\nnecessary. Since partitioning is a key scalability measure, our goal is to\nmake it as simple as possible without removing the flexibility needed\nfor sophisticated applications. Basic partitioning can be enabled at\ncompile time through the DAG specification. A slightly involved\npartitioning involves writing custom codecs to calculate data aware\npartitioning scheme. More complex partitioning cases may require users\nto provide a custom implementation of Partitioner, which gives the\ndeveloper full control over state transfer between multiple instances of\nthe partitioned operator. Default Partitioning The platform provides a default partitioning implementation that\ncan be enabled without implementing Partitioner\u00a0(or writing any other extra Java\ncode), which is designed to support simple sticky partitioning out of\nthe box for operators with logic agnostic to the partitioning scheme\nthat can be enabled by means of DAG construction alone. Typically an operator that can work with the default partitioning\nscheme would have a single input port. If there are multiple input\nports, only one port will be partitioned (the port first connected in\nthe DAG). The number of partitions will be calculated based on the\ninitial partition count - set as attribute on the operator in the DAG\n(if the attribute is not present, partitioning is off). Each partition\nwill handle tuples based on matching the lower bits of the hash code.\nFor example, if the tuple type was Integer and 2 partitions requested,\nall even numbers would go to one operator instance and all odd numbers\nto the other. Default Dynamic Partitioning Triggering partition load evaluation and repartitioning action\nitself are separate concerns. Triggers are not specified further here,\nwe are planning to support it in a customizable fashion that, for\nexample, allows latency or SLA based implementations. Triggers calculate\na load indicator (signed number) that tells the framework that a given\npartition is either underutilized, operating normally within the\nexpected thresholds or overloaded and becoming a bottleneck. The\nindicator is then presented to the partitioning logic (default or custom\nimplementation of Partitioner) to provide the opportunity to make any\nneeded adjustments. The default partitioning logic divides the key space\naccording to the lower bits of the hash codes that are generated by the\nstream codec, by assigning each partitioned operator instance via a bit\nmask and the respective value. For example, the operator may have\ninitially two partitions, 0and 1, each\nwith a bit mask of 1.\nIn the case where load evaluation flags partition\n0 as over utilized\n(most data tuples processed yield a hash code with lowest bit cleared),\napartition split\u00a0occurs, resulting in 00\nand 10with mask 11. Operator instance 0 will be replaced with 2 new instances and partition\n1 remains unchanged,\nresulting in three active partitions. The same process could repeat if\nmost tuples fall into the01 partition, leading to a split into 001 and101\nwith mask 111, etc. Should load decrease in two sibling partitions, a\npartition merge\u00a0could\nreverse the split, reducing the mask length and replacing two operators\nwith one. Should only one of two sibling partitions be underutilized,\n it cannot be\u00a0merged.\nInstead, the platform can attempt to deploy the affected operator\ninstance along with other operator instances for resource sharing\namongst underutilized partitions (not implemented yet). Keeping separate\noperator instances allows\u00a0us to\npin load increases directly to the affected instance with a single\nspecific partition key, which would not be the case had we assigned a\nshared instance to handle multiple keys.",
+ "text": "If all tuples sent through the stream(s) that are connected to the\ninput port(s) of an operator in the DAG are received by a single\nphysical instance of that operator, that operator can become a\nperformance bottleneck. This leads to scalability issues when\nthroughput, memory, or CPU needs exceed the processing capacity of that\nsingle instance. To address the problem, the platform offers the capability to\npartition the inflow of data so that it is divided across multiple\nphysical instances of a logical operator in the DAG. There are two\nfunctional ways to partition Load balance: Incoming load is simply partitioned\n into stream(s) that go to separate instances of physical operators\n and scalability is achieved via adding more physical operators. Each\n tuple is sent to physical operator (partition) based on a\n round-robin or other similar algorithm. This scheme scales linearly.\n A lot of key based computations can load balance in the platform due\n to the ability to insert Unifiers. For many computations, the\n endWindow and Unifier setup is similar to the combiner and reducer\n mechanism in a Map-Reduce computation. Sticky Key: The key assertion is that distribution of tuples\n are sticky, i.e the data with\n same key will always be processed by the same physical operator, no\n matter how many times it is sent through the stream. This stickiness\n will continue even if the number of partitions grows dynamically and\n can eventually be leveraged for advanced features like\n bucket testing. How this is accomplished and what is required to\n develop compliant operators will be explained below. We plan to add more partitioning mechanisms proactively to the\nplatform over time as needed by emerging usage patterns. The aim is to\nallow enterprises to be able to focus on their business logic, and\nsignificantly reduce the cost of operability. As an enabling technology\nfor managing high loads, this platform provides enterprises with a\nsignificant innovative edge. Scalability and Partitioning is a\nfoundational building block for this platform.",
"title": "Partitioning"
},
{
+ "location": "/application_development/#sticky-partition-vs-round-robin",
+ "text": "As noted above, partitioning via sticky key is data aware but\nround-robin partitioning is not. An example for non-sticky load\nbalancing would be round robin distribution over multiple instances,\nwhere for example a tuple stream of A, A,\nA with 3 physical operator\ninstances would result in processing of a single A by each of the instances, In contrast, sticky\npartitioning means that exactly one instance of the operators will\nprocess all of the Atuples if they\nfall into the same bucket, while B\nmay be processed by another operator. Data aware mapping of\ntuples to partitions (similar to distributed hash table) is accomplished\nvia Stream Codecs. In later sections we would show how these two\napproaches can be used in combination.",
+ "title": "Sticky Partition vs Round Robin"
+ },
+ {
+ "location": "/application_development/#stream-codec",
+ "text": "The platform does not make assumptions about the tuple\ntype, it could be any Java object. The operator developer knows what\ntuple type an input port expects and is capable of processing. Each\ninput port has a stream codec \u00a0associated thatdefines how data is serialized when transmitted over a socket\nstream; it also defines another\nfunction that computes the partition hash key for the tuple. The engine\nuses that key to determine which physical instance(s) \u00a0(for a\npartitioned operator) receive that \u00a0tuple. For this to work, consistent hashing is required.\nThe default codec uses the Java Object#hashCode function, which is\nsufficient for basic types such as Integer, String etc. It will also\nwork with custom tuple classes as long as they implement hashCode\nappropriately. Reliance on hashCode may not work when generic containers\nare used that do not hash the actual data, such as standard collection\nclasses (HashMap etc.), in which case a custom stream codec must be\nassigned to the input port.",
+ "title": "Stream Codec"
+ },
+ {
+ "location": "/application_development/#static-partitioning",
+ "text": "DAG designers can specify at design time how they would like\ncertain operators to be partitioned. STRAM then instantiates the DAG\nwith the physical plan which adheres to the partitioning scheme defined\nby the design. This plan is the initial partition of the application. In\nother words, Static Partitioning is used to tell STRAM to compute the\nphysical DAG from a logical DAG once, without taking into consideration\nruntime states or loads of various operators.",
+ "title": "Static Partitioning"
+ },
+ {
+ "location": "/application_development/#dynamic-partitioning",
+ "text": "In streaming applications the load changes during the day, thus\ncreating situations where the number of partitioned operator instances\nneeds to adjust dynamically. The load can be measured in terms of\nprocessing within the DAG based on throughput, or latency, or\nconsiderations in external system components (time based etc.) that the\nplatform may not be aware of. Whatever the trigger, the resource\nrequirement for the current processing needs to be adjusted at run-time.\nThe platform may detect that operator instances are over or under\nutilized and may need to dynamically adjust the number of instances on\nthe fly. More instances of a logical operator may be required (partition\nsplit) or underutilized operator instances may need decommissioning\n(partition merge). We refer to either of the changes as dynamic\npartitioning. The default partitioning scheme supports split and merge\nof partitions, but without state transfer. The contract of the\nPartitioner\u00a0interface allows the operator\ndeveloper to implement split/merge and the associated state transfer, if\nnecessary. Since partitioning is a key scalability measure, our goal is to\nmake it as simple as possible without removing the flexibility needed\nfor sophisticated applications. Basic partitioning can be enabled at\ncompile time through the DAG specification. A slightly involved\npartitioning involves writing custom codecs to calculate data aware\npartitioning scheme. More complex partitioning cases may require users\nto provide a custom implementation of Partitioner, which gives the\ndeveloper full control over state transfer between multiple instances of\nthe partitioned operator.",
+ "title": "Dynamic Partitioning"
+ },
+ {
+ "location": "/application_development/#default-partitioning",
+ "text": "The platform provides a default partitioning implementation that\ncan be enabled without implementing Partitioner\u00a0(or writing any other extra Java\ncode), which is designed to support simple sticky partitioning out of\nthe box for operators with logic agnostic to the partitioning scheme\nthat can be enabled by means of DAG construction alone. Typically an operator that can work with the default partitioning\nscheme would have a single input port. If there are multiple input\nports, only one port will be partitioned (the port first connected in\nthe DAG). The number of partitions will be calculated based on the\ninitial partition count - set as attribute on the operator in the DAG\n(if the attribute is not present, partitioning is off). Each partition\nwill handle tuples based on matching the lower bits of the hash code.\nFor example, if the tuple type was Integer and 2 partitions requested,\nall even numbers would go to one operator instance and all odd numbers\nto the other.",
+ "title": "Default Partitioning"
+ },
+ {
+ "location": "/application_development/#default-dynamic-partitioning",
+ "text": "Triggering partition load evaluation and repartitioning action\nitself are separate concerns. Triggers are not specified further here,\nwe are planning to support it in a customizable fashion that, for\nexample, allows latency or SLA based implementations. Triggers calculate\na load indicator (signed number) that tells the framework that a given\npartition is either underutilized, operating normally within the\nexpected thresholds or overloaded and becoming a bottleneck. The\nindicator is then presented to the partitioning logic (default or custom\nimplementation of Partitioner) to provide the opportunity to make any\nneeded adjustments. The default partitioning logic divides the key space\naccording to the lower bits of the hash codes that are generated by the\nstream codec, by assigning each partitioned operator instance via a bit\nmask and the respective value. For example, the operator may have\ninitially two partitions, 0and 1, each\nwith a bit mask of 1.\nIn the case where load evaluation flags partition\n0 as over utilized\n(most data tuples processed yield a hash code with lowest bit cleared),\napartition split\u00a0occurs, resulting in 00\nand 10with mask 11. Operator instance 0 will be replaced with 2 new instances and partition\n1 remains unchanged,\nresulting in three active partitions. The same process could repeat if\nmost tuples fall into the01 partition, leading to a split into 001 and101\nwith mask 111, etc. Should load decrease in two sibling partitions, a\npartition merge\u00a0could\nreverse the split, reducing the mask length and replacing two operators\nwith one. Should only one of two sibling partitions be underutilized,\n it cannot be\u00a0merged.\nInstead, the platform can attempt to deploy the affected operator\ninstance along with other operator instances for resource sharing\namongst underutilized partitions (not implemented yet). Keeping separate\noperator instances allows\u00a0us to\npin load increases directly to the affected instance with a single\nspecific partition key, which would not be the case had we assigned a\nshared instance to handle multiple keys.",
+ "title": "Default Dynamic Partitioning"
+ },
+ {
"location": "/application_development/#nxm-partitions",
"text": "When two consecutive logical operators are partitioned a special\noptimization is done. Technically the output of the first operator\nshould be unified and streamed to the next logical node. But that can\ncreate a network bottleneck. The platform optimizes this by partitioning\nthe output stream of each partition of the first operator as per the\npartitions needed by the next operator. For example if the first\noperator has N partitions and the second operator has M partitions then\neach of the N partitions would send out M streams. The first of each of\nthese M streams would be unified and routed to the first of the M\npartitions, and so on. Such an optimization allows for higher\nscalability and eliminates a network bottleneck (one unifier in between\nthe two operators) by having M unifiers. This also enables the\napplication to perform within the resource limits enforced by YARN.\nSTRAM has a much better understanding and estimation of unifier resource\nneeds and is thus able to optimize for resource constraints. Figure 5 shows a case where we have a 3x2 partition; the single\nintermediate unifier between operator 1\u00a0and 2\u00a0is\noptimized away. The partition computation for operator 2\u00a0is executed on outbound streams of each\npartitions of operator 1. Each\npartition of operator 2\u00a0has its own\nCONTAINER_LOCAL unifier. In such a situation, the in-bound network\ntuple flow is split between containers for 2a\u00a0and 2b\u00a0each of which take half the traffic. STRAM\ndoes this by default since it always has better performance.",
"title": "NxM Partitions"
@@ -202,10 +392,30 @@
},
{
"location": "/application_development/#recovery-mechanisms_1",
- "text": "Recovery mechanism are ways to recover from a container (or an\noperator) outage. In this section we discuss a single container outage.\nMultiple container outages are handled as independent events. Recovery\nrequires the upstream buffer server to replay windows and it would\nsimply go one more level upstream if the immediate upstream container\nhas also failed. If multiple operators are in a container (THREAD_LOCAL\nor CONTAINER_LOCAL stream) the container recovery treats each operator\nas an independent object when figuring out the recovery steps.\nApplication developers can set any of the recovery mechanisms discussed\nbelow for node outage. In general, the cost of recovery depends on the state of the\noperator and the recovery mechanism selected, while data loss tolerance\nis specified by the application. For example a data-loss tolerant\napplication would prefer at most\nonce\u00a0recovery. All recovery mechanisms treat a streaming\nwindow as an atomic computation unit. In all three recovery mechanisms\nthe new operator connects to the upstream buffer server and asks for\ndata from a particular window onwards. Thus all recovery methods\ntranslate to deciding which atomic units to re-compute and which state\nthe new operator resumes from. A partially computed micro-batch is\nalways dropped. Such micro-batches are re-computed in at-least-once or\nexactly-once mode and skipped in at-most-once mode. The notiion of an\natomic micro-batch is a critical guiding principle as it enables very\nlow bookkeeping costs, high throughput, low recovery times, and high\nscalability. Within an application each operator can have its own\nrecovery mechanism. At Least Once At least once recovery is the default recovery mechanism, i.e it\nis used when no mechanism is specified. In this method, the lost\noperator is brought back to its latest viable checkpointed state and the\nupstream buffer server is asked to replay all subsequent windows. There\nis no data loss in recovery. The viable checkpoint state is defined as\nthe one whose window id is in the past as compared to all the\ncheckpoints of all the downstream operators. All downstream operators\nare restarted at their checkpointed state. They ignore all incoming data\nthat belongs to windows prior their checkpointed window. The lost\nwindows are thus recomputed and the application catches up with live\nincoming data. This is called \" at least\nonce\"\u00a0because lost windows are recomputed. For example if\nthe streaming window is 0.5 seconds and checkpointing is being done\nevery 30 seconds, then upon node outage all windows since the last\ncheckpoint (up to 60 windows) need to be re-processed. If the\napplication can handle loss of data, then this is not the most optimal\nrecovery mechanism. In general for this recovery mode, the average time lag on a node\noutage is = (CP/2*SW)*T + HC where CP \u00a0\u00a0- Checkpointing period (default value is 30 seconds) SW \u00a0\u00a0- Streaming window period (default value is 0.5 seconds) T \u00a0\u00a0\u00a0- \u00a0Time taken to re-compute one lost window from data in memory HC \u00a0\u00a0- Time it takes to get a new Hadoop Container, or make do with the current ones A lower CP is a trade off between cost of checkpointing and the\nneed to have to use it in case of outage. Input adapters cannot use\nat-least-once recovery without the support from sources outside Hadoop.\nFor an output adapter care may needed if the external system cannot\nhandle re-write of the same data. At Most Once This recovery mechanism is for applications that can tolerate\ndata-loss; they get the quickest recovery in return. The restarted node\nconnects to the upstream buffer server, subscribing to data from the\nstart of the next window. It then starts processing that window. The\ndownstream operators ignore the lost windows and continue to process\nincoming data normally. Thus, this mechanism forces all downstream\noperators to follow. For multiple inputs, the operator waits for all ports with the\nat-most-once attribute to get responses from their respective buffer\nservers. Then, the operator starts processing till the end window of the\nlatest window id on each input port is reached. In this case the end\nwindow tuple is non-blocking till the common window id is reached. At\nthis point the input ports are now properly synchronized. Upstream nodes\nreconnect under at most\nonce\u00a0paradigm in same way. \u00a0For example, assume an operator\nhas ports in1\u00a0and in2\u00a0and a checkpointed window of 95. Assume further that the buffer servers of\noperators upstream of in1\u00a0and\nin2\u00a0respond with window id 100 and\n102 respectively. Then port in1\u00a0would continue to process till end window of\n101, while port in2\u00a0will wait for in1\nto catch up to 102.\nFrom \u00a0then on, both ports process their tuples normally. So windows from\n96 to 99are lost. Window 100\nand 101 has only\nin1 active, and 102 onwards both ports are active. The other\nports of upstream nodes would also catch up till 102in a similar fashion. This operator may not\nneed to be checkpointed. Currently the option to not do checkpoint in\nsuch cases is not available. In general, in this recovery mode, the average time lag on a node\noutage is = SW/2 + HC where SW \u00a0- Streaming window period (default value is 0.5\nseconds) HC \u00a0- Time it takes to get a new Hadoop Container, or make\ndo with the current ones Exactly Once This recovery mechanism is for applications that require no\ndata-loss as well are no recomputation. Since a window is an atomic\ncompute unit, exactly once applies to the window as a whole. In this\nrecovery mode, the operator is brought back to the start of the window\nin which the outage happened and the window is recomputed. The window is\nconsidered closed when all the data computations are done and end window\ntuple is emitted. \u00a0Exactly once requires every window to be\ncheckpointed. From then on, the operator asks the upstream buffer server\nto send data from the last checkpoint. The upstream node behaves the\nsame as in at-most-once recovery. Checkpointing after every streaming\nwindow is very costly, but users would most often do exactly once per\napplication window; if the application window size is substantially\nlarger than the streaming window size (which typically is the case) the\ncost of running an operator in this recovery mode may not be as\nhigh. Speculative Execution In future we looking at possibility of adding speculative execution for the applications. This would be enabled in multiple ways. At an operator level: The upstream operator would emit to\n two copies. The downstream operator would receive from both copies\n and pick a winner. The winner (primary) would be picked in either of\n the following ways Statically as dictated by STRAM Dynamically based on whose tuple arrives first. This mode\n needs both copies to guarantee that the computation result would\n have identical functionality At a sub-query level: A part of the application DAG would be\n run in parallel and all upstream operators would feed to two copies\n and all downstream operators would receive from both copies. The\n winners would again be picked in a static or dynamic manner Entire DAG: Another copy of the application would be run by\n STRAM and the winner would be decided outside the application. In\n this mode the output adapters would both be writing\n the result. In all cases the two copies would run on different Hadoop nodes.\nSpeculative execution is under development and\nis not yet available.",
+ "text": "Recovery mechanism are ways to recover from a container (or an\noperator) outage. In this section we discuss a single container outage.\nMultiple container outages are handled as independent events. Recovery\nrequires the upstream buffer server to replay windows and it would\nsimply go one more level upstream if the immediate upstream container\nhas also failed. If multiple operators are in a container (THREAD_LOCAL\nor CONTAINER_LOCAL stream) the container recovery treats each operator\nas an independent object when figuring out the recovery steps.\nApplication developers can set any of the recovery mechanisms discussed\nbelow for node outage. In general, the cost of recovery depends on the state of the\noperator and the recovery mechanism selected, while data loss tolerance\nis specified by the application. For example a data-loss tolerant\napplication would prefer at most\nonce\u00a0recovery. All recovery mechanisms treat a streaming\nwindow as an atomic computation unit. In all three recovery mechanisms\nthe new operator connects to the upstream buffer server and asks for\ndata from a particular window onwards. Thus all recovery methods\ntranslate to deciding which atomic units to re-compute and which state\nthe new operator resumes from. A partially computed micro-batch is\nalways dropped. Such micro-batches are re-computed in at-least-once or\nexactly-once mode and skipped in at-most-once mode. The notiion of an\natomic micro-batch is a critical guiding principle as it enables very\nlow bookkeeping costs, high throughput, low recovery times, and high\nscalability. Within an application each operator can have its own\nrecovery mechanism.",
"title": "Recovery Mechanisms"
},
{
+ "location": "/application_development/#at-least-once",
+ "text": "At least once recovery is the default recovery mechanism, i.e it\nis used when no mechanism is specified. In this method, the lost\noperator is brought back to its latest viable checkpointed state and the\nupstream buffer server is asked to replay all subsequent windows. There\nis no data loss in recovery. The viable checkpoint state is defined as\nthe one whose window id is in the past as compared to all the\ncheckpoints of all the downstream operators. All downstream operators\nare restarted at their checkpointed state. They ignore all incoming data\nthat belongs to windows prior their checkpointed window. The lost\nwindows are thus recomputed and the application catches up with live\nincoming data. This is called \" at least\nonce\"\u00a0because lost windows are recomputed. For example if\nthe streaming window is 0.5 seconds and checkpointing is being done\nevery 30 seconds, then upon node outage all windows since the last\ncheckpoint (up to 60 windows) need to be re-processed. If the\napplication can handle loss of data, then this is not the most optimal\nrecovery mechanism. In general for this recovery mode, the average time lag on a node\noutage is = (CP/2*SW)*T + HC where CP \u00a0\u00a0- Checkpointing period (default value is 30 seconds) SW \u00a0\u00a0- Streaming window period (default value is 0.5 seconds) T \u00a0\u00a0\u00a0- \u00a0Time taken to re-compute one lost window from data in memory HC \u00a0\u00a0- Time it takes to get a new Hadoop Container, or make do with the current ones A lower CP is a trade off between cost of checkpointing and the\nneed to have to use it in case of outage. Input adapters cannot use\nat-least-once recovery without the support from sources outside Hadoop.\nFor an output adapter care may needed if the external system cannot\nhandle re-write of the same data.",
+ "title": "At Least Once"
+ },
+ {
+ "location": "/application_development/#at-most-once",
+ "text": "This recovery mechanism is for applications that can tolerate\ndata-loss; they get the quickest recovery in return. The restarted node\nconnects to the upstream buffer server, subscribing to data from the\nstart of the next window. It then starts processing that window. The\ndownstream operators ignore the lost windows and continue to process\nincoming data normally. Thus, this mechanism forces all downstream\noperators to follow. For multiple inputs, the operator waits for all ports with the\nat-most-once attribute to get responses from their respective buffer\nservers. Then, the operator starts processing till the end window of the\nlatest window id on each input port is reached. In this case the end\nwindow tuple is non-blocking till the common window id is reached. At\nthis point the input ports are now properly synchronized. Upstream nodes\nreconnect under at most\nonce\u00a0paradigm in same way. \u00a0For example, assume an operator\nhas ports in1\u00a0and in2\u00a0and a checkpointed window of 95. Assume further that the buffer servers of\noperators upstream of in1\u00a0and\nin2\u00a0respond with window id 100 and\n102 respectively. Then port in1\u00a0would continue to process till end window of\n101, while port in2\u00a0will wait for in1\nto catch up to 102.\nFrom \u00a0then on, both ports process their tuples normally. So windows from\n96 to 99are lost. Window 100\nand 101 has only\nin1 active, and 102 onwards both ports are active. The other\nports of upstream nodes would also catch up till 102in a similar fashion. This operator may not\nneed to be checkpointed. Currently the option to not do checkpoint in\nsuch cases is not available. In general, in this recovery mode, the average time lag on a node\noutage is = SW/2 + HC where SW \u00a0- Streaming window period (default value is 0.5\nseconds) HC \u00a0- Time it takes to get a new Hadoop Container, or make\ndo with the current ones",
+ "title": "At Most Once"
+ },
+ {
+ "location": "/application_development/#exactly-once",
+ "text": "This recovery mechanism is for applications that require no\ndata-loss as well are no recomputation. Since a window is an atomic\ncompute unit, exactly once applies to the window as a whole. In this\nrecovery mode, the operator is brought back to the start of the window\nin which the outage happened and the window is recomputed. The window is\nconsidered closed when all the data computations are done and end window\ntuple is emitted. \u00a0Exactly once requires every window to be\ncheckpointed. From then on, the operator asks the upstream buffer server\nto send data from the last checkpoint. The upstream node behaves the\nsame as in at-most-once recovery. Checkpointing after every streaming\nwindow is very costly, but users would most often do exactly once per\napplication window; if the application window size is substantially\nlarger than the streaming window size (which typically is the case) the\ncost of running an operator in this recovery mode may not be as\nhigh.",
+ "title": "Exactly Once"
+ },
+ {
+ "location": "/application_development/#speculative-execution",
+ "text": "In future we looking at possibility of adding speculative execution for the applications. This would be enabled in multiple ways. At an operator level: The upstream operator would emit to\n two copies. The downstream operator would receive from both copies\n and pick a winner. The winner (primary) would be picked in either of\n the following ways Statically as dictated by STRAM Dynamically based on whose tuple arrives first. This mode\n needs both copies to guarantee that the computation result would\n have identical functionality At a sub-query level: A part of the application DAG would be\n run in parallel and all upstream operators would feed to two copies\n and all downstream operators would receive from both copies. The\n winners would again be picked in a static or dynamic manner Entire DAG: Another copy of the application would be run by\n STRAM and the winner would be decided outside the application. In\n this mode the output adapters would both be writing\n the result. In all cases the two copies would run on different Hadoop nodes.\nSpeculative execution is under development and\nis not yet available.",
+ "title": "Speculative Execution"
+ },
+ {
"location": "/application_development/#dynamic-application-modifications",
"text": "Dynamic application modifications are being worked on and most of\nthe features discussed here are now available. The platform supports the\nability to modify the DAG of the application as per inputs as well as\nset constraints, and will continue to provide abilities to deepen\nfeatures based on this ability. All these changes have one thing in\ncommon and that is the application does not need to be restarted as\nSTRAM will instrument the changes and the streaming will catch-up and\ncontinue. Some examples are Dynamic Partitioning:\u00a0Automatic\n changes in partitioning of computations to match constraints on a\n run time basis. Examples includes STRAM adding resource during spike\n in streams and returning them once spike is gone. Scale up and scale\n down is done automatically without human intervention. Modification via constraints: Attributes can be changed via\n Webservices and STRAM would adapt the execution plan to meet these.\n Examples include operations folks asking STRAM to reduce container\n count, or changing network resource restrictions. Modification via properties: Properties of operators can be\n changed in run time. This enables application developers to trigger\n a new behavior as need be. Examples include triggering an alert ON.\n The platform supports changes to any property of an operator that\n has a setter function defined. Modification of DAG structure: Operators and streams can be\n added to or removed from a running DAG, provided the code of the\n operator being added is already in the classpath of the running\n application master. \u00a0This enables application developers to add or\n remove processing pipelines on the fly without having to restart\n the application. Query Insertion: Addition of sub-queries to currently\n running application. This query would take current streams as inputs\n and start computations as per their specs. Examples insertion of\n SQL-queries on live data streams, dynamic query submission and\n result from STRAM (not yet available). Dynamic modifications to applications are foundational part of the\nplatform. They enable users to build layers over the applications. Users\ncan also save all the changes done since the application launch, and\ntherefore predictably get the application to its current state. For\ndetails refer to Configuration Guide \n.",
"title": "Dynamic Application Modifications"
@@ -237,25 +447,120 @@
},
{
"location": "/application_packages/#creating-your-first-apex-app-package",
- "text": "You can create an Apex Application Package using your Linux command\nline, or using your favorite IDE. Using Command Line First, change to the directory where you put your projects, and create\nan Apex application project using Maven by running the following\ncommand. Replace \"com.example\", \"mydtapp\" and \"1.0-SNAPSHOT\" with the\nappropriate values (make sure this is all on one line): $ mvn archetype:generate \\\n -DarchetypeGroupId=org.apache.apex \\\n -DarchetypeArtifactId=apex-app-archetype -DarchetypeVersion=3.2.0-incubating \\\n -DgroupId=com.example -Dpackage=com.example.mydtapp -DartifactId=mydtapp \\\n -Dversion=1.0-SNAPSHOT This creates a Maven project named \"mydtapp\". Open it with your favorite\nIDE (e.g. NetBeans, Eclipse, IntelliJ IDEA). In the project, there is a\nsample DAG that generates a number of tuples with a random number and\nprints out \"hello world\" and the random number in the tuples. The code\nthat builds the DAG is in\nsrc/main/java/com/example/mydtapp/Application.java, and the code that\nruns the unit test for the DAG is in\nsrc/test/java/com/example/mydtapp/ApplicationTest.java. Try it out by\nrunning the following command: $cd mydtapp; mvn package This builds the App Package runs the unit test of the DAG. You should\nbe getting test output similar to this: -------------------------------------------------------\n TESTS\n -------------------------------------------------------\n\n Running com.example.mydtapp.ApplicationTest\n hello world: 0.8015370953286478\n hello world: 0.9785359225545481\n hello world: 0.6322611586644047\n hello world: 0.8460953663451775\n hello world: 0.5719372906929072\n hello world: 0.6361174312337172\n hello world: 0.14873007534816318\n hello world: 0.8866986277418261\n hello world: 0.6346526809866057\n hello world: 0.48587295703904465\n hello world: 0.6436832429676687\n\n ...\n\n Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.863\n sec\n\n Results :\n\n Tests run: 1, Failures: 0, Errors: 0, Skipped: 0 The \"mvn package\" command creates the App Package file in target\ndirectory as target/mydtapp-1.0-SNAPSHOT.apa. You will be able to use\nthat App Package file to launch this sample application in your actual\nApex installation. Alternatively you can perform the same steps within your IDE (IDEA IntelliJ, Eclipse, NetBeans all support it). Please check the IDE documentation for details. Group ID: org.apache.apex\nArtifact ID: apex-app-archetype\nVersion: 3.2.0-incubating (or any later version)",
+ "text": "You can create an Apex Application Package using your Linux command\nline, or using your favorite IDE.",
"title": "Creating Your First Apex App Package"
},
{
+ "location": "/application_packages/#using-command-line",
+ "text": "First, change to the directory where you put your projects, and create\nan Apex application project using Maven by running the following\ncommand. Replace \"com.example\", \"mydtapp\" and \"1.0-SNAPSHOT\" with the\nappropriate values (make sure this is all on one line): $ mvn archetype:generate \\\n -DarchetypeGroupId=org.apache.apex \\\n -DarchetypeArtifactId=apex-app-archetype -DarchetypeVersion=3.2.0-incubating \\\n -DgroupId=com.example -Dpackage=com.example.mydtapp -DartifactId=mydtapp \\\n -Dversion=1.0-SNAPSHOT This creates a Maven project named \"mydtapp\". Open it with your favorite\nIDE (e.g. NetBeans, Eclipse, IntelliJ IDEA). In the project, there is a\nsample DAG that generates a number of tuples with a random number and\nprints out \"hello world\" and the random number in the tuples. The code\nthat builds the DAG is in\nsrc/main/java/com/example/mydtapp/Application.java, and the code that\nruns the unit test for the DAG is in\nsrc/test/java/com/example/mydtapp/ApplicationTest.java. Try it out by\nrunning the following command: $cd mydtapp; mvn package This builds the App Package runs the unit test of the DAG. You should\nbe getting test output similar to this: -------------------------------------------------------\n TESTS\n -------------------------------------------------------\n\n Running com.example.mydtapp.ApplicationTest\n hello world: 0.8015370953286478\n hello world: 0.9785359225545481\n hello world: 0.6322611586644047\n hello world: 0.8460953663451775\n hello world: 0.5719372906929072\n hello world: 0.6361174312337172\n hello world: 0.14873007534816318\n hello world: 0.8866986277418261\n hello world: 0.6346526809866057\n hello world: 0.48587295703904465\n hello world: 0.6436832429676687\n\n ...\n\n Tests run: 1, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 11.863\n sec\n\n Results :\n\n Tests run: 1, Failures: 0, Errors: 0, Skipped: 0 The \"mvn package\" command creates the App Package file in target\ndirectory as target/mydtapp-1.0-SNAPSHOT.apa. You will be able to use\nthat App Package file to launch this sample application in your actual\nApex installation. Alternatively you can perform the same steps within your IDE (IDEA IntelliJ, Eclipse, NetBeans all support it). Please check the IDE documentation for details. Group ID: org.apache.apex\nArtifact ID: apex-app-archetype\nVersion: 3.2.0-incubating (or any later version)",
+ "title": "Using Command Line"
+ },
+ {
"location": "/application_packages/#writing-your-own-app-package",
- "text": "Please refer to the Creating Apps on the basics on how to write an Apache Apex application. In your AppPackage project, you can add custom operators (refer to Operator Development Guide , project dependencies, default and required configuration properties, pre-set configurations and other metadata. Adding (and removing) project dependencies Under the project, you can add project dependencies in pom.xml, or do it\nthrough your IDE. Here\u2019s the section that describes the dependencies in\nthe default pom.xml: dependencies \n !-- add your dependencies here -- \n dependency \n groupId org.apache.apex /groupId \n artifactId malhar-library /artifactId \n version ${apex.version} /version \n !--\n If you know your application do not need the transitive dependencies that are pulled in by malhar-library,\n Uncomment the following to reduce the size of your app package.\n -- \n !--\n exclusions \n exclusion \n groupId * /groupId \n artifactId * /artifactId \n /exclusion \n /exclusions \n -- \n /dependency \n dependency \n groupId org.apache.apex /groupId \n artifactId apex-engine /artifactId \n version ${apex.version} /version \n scope provided /scope \n /dependency \n dependency \n groupId junit /groupId \n artifactId junit /artifactId \n version 4.10 /version \n scope test /scope \n /dependency \n /dependencies By default, as shown above, the default dependencies include\nmalhar-library in compile scope, dt-engine in provided scope, and junit\nin test scope. Do not remove these three dependencies since they are\nnecessary for any Apex application. You can, however, exclude\ntransitive dependencies from malhar-library to reduce the size of your\nApp Package, provided that none of the operators in malhar-library that\nneed the transitive dependencies will be used in your application. In the sample application, it is safe to remove the transitive\ndependencies from malhar-library, by uncommenting the \"exclusions\"\nsection. It will reduce the size of the sample App Package from 8MB to\n700KB. Note that if we exclude *, in some versions of Maven, you may get\nwarnings similar to the following: \n [WARNING] 'dependencies.dependency.exclusions.exclusion.groupId' for\n org.apache.apex:malhar-library:jar with value '*' does not match a\n valid id pattern.\n\n [WARNING]\n [WARNING] It is highly recommended to fix these problems because they\n threaten the stability of your build.\n [WARNING]\n [WARNING] For this reason, future Maven versions might no longer support\n building such malformed projects.\n [WARNING] This is a bug in early versions of Maven 3. The dependency exclusion is\nstill valid and it is safe to ignore these warnings. Application Configuration A configuration file can be used to configure an application. Different\nkinds of configuration parameters can be specified. They are application\nattributes, operator attributes and properties, port attributes, stream\nproperties and application specific properties. They are all specified\nas name value pairs, in XML format, like the following. ?xml version= 1.0 ? configuration \n property \n name some_name_1 /name \n value some_default_value /value \n /property \n property \n name some_name_2 /name \n value some_default_value /value \n /property /configuration Application attributes Application attributes are used to specify the platform behavior for the\napplication. They can be specified using the parameter dt.attr. attribute . The prefix \u201cdt\u201d is a constant, \u201cattr\u201d is a\nconstant denoting an attribute is being specified and attribute \nspecifies the name of the attribute. Below is an example snippet setting\nthe streaming windows size of the application to be 1000 milliseconds. property \n name dt.attr.STREAMING_WINDOW_SIZE_MILLIS /name \n value 1000 /value \n /property The name tag specifies the attribute and value tag specifies the\nattribute value. The name of the attribute is a JAVA constant name\nidentifying the attribute. The constants are defined in\ncom.datatorrent.api.Context.DAGContext and the different attributes can\nbe specified in the format described above. Operator attributes Operator attributes are used to specify the platform behavior for the\noperator. They can be specified using the parameter dt.operator. operator-name .attr. attribute . The prefix \u201cdt\u201d is a\nconstant, \u201coperator\u201d is a constant denoting that an operator is being\nspecified, operator-name denotes the name of the operator, \u201cattr\u201d is\nthe constant denoting that an attribute is being specified and attribute is the name of the attribute. The operator name is the\nsame name that is specified when the operator is added to the DAG using\nthe addOperator method. An example illustrating the specification is\nshown below. It specifies the number of streaming windows for one\napplication window of an operator named \u201cinput\u201d to be 10 property \n name dt.operator.input.attr.APPLICATION_WINDOW_COUNT /name \n value 10 /value /property The name tag specifies the attribute and value tag specifies the\nattribute value. The name of the attribute is a JAVA constant name\nidentifying the attribute. The constants are defined in\ncom.datatorrent.api.Context.OperatorContext and the different attributes\ncan be specified in the format described above. Operator properties Operators can be configured using operator specific properties. The\nproperties can be specified using the parameter dt.operator. operator-name .prop. property-name . The difference\nbetween this and the operator attribute specification described above is\nthat the keyword \u201cprop\u201d is used to denote that it is a property and property-name specifies the property name. An example illustrating\nthis is specified below. It specifies the property \u201chostname\u201d of the\nredis server for a \u201credis\u201d output operator. property \n name dt.operator.redis.prop.host /name \n value 127.0.0.1 /value \n /property The name tag specifies the property and the value specifies the property\nvalue. The property name is converted to a setter method which is called\non the actual operator. The method name is composed by appending the\nword \u201cset\u201d and the property name with the first character of the name\ncapitalized. In the above example the setter method would become\nsetHost. The method is called using JAVA reflection and the property\nvalue is passed as an argument. In the above example the method setHost\nwill be called on the \u201credis\u201d operator with \u201c127.0.0.1\u201d as the argument. Port attributes Port attributes are used to specify the platform behavior for input and\noutput ports. They can be specified using the parameter dt.operator. operator-name .inputport. port-name .attr. attribute \nfor input port and dt.operator. operator-name .outputport. port-name .attr. attribute \nfor output port. The keyword \u201cinputport\u201d is used to denote an input port\nand \u201coutputport\u201d to denote an output port. The rest of the specification\nfollows the conventions described in other specifications above. An\nexample illustrating this is specified below. It specifies the queue\ncapacity for an input port named \u201cinput\u201d of an operator named \u201crange\u201d to\nbe 4k. property \n name dt.operator.range.inputport.input.attr.QUEUE_CAPACITY /name \n value 4000 /value /property The name tag specifies the attribute and value tag specifies the\nattribute value. The name of the attribute is a JAVA constant name\nidentifying the attribute. The constants are defined in\ncom.datatorrent.api.Context.PortContext and the different attributes can\nbe specified in the format described above. The attributes for an output port can also be specified in a similar way\nas described above with a change that keyword \u201coutputport\u201d is used\ninstead of \u201cintputport\u201d. A generic keyword \u201cport\u201d can be used to specify\neither an input or an output port. It is useful in the wildcard\nspecification described below. Stream properties Streams can be configured using stream properties. The properties can be\nspecified using the parameter dt.stream. stream-name .prop. property-name The constant \u201cstream\u201d\nspecifies that it is a stream, stream-name specifies the name of the\nstream and property-name the name of the property. The name of the\nstream is the same name that is passed when the stream is added to the\nDAG using the addStream method. An example illustrating the\nspecification is shown below. It sets the locality of the stream named\n\u201cstream1\u201d to container local indicating that the operators the stream is\nconnecting be run in the same container. property \n name dt.stream.stream1.prop.locality /name \n value CONTAINER_LOCAL /value \n /property The property name is converted into a set method on the stream in the\nsame way as described in operator properties section above. In this case\nthe method would be setLocality and it will be called in the stream\n\u201cstream1\u201d with the value as the argument. Along with the above system defined parameters, the applications can\ndefine their own specific parameters they can be specified in the\nconfiguration file. The only condition is that the names of these\nparameters don\u2019t conflict with the system defined parameters or similar\napplication parameters defined by other applications. To this end, it is\nrecommended that the application parameters have the format full-application-class-name . param-name . The\nfull-application-class-name is the full JAVA class name of the\napplication including the package path and param-name is the name of the\nparameter within the application. The application will still have to\nstill read the parameter in using the configuration API of the\nconfiguration object that is passed in populateDAG. Wildcards Wildcards and regular expressions can be used in place of names to\nspecify a group for applications, operators, ports or streams. For\nexample, to specify an attribute for all ports of an operator it can be\ndone as follows property \n name dt.operator.range.port.*.attr.QUEUE_CAPACITY /name \n value 4000 /value /property The wildcard \u201c*\u201d was used instead of the name of the port. Wildcard can\nalso be used for operator name, stream name or application name. Regular\nexpressions can also be used for names to specify attributes or\nproperties for a specific set. Adding configuration properties It is common for applications to require configuration parameters to\nrun. For example, the address and port of the database, the location of\na file for ingestion, etc. You can specify them in\nsrc/main/resources/META-INF/properties.xml under the App Package\nproject. The properties.xml may look like: ?xml version= 1.0 ? configuration \n property \n name some_name_1 /name \n /property \n property \n name some_name_2 /name \n value some_default_value /value \n /property /configuration The name of an application-specific property takes the form of: dt.operator.{opName}.prop.{propName} The first represents the property with name propName of operator opName.\n Or you can set the application name at run time by setting this\nproperty: dt.attr.APPLICATION_NAME In this example, property some_name_1 is a required property which\nmust be set at launch time, or it must be set by a pre-set configuration\n(see next section). Property some_name_2 is a property that is\nassigned with value some_default_value unless it is overridden at\nlaunch time. Adding pre-set configurations At build time, you can add pre-set configurations to the App Package by\nadding configuration XML files under src/site/conf/ conf .xml in your\nproject. You can then specify which configuration to use at launch\ntime. The configuration XML is of the same format of the properties.xml\nfile. Application-specific properties file You can also specify properties.xml per application in the application\npackage. Just create a file with the name properties-{appName}.xml and\nit will be picked up when you launch the application with the specified\nname within the application package. In short: properties.xml: Properties that are global to the Configuration\nPackage properties-{appName}.xml: Properties that are specific when launching\nan application with the specified appName. Properties source precedence If properties with the same key appear in multiple sources (e.g. from\napp package default configuration as META-INF/properties.xml, from app\npackage configuration in the conf directory, from launch time defines,\netc), the precedence of sources, from highest to lowest, is as follows: Launch time defines (using -D option in CLI) Launch time specified configuration file in file system (using -conf\n option in CLI) Launch time specified package configuration (using -apconf option in\n CLI) Configuration from \\$HOME/.dt/dt-site.xml Application defaults within the package as\n META-INF/properties-{appname}.xml Package defaults as META-INF/properties.xml dt-site.xml in local DT installation dt-site.xml stored in HDFS Other meta-data In a Apex App Package project, the pom.xml file contains a\nsection that looks like: properties \n apex.version 3.2.0-incubating /apex.version \n apex.apppackage.classpath\\ lib*.jar /apex.apppackage.classpath /properties apex.version is the Apache Apex version that are to be used\nwith this Application Package. apex.apppackage.classpath is the classpath that is used when\nlaunching the application in the Application Package. The default is\nlib/*.jar, where lib is where all the dependency jars are kept within\nthe Application Package. One reason to change this field is when your\nApplication Package needs the classpath in a specific order. Logging configuration Just like other Java projects, you can change the logging configuration\nby having your log4j.properties under src/main/resources. For example,\nif you have the following in src/main/resources/log4j.properties: log4j.rootLogger=WARN,CONSOLE\n log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender\n log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout\n log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [%t] %-5p\n %c{2} %M - %m%n The root logger\u2019s level is set to WARN and the output is set to the console (stdout). Note that by default from project created from the maven archetype,\nthere is already a log4j.properties file under src/test/resources and\nthat file is only used for the unit test.",
+ "text": "Please refer to the Creating Apps on the basics on how to write an Apache Apex application. In your AppPackage project, you can add custom operators (refer to Operator Development Guide , project dependencies, default and required configuration properties, pre-set configurations and other metadata.",
"title": "Writing Your Own App Package"
},
{
+ "location": "/application_packages/#adding-and-removing-project-dependencies",
+ "text": "Under the project, you can add project dependencies in pom.xml, or do it\nthrough your IDE. Here\u2019s the section that describes the dependencies in\nthe default pom.xml: dependencies \n !-- add your dependencies here -- \n dependency \n groupId org.apache.apex /groupId \n artifactId malhar-library /artifactId \n version ${apex.version} /version \n !--\n If you know your application do not need the transitive dependencies that are pulled in by malhar-library,\n Uncomment the following to reduce the size of your app package.\n -- \n !--\n exclusions \n exclusion \n groupId * /groupId \n artifactId * /artifactId \n /exclusion \n /exclusions \n -- \n /dependency \n dependency \n groupId org.apache.apex /groupId \n artifactId apex-engine /artifactId \n version ${apex.version} /version \n scope provided /scope \n /dependency \n dependency \n groupId junit /groupId \n artifactId junit /artifactId \n version 4.10 /version \n scope test /scope \n /dependency \n /dependencies By default, as shown above, the default dependencies include\nmalhar-library in compile scope, dt-engine in provided scope, and junit\nin test scope. Do not remove these three dependencies since they are\nnecessary for any Apex application. You can, however, exclude\ntransitive dependencies from malhar-library to reduce the size of your\nApp Package, provided that none of the operators in malhar-library that\nneed the transitive dependencies will be used in your application. In the sample application, it is safe to remove the transitive\ndependencies from malhar-library, by uncommenting the \"exclusions\"\nsection. It will reduce the size of the sample App Package from 8MB to\n700KB. Note that if we exclude *, in some versions of Maven, you may get\nwarnings similar to the following: \n [WARNING] 'dependencies.dependency.exclusions.exclusion.groupId' for\n org.apache.apex:malhar-library:jar with value '*' does not match a\n valid id pattern.\n\n [WARNING]\n [WARNING] It is highly recommended to fix these problems because they\n threaten the stability of your build.\n [WARNING]\n [WARNING] For this reason, future Maven versions might no longer support\n building such malformed projects.\n [WARNING] This is a bug in early versions of Maven 3. The dependency exclusion is\nstill valid and it is safe to ignore these warnings.",
+ "title": "Adding (and removing) project dependencies"
+ },
+ {
+ "location": "/application_packages/#application-configuration",
+ "text": "A configuration file can be used to configure an application. Different\nkinds of configuration parameters can be specified. They are application\nattributes, operator attributes and properties, port attributes, stream\nproperties and application specific properties. They are all specified\nas name value pairs, in XML format, like the following. ?xml version= 1.0 ? configuration \n property \n name some_name_1 /name \n value some_default_value /value \n /property \n property \n name some_name_2 /name \n value some_default_value /value \n /property /configuration",
+ "title": "Application Configuration"
+ },
+ {
+ "location": "/application_packages/#application-attributes",
+ "text": "Application attributes are used to specify the platform behavior for the\napplication. They can be specified using the parameter dt.attr. attribute . The prefix \u201cdt\u201d is a constant, \u201cattr\u201d is a\nconstant denoting an attribute is being specified and attribute \nspecifies the name of the attribute. Below is an example snippet setting\nthe streaming windows size of the application to be 1000 milliseconds. property \n name dt.attr.STREAMING_WINDOW_SIZE_MILLIS /name \n value 1000 /value \n /property The name tag specifies the attribute and value tag specifies the\nattribute value. The name of the attribute is a JAVA constant name\nidentifying the attribute. The constants are defined in\ncom.datatorrent.api.Context.DAGContext and the different attributes can\nbe specified in the format described above.",
+ "title": "Application attributes"
+ },
+ {
+ "location": "/application_packages/#operator-attributes",
+ "text": "Operator attributes are used to specify the platform behavior for the\noperator. They can be specified using the parameter dt.operator. operator-name .attr. attribute . The prefix \u201cdt\u201d is a\nconstant, \u201coperator\u201d is a constant denoting that an operator is being\nspecified, operator-name denotes the name of the operator, \u201cattr\u201d is\nthe constant denoting that an attribute is being specified and attribute is the name of the attribute. The operator name is the\nsame name that is specified when the operator is added to the DAG using\nthe addOperator method. An example illustrating the specification is\nshown below. It specifies the number of streaming windows for one\napplication window of an operator named \u201cinput\u201d to be 10 property \n name dt.operator.input.attr.APPLICATION_WINDOW_COUNT /name \n value 10 /value /property The name tag specifies the attribute and value tag specifies the\nattribute value. The name of the attribute is a JAVA constant name\nidentifying the attribute. The constants are defined in\ncom.datatorrent.api.Context.OperatorContext and the different attributes\ncan be specified in the format described above.",
+ "title": "Operator attributes"
+ },
+ {
+ "location": "/application_packages/#operator-properties",
+ "text": "Operators can be configured using operator specific properties. The\nproperties can be specified using the parameter dt.operator. operator-name .prop. property-name . The difference\nbetween this and the operator attribute specification described above is\nthat the keyword \u201cprop\u201d is used to denote that it is a property and property-name specifies the property name. An example illustrating\nthis is specified below. It specifies the property \u201chostname\u201d of the\nredis server for a \u201credis\u201d output operator. property \n name dt.operator.redis.prop.host /name \n value 127.0.0.1 /value \n /property The name tag specifies the property and the value specifies the property\nvalue. The property name is converted to a setter method which is called\non the actual operator. The method name is composed by appending the\nword \u201cset\u201d and the property name with the first character of the name\ncapitalized. In the above example the setter method would become\nsetHost. The method is called using JAVA reflection and the property\nvalue is passed as an argument. In the above example the method setHost\nwill be called on the \u201credis\u201d operator with \u201c127.0.0.1\u201d as the argument.",
+ "title": "Operator properties"
+ },
+ {
+ "location": "/application_packages/#port-attributes",
+ "text": "Port attributes are used to specify the platform behavior for input and\noutput ports. They can be specified using the parameter dt.operator. operator-name .inputport. port-name .attr. attribute \nfor input port and dt.operator. operator-name .outputport. port-name .attr. attribute \nfor output port. The keyword \u201cinputport\u201d is used to denote an input port\nand \u201coutputport\u201d to denote an output port. The rest of the specification\nfollows the conventions described in other specifications above. An\nexample illustrating this is specified below. It specifies the queue\ncapacity for an input port named \u201cinput\u201d of an operator named \u201crange\u201d to\nbe 4k. property \n name dt.operator.range.inputport.input.attr.QUEUE_CAPACITY /name \n value 4000 /value /property The name tag specifies the attribute and value tag specifies the\nattribute value. The name of the attribute is a JAVA constant name\nidentifying the attribute. The constants are defined in\ncom.datatorrent.api.Context.PortContext and the different attributes can\nbe specified in the format described above. The attributes for an output port can also be specified in a similar way\nas described above with a change that keyword \u201coutputport\u201d is used\ninstead of \u201cintputport\u201d. A generic keyword \u201cport\u201d can be used to specify\neither an input or an output port. It is useful in the wildcard\nspecification described below.",
+ "title": "Port attributes"
+ },
+ {
+ "location": "/application_packages/#stream-properties",
+ "text": "Streams can be configured using stream properties. The properties can be\nspecified using the parameter dt.stream. stream-name .prop. property-name The constant \u201cstream\u201d\nspecifies that it is a stream, stream-name specifies the name of the\nstream and property-name the name of the property. The name of the\nstream is the same name that is passed when the stream is added to the\nDAG using the addStream method. An example illustrating the\nspecification is shown below. It sets the locality of the stream named\n\u201cstream1\u201d to container local indicating that the operators the stream is\nconnecting be run in the same container. property \n name dt.stream.stream1.prop.locality /name \n value CONTAINER_LOCAL /value \n /property The property name is converted into a set method on the stream in the\nsame way as described in operator properties section above. In this case\nthe method would be setLocality and it will be called in the stream\n\u201cstream1\u201d with the value as the argument. Along with the above system defined parameters, the applications can\ndefine their own specific parameters they can be specified in the\nconfiguration file. The only condition is that the names of these\nparameters don\u2019t conflict with the system defined parameters or similar\napplication parameters defined by other applications. To this end, it is\nrecommended that the application parameters have the format full-application-class-name . param-name . The\nfull-application-class-name is the full JAVA class name of the\napplication including the package path and param-name is the name of the\nparameter within the application. The application will still have to\nstill read the parameter in using the configuration API of the\nconfiguration object that is passed in populateDAG.",
+ "title": "Stream properties"
+ },
+ {
+ "location": "/application_packages/#wildcards",
+ "text": "Wildcards and regular expressions can be used in place of names to\nspecify a group for applications, operators, ports or streams. For\nexample, to specify an attribute for all ports of an operator it can be\ndone as follows property \n name dt.operator.range.port.*.attr.QUEUE_CAPACITY /name \n value 4000 /value /property The wildcard \u201c*\u201d was used instead of the name of the port. Wildcard can\nalso be used for operator name, stream name or application name. Regular\nexpressions can also be used for names to specify attributes or\nproperties for a specific set.",
+ "title": "Wildcards"
+ },
+ {
+ "location": "/application_packages/#adding-configuration-properties",
+ "text": "It is common for applications to require configuration parameters to\nrun. For example, the address and port of the database, the location of\na file for ingestion, etc. You can specify them in\nsrc/main/resources/META-INF/properties.xml under the App Package\nproject. The properties.xml may look like: ?xml version= 1.0 ? configuration \n property \n name some_name_1 /name \n /property \n property \n name some_name_2 /name \n value some_default_value /value \n /property /configuration The name of an application-specific property takes the form of: dt.operator.{opName}.prop.{propName} The first represents the property with name propName of operator opName.\n Or you can set the application name at run time by setting this\nproperty: dt.attr.APPLICATION_NAME In this example, property some_name_1 is a required property which\nmust be set at launch time, or it must be set by a pre-set configuration\n(see next section). Property some_name_2 is a property that is\nassigned with value some_default_value unless it is overridden at\nlaunch time.",
+ "title": "Adding configuration properties"
+ },
+ {
+ "location": "/application_packages/#adding-pre-set-configurations",
+ "text": "At build time, you can add pre-set configurations to the App Package by\nadding configuration XML files under src/site/conf/ conf .xml in your\nproject. You can then specify which configuration to use at launch\ntime. The configuration XML is of the same format of the properties.xml\nfile.",
+ "title": "Adding pre-set configurations"
+ },
+ {
+ "location": "/application_packages/#application-specific-properties-file",
+ "text": "You can also specify properties.xml per application in the application\npackage. Just create a file with the name properties-{appName}.xml and\nit will be picked up when you launch the application with the specified\nname within the application package. In short: properties.xml: Properties that are global to the Configuration\nPackage properties-{appName}.xml: Properties that are specific when launching\nan application with the specified appName.",
+ "title": "Application-specific properties file"
+ },
+ {
+ "location": "/application_packages/#properties-source-precedence",
+ "text": "If properties with the same key appear in multiple sources (e.g. from\napp package default configuration as META-INF/properties.xml, from app\npackage configuration in the conf directory, from launch time defines,\netc), the precedence of sources, from highest to lowest, is as follows: Launch time defines (using -D option in CLI) Launch time specified configuration file in file system (using -conf\n option in CLI) Launch time specified package configuration (using -apconf option in\n CLI) Configuration from \\$HOME/.dt/dt-site.xml Application defaults within the package as\n META-INF/properties-{appname}.xml Package defaults as META-INF/properties.xml dt-site.xml in local DT installation dt-site.xml stored in HDFS",
+ "title": "Properties source precedence"
+ },
+ {
+ "location": "/application_packages/#other-meta-data",
+ "text": "In a Apex App Package project, the pom.xml file contains a\nsection that looks like: properties \n apex.version 3.2.0-incubating /apex.version \n apex.apppackage.classpath\\ lib*.jar /apex.apppackage.classpath /properties apex.version is the Apache Apex version that are to be used\nwith this Application Package. apex.apppackage.classpath is the classpath that is used when\nlaunching the application in the Application Package. The default is\nlib/*.jar, where lib is where all the dependency jars are kept within\nthe Application Package. One reason to change this field is when your\nApplication Package needs the classpath in a specific order.",
+ "title": "Other meta-data"
+ },
+ {
+ "location": "/application_packages/#logging-configuration",
+ "text": "Just like other Java projects, you can change the logging configuration\nby having your log4j.properties under src/main/resources. For example,\nif you have the following in src/main/resources/log4j.properties: log4j.rootLogger=WARN,CONSOLE\n log4j.appender.CONSOLE=org.apache.log4j.ConsoleAppender\n log4j.appender.CONSOLE.layout=org.apache.log4j.PatternLayout\n log4j.appender.CONSOLE.layout.ConversionPattern=%d{ISO8601} [%t] %-5p\n %c{2} %M - %m%n The root logger\u2019s level is set to WARN and the output is set to the console (stdout). Note that by default from project created from the maven archetype,\nthere is already a log4j.properties file under src/test/resources and\nthat file is only used for the unit test.",
+ "title": "Logging configuration"
+ },
+ {
"location": "/application_packages/#zip-structure-of-application-package",
"text": "Apache Apex Application Package files are zip files. You can examine the content of any Application Package by using unzip -t on your Linux command line. There are four top level directories in an Application Package: \"app\" contains the jar files of the DAG code and any custom operators. \"lib\" contains all dependency jars \"conf\" contains all the pre-set configuration XML files. \"META-INF\" contains the MANIFEST.MF file and the properties.xml file. \u201cresources\u201d contains any other files",
"title": "Zip Structure of Application Package"
},
{
"location": "/application_packages/#examining-and-launching-application-packages-through-cli",
- "text": "If you are working with Application Packages in the local filesystem, you can use the Apex Command Line Interface (dtcli). Getting Application Package Meta Information You can get the meta information about the Application Package using\nthis Apex CLI command. dt get-app-package-info app-package-file Getting Available Operators In Application Package You can get the list of available operators in the Application Package\nusing this command. dt get-app-package-operators app-package-file package-prefix \n [parent-class] Getting Properties of Operators in Application Package You can get the list of properties of any operator in the Application\nPackage using this command. dt get-app-package-operator-properties Launching an Application Package You can launch an application within an Application Package. dt launch [-D property-name=property-value, ...] [-conf config-name]\n [-apconf config-file-within-app-package] app-package-file \n [matching-app-name] Note that -conf expects a configuration file in the file system, while -apconf expects a configuration file within the app package.",
+ "text": "If you are working with Application Packages in the local filesystem, you can use the Apex Command Line Interface (dtcli).",
"title": "Examining and Launching Application Packages Through CLI"
},
{
+ "location": "/application_packages/#getting-application-package-meta-information",
+ "text": "You can get the meta information about the Application Package using\nthis Apex CLI command. dt get-app-package-info app-package-file",
+ "title": "Getting Application Package Meta Information"
+ },
+ {
+ "location": "/application_packages/#getting-available-operators-in-application-package",
+ "text": "You can get the list of available operators in the Application Package\nusing this command. dt get-app-package-operators app-package-file package-prefix \n [parent-class]",
+ "title": "Getting Available Operators In Application Package"
+ },
+ {
+ "location": "/application_packages/#getting-properties-of-operators-in-application-package",
+ "text": "You can get the list of properties of any operator in the Application\nPackage using this command. dt get-app-package-operator-properties",
+ "title": "Getting Properties of Operators in Application Package"
+ },
+ {
+ "location": "/application_packages/#launching-an-application-package",
+ "text": "You can launch an application within an Application Package. dt launch [-D property-name=property-value, ...] [-conf config-name]\n [-apconf config-file-within-app-package] app-package-file \n [matching-app-name] Note that -conf expects a configuration file in the file system, while -apconf expects a configuration file within the app package.",
+ "title": "Launching an Application Package"
+ },
+ {
"location": "/application_packages/#configuration-packages",
"text": "Sometimes just a configuration file is not enough for launching an application package. If a configuration requires\nadditional files to be packaged, you can use an Apex Configuration Package.",
"title": "Configuration Packages"
@@ -267,10 +572,40 @@
},
{
"location": "/application_packages/#assembling-your-own-configuration-package",
- "text": "Inside the project created by the archetype, these are the files that\nyou should know about when assembling your own configuration package: ./pom.xml\n./src/main/resources/classpath\n./src/main/resources/files\n./src/main/resources/META-INF/properties.xml\n./src/main/resources/META-INF/properties-{appname}.xml pom.xml Example: groupId com.example /groupId \n version 1.0.0 /version \n artifactId mydtconf /artifactId \n packaging jar /packaging \n !-- change these to the appropriate values -- \n name My Apex Application Configuration /name \n description My Custom Application Configuration Description /description \n properties \n apex.apppackage.name myapexapp /apex.apppackage.name \n apex.apppackage.minversion 1.0.0 /apex.apppackage.minversion \n apex.apppackage.maxversion 1.9999.9999 /apex.apppackage.maxversion \n apex.appconf.classpath classpath/* /apex.appconf.classpath \n apex.appconf.files files/* /apex.appconf.files \n /properties In pom.xml, you can change the following keys to your desired values groupId version artifactId name description You can also change the values of apex.apppackage.name apex.apppackage.minversion apex.apppackage.maxversion to reflect what Application Packages can be used with this configuration package. Apex will use this information to check whether a\nconfiguration package is compatible with the Application Package when you issue a launch command. ./src/main/resources/classpath Place any file in this directory that you\u2019d like to be copied to the\ncompute machines when launching an application and included in the\nclasspath of the application. Example of such files are Java properties\nfiles and jar files. ./src/main/resources/files Place any file in this directory that you\u2019d like to be copied to the\ncompute machines when launching an application but not included in the\nclasspath of the application. Properties XML file A properties xml file consists of a set of key-value pairs. The set of\nkey-value pairs specifies the configuration options the application\nshould be launched with. Example: configuration \n property \n name some-property-name /name \n value some-property-value /value \n /property \n ... /configuration Names of properties XML file: properties.xml: Properties that are global to the Configuration\nPackage properties-{appName}.xml: Properties that are specific when launching\nan application with the specified appName within the Application\nPackage. After you are done with the above, remember to do mvn package to\ngenerate a new configuration package, which will be located in the\ntarget directory in your project. Zip structure of configuration package Apex Application Configuration Package files are zip files. You\ncan examine the content of any Application Configuration Package by\nusing unzip -t on your Linux command line. The structure of the zip\nfile is as follow: META-INF\n MANIFEST.MF\n properties.xml\n properties-{appname}.xml\nclasspath\n {classpath files}\nfiles\n {files} Launching with CLI -conf option of the launch command in CLI supports specifying configuration package in the local filesystem. Example: dt\\ launch mydtapp-1.0.0.apa -conf mydtconfig.apc This command expects both the application package and the configuration package to be in the local file system.",
+ "text": "Inside the project created by the archetype, these are the files that\nyou should know about when assembling your own configuration package: ./pom.xml\n./src/main/resources/classpath\n./src/main/resources/files\n./src/main/resources/META-INF/properties.xml\n./src/main/resources/META-INF/properties-{appname}.xml",
"title": "Assembling your own configuration package"
},
{
+ "location": "/application_packages/#pomxml",
+ "text": "Example: groupId com.example /groupId \n version 1.0.0 /version \n artifactId mydtconf /artifactId \n packaging jar /packaging \n !-- change these to the appropriate values -- \n name My Apex Application Configuration /name \n description My Custom Application Configuration Description /description \n properties \n apex.apppackage.name myapexapp /apex.apppackage.name \n apex.apppackage.minversion 1.0.0 /apex.apppackage.minversion \n apex.apppackage.maxversion 1.9999.9999 /apex.apppackage.maxversion \n apex.appconf.classpath classpath/* /apex.appconf.classpath \n apex.appconf.files files/* /apex.appconf.files \n /properties In pom.xml, you can change the following keys to your desired values groupId version artifactId name description You can also change the values of apex.apppackage.name apex.apppackage.minversion apex.apppackage.maxversion to reflect what Application Packages can be used with this configuration package. Apex will use this information to check whether a\nconfiguration package is compatible with the Application Package when you issue a launch command.",
+ "title": "pom.xml"
+ },
+ {
+ "location": "/application_packages/#srcmainresourcesclasspath",
+ "text": "Place any file in this directory that you\u2019d like to be copied to the\ncompute machines when launching an application and included in the\nclasspath of the application. Example of such files are Java properties\nfiles and jar files.",
+ "title": "./src/main/resources/classpath"
+ },
+ {
+ "location": "/application_packages/#srcmainresourcesfiles",
+ "text": "Place any file in this directory that you\u2019d like to be copied to the\ncompute machines when launching an application but not included in the\nclasspath of the application.",
+ "title": "./src/main/resources/files"
+ },
+ {
+ "location": "/application_packages/#properties-xml-file",
+ "text": "A properties xml file consists of a set of key-value pairs. The set of\nkey-value pairs specifies the configuration options the application\nshould be launched with. Example: configuration \n property \n name some-property-name /name \n value some-property-value /value \n /property \n ... /configuration Names of properties XML file: properties.xml: Properties that are global to the Configuration\nPackage properties-{appName}.xml: Properties that are specific when launching\nan application with the specified appName within the Application\nPackage. After you are done with the above, remember to do mvn package to\ngenerate a new configuration package, which will be located in the\ntarget directory in your project.",
+ "title": "Properties XML file"
+ },
+ {
+ "location": "/application_packages/#zip-structure-of-configuration-package",
+ "text": "Apex Application Configuration Package files are zip files. You\ncan examine the content of any Application Configuration Package by\nusing unzip -t on your Linux command line. The structure of the zip\nfile is as follow: META-INF\n MANIFEST.MF\n properties.xml\n properties-{appname}.xml\nclasspath\n {classpath files}\nfiles\n {files}",
+ "title": "Zip structure of configuration package"
+ },
+ {
+ "location": "/application_packages/#launching-with-cli",
+ "text": "-conf option of the launch command in CLI supports specifying configuration package in the local filesystem. Example: dt\\ launch mydtapp-1.0.0.apa -conf mydtconfig.apc This command expects both the application package and the configuration package to be in the local file system.",
+ "title": "Launching with CLI"
+ },
+ {
"location": "/operator_development/",
"text": "Operator Development Guide\n\n\nOperators are basic building blocks of an application built to run on\nApache Apex\u00a0platform. An application may consist of one or more\noperators each of which define some logical operation to be done on the\ntuples arriving at the operator. These operators are connected together\nusing streams forming a Directed Acyclic Graph (DAG). In other words, a streaming\napplication is represented by a DAG that consists of operations (called operators) and\ndata flow (called streams).\n\n\nIn this document we will discuss details on how an operator works and\nits internals. This document is intended to serve the following purposes\n\n\n\n\nApache Apex Operators\n\u00a0- Introduction to operator terminology and concepts.\n\n\nWriting Custom Operators\n\u00a0- Designing, coding and testing new operators from scratch. Includes code examples.\n\n\nOperator Reference\n - Details of operator internals, lifecycle, and best practices and optimizations.\n\n\n\n\n\n\nApache Apex Operators \n\n\nOperators - \u201cWhat\u201d in a nutshell\n\n\nOperators are independent units of logical operations which can\ncontribute in executing the business logic of a use case. For example,\nin an ETL workflow, a filtering operation can be represented by a single\noperator. This filtering operator will be responsible for doing just one\ntask in the ETL pipeline, i.e. filter incoming tuples. Operators do not\nimpose any restrictions on what can or cannot be done as part of a\noperator. An operator may as well contain the entire business logic.\nHowever, it is recommended, that the operators are light weight\nindependent tasks, in\norder to take advantage of the distributed framework that Apache Apex\nprovides.\u00a0The structure of a streaming application shares resemblance\nwith the way CPU pipelining works. CPU pipelining breaks down the\ncomputation engine into different stages viz. instruction fetch,\ninstruction decode, etc. so that each of them can perform their task on\ndifferent instructions\nparallely. Similarly,\nApache Apex APIs allow the user to break down their tasks into different\nstages so that all of the tasks can be executed on different tuples\nparallely.\n\n\n\n\nOperators - \u201cHow\u201d in a nutshell\n\n\nAn Apache Apex application runs as a YARN application. Hence, each of\nthe operators that the application DAG contains, runs in one of the\ncontainers provisioned by YARN.\u00a0Further, Apache Apex exposes APIs to\nallow the user to request bundling multiple operators in a single node,\na single container or even a single thread. We shall look at these calls\nin the reference sections [cite reference sections]. For now, consider\nan operator as some piece of code that runs on some machine of a YARN\ncluster.\n\n\nTypes of Operators\n\n\nAn operator works on one tuple at a time. These tuples may be supplied\nby other operators in the application or by external sources,\nsuch as a database or a message bus. Similarly, after the tuples are\nprocessed, these may be passed on to other operators, or stored into an external system. \nTherea are 3 type of operators based on function: \n\n\n\n\nInput Adapter\n - This is one of the starting points in\n the\u00a0application DAG and is responsible for getting tuples from an\n external system. At the same time, such data may also be generated\n by the operator itself, without interacting with the outside\n world.\u00a0These input tuples will form the initial universe of\n data\u00a0that the application works on.\n\n\nGeneric Operator\n - This type of operator accepts input tuples from\n the previous operators and passes\u00a0them on to the following operators\n in the DAG.\n\n\nOutput Adapter\n - This is one of the ending points in the application\n DAG and is responsible for writing the data out to some external\n system.\n\n\n\n\nNote: There can be multiple operators of all types in an application\nDAG.\n\n\nOperators Position in a DAG\n\n\nWe may refer to operators depending on their position with respect to\none another. For any operator opr (see image below), there are two types of operators.\n\n\n\n\nUpstream operators\n - These are the operators from which there is a\n directed path to opr\u00a0in the application DAG.\n\n\nDownstream operators\n - These are the operators to which there is a\n directed path from opr\u00a0in the application DAG.\n\n\n\n\nNote that there are no cycles formed in the application\u00a0DAG.\n\n\n\n\nPorts\n\n\nOperators in a DAG are connected together via directed flows\ncalled streams. Each\u00a0stream\u00a0has end-points located on the operators\ncalled ports. Therea are 2 types of ports.\n\n\n\n\nInput Port\n - This is a\u00a0port through which an operator accepts input\n tuples\u00a0from an upstream operator.\n\n\nOutput port\n - This is a\u00a0port through which an operator passes on the\n processed data to downstream operators.\n\n\n\n\nLooking at the number of input ports, an Input Adapter is an operator\nwith no input ports, a Generic operator has both input and output ports,\nwhile an Output Adapter has no output ports. At the same time, note that\nan operator may act as an Input Adapter while at the same time have an\ninput port. In such cases, the operator is getting data from two\ndifferent sources, viz.\u00a0the input stream from the input port and an\nexternal source.\n\n\n\n\n\n\nHow Operator Works\n\n\nAn operator passes through various stages during its lifetime. Each\nstage is an API call that the Streaming Application Master makes for an\noperator. \u00a0The following figure illustrates the stages through which an\noperator passes.\n\n\n\n\n\n\nThe \nsetup()\n call initializes the operator and prepares itself to\n start processing tuples.\n\n\nThe \nbeginWindow()\n call marks the beginning\u00a0of an\u00a0application window\n and allows for any processing to be done before a window starts.\n\n\nThe \nprocess()\n call belongs to the \nInputPort\n and gets triggered when\n any tuple arrives at the Input port of the operator. This call is\n specific only to Generic and Output adapters, since Input Adapters\n do not have an input port. This is made for all the tuples at the\n input port until the end window marker tuple is received on the\n input port.\n\n\nThe \nemitTuples()\n is the counterpart of \nprocess()\n call for Input\n Adapters.\n This call is used by Input adapters to emit any tuples that are\n fetched from the external systems, or generated by the operator.\n This method is called continuously until the pre-configured window\n time is elapsed, at which the end window marker tuple is sent out on\n the output port.\n\n\nThe \nendWindow()\n call marks the end of the window and allows for any\n processing to be done after the window ends.\n\n\nThe \nteardown()\n call is used for gracefully shutting down the\n operator and releasing any resources held by the operator.\n\n\n\n\nDeveloping Custom Operators \n\n\nAbout this tutorial\n\n\nThis tutorial will guide the user towards developing a operator from\nscratch. It includes all aspects of writing an operator including\ndesign, code and unit testing.\n\n\nIntroduction\n\n\nIn this tutorial, we will design and write, from scratch, an operator\ncalled Word Count. This operator will accept tuples of type String,\ncount the number of occurrences for each word appearing in the tuple and\nsend out the updated counts for all the words encountered in the tuple.\nFurther, the operator will also accept a file path on HDFS which will\ncontain the stop-words which need to be ignored when counting\noccurrences.\n\n\nDesign\n\n\nDesign of the operator must be finalized before starting to write an\noperator. Many aspects including the functionality, the data sources,\nthe types involved etc. need to be first finalized before writing the\noperator. Let us dive into each of these while considering the Word\nCount\u00a0operator.\n\n\nFunctionality\n\n\nWe can define the scope of operator functionality using the following\ntasks:\n\n\n\n\nParse the input tuple to identify the words in the tuple\n\n\nIdentify the stop-words in the tuple by looking up the stop-word\n file as configured\n\n\nFor each non-stop-word in the tuple, count the occurrences in that\n tuple and add it to a global counts\n\n\n\n\nLet\u2019s consider an example. Suppose we have the following tuples flow\ninto the Word Count operator.\n\n\n\n\nHumpty dumpty sat on a wall\n\n\nHumpty dumpty had a great fall\n\n\n\n\nInitially counts for all words\u00a0is 0. Once the first tuple is processed,\nthe counts that must be emitted are:\n\n\nhumpty - 1\ndumpty - 1\nsat - 1\nwall - 1\n\n\n\n\nNote that we are ignoring the stop-words, \u201con\u201d and \u201ca\u201d in this case.\nAlso note that as a rule, we\u2019ll ignore the case of the words when\ncounting occurrences.\n\n\nSimilarly, after the second tuple is processed, the counts that must be\nemitted are:\n\n\nhumpty - 2\ndumpty - 2\ngreat - 1\nfall - 1\n\n\n\n\nAgain, we ignore the words \n\u201chad\u201d\n and \n\u201ca\u201d\n since these are stop-words.\n\n\nNote that the most recent count for any word is correct count for that\nword. In other words, any new output for a word, invalidated all the\nprevious counts for that word.\n\n\nInputs\n\n\nAs seen from the example\u00a0above, the following\u00a0inputs are expected for\nthe operator:\n\n\n\n\nInput stream whose tuple type is String\n\n\nInput HDFS file path, pointing to a file containing stop-words\n\n\n\n\nOnly one input port is needed. The stop-word file will be small enough\nto be read completely in a single read. In addition this will be a one\ntime activity for the lifetime of the operator. This does not need a\nseparate input port.\n\n\n\n\nOutputs\n\n\nWe can define the output for this operator in multiple ways.\n\n\n\n\nThe operator may send out the set of counts for which the counts\n have changed after processing each tuple.\n\n\nSome applications might not need an update after every tuple, but\n only after\u00a0a certain time duration.\n\n\n\n\nLet us try and implement both these options depending on the\nconfiguration. Let us define a\u00a0boolean configuration parameter\n\n\u201csendPerTuple\u201d\n. The value of this parameter will indicate whether the\nupdated counts for words need to be emitted after processing each\ntuple\u00a0(true)\u00a0or after a certain time duration (false).\n\n\nThe type of information the operator will be sending out on the output\nport is the same for all the cases. This will be a \n key, value \n\u00a0pair,\nwhere the key\u00a0is the word while, the value is the latest count for that\nword. This means we just need one output port on which this information\nwill go out.\n\n\n\n\nConfiguration\n\n\nWe have the following configuration parameters:\n\n\n\n\nstopWordFilePath\n\u00a0- This parameter will store the path to the stop\n word file on HDFS as configured by the user.\n\n\nsendPerTuple\n\u00a0- This parameter decides whether we send out the\n updated counts after processing each tuple or at the end of a\n window. When set to true, the operator will send out the updated\n counts after each tuple, else it will send at the end of\n each\u00a0window.\n\n\n\n\nCode\n\n\nThe source code for the tutorial can be found here:\n\n\nhttps://github.com/DataTorrent/examples/tree/master/tutorials/operatorTutorial\n\n\nOperator Reference \n\n\nThe Operator Class\n\n\nThe operator will exist physically as a class which implements the\nOperator\u00a0interface. This interface will require implementations for the\nfollowing method calls:\n\n\n\n\nsetup(OperatorContext context)\n\n\nbeginWindow(long windowId)\n\n\nendWindow()\n\n\ntearDown()\n\n\n\n\nIn order to simplify the creation of an operator, Apache\u00a0Apex\nlibrary also provides a base class \u201cBaseOperator\u201d which has empty\nimplementations for these methods. Please refer to the \nApex Operators\n\u00a0section and the\n\nReference\n\u00a0section for details on these.\n\n\nWe extend the class \u201cBaseOperator\u201d to create our own operator\n\u201cWordCountOperator\u201d.\n\n\npublic class WordCountOperator extends BaseOperator\n{\n}\n\n\n\n\nClass (Operator) properties\n\n\nWe define the following class variables:\n\n\n\n\nsendPerTuple\n\u00a0- Configures the output frequency from the operator\n\n\n\n\nprivate boolean sendPerTuple = true; // default\n\n\n\n\n\n\nstopWordFilePath\n\u00a0- Stores the path to the stop words file on HDFS\n\n\n\n\nprivate String stopWordFilePath;\u00a0// no default\n\n\n\n\n\n\nstopWords\n\u00a0- Stores the stop words read from the configured file\n\n\n\n\nprivate transient String[] stopWords;\n\n\n\n\n\n\nglobalCounts\n\u00a0- A Map which stores the counts of all the words\n encountered so far. Note that this variable is non transient, which\n means that this variable is saved as part of the checkpoint and can be recovered in event of a crash.\n\n\n\n\nprivate Map\nString, Long\n globalCounts;\n\n\n\n\n\n\nupdatedCounts\n\u00a0- A Map which stores the counts for only the most\n recent tuple(s). sendPerTuple configuration determines whether to store the most recent or the recent\n window worth of tuples.\n\n\n\n\nprivate transient Map\nString, Long\n updatedCounts;\n\n\n\n\n\n\ninput\n - The input port for the operator. The type of this input port\n is String\u00a0which means it will only accept tuples of type String. The\n definition of an input port requires implementation of a method\n called process(String tuple), which should\u00a0have the processing logic\n for the input tuple which \u00a0arrives at this input port. We delegate\n this task to another method called processTuple(String tuple). This\n helps in keeping the operator classes extensible by overriding the\n processing logic for the input tuples.\n\n\n\n\npublic transient DefaultInputPort\nString\n input = new \u00a0 \u00a0\nDefaultInputPort\nString\n()\n{\n\u00a0\u00a0\u00a0\u00a0@Override\n\u00a0\u00a0\u00a0\u00a0public void process(String tuple)\n\u00a0\u00a0\u00a0\u00a0{\n \u00a0\u00a0\u00a0\u00a0processTuple(tuple);\n\u00a0\u00a0\u00a0\u00a0}\n};\n\n\n\n\n\n\noutput - The output port for the operator. The type of this port is\n Entry \n String, Long \n, which means the operator will emit \n word,\n count \n pairs for the updated counts.\n\n\n\n\npublic transient DefaultOutputPort \nEntry\nString, Long\n output = new\nDefaultOutputPort\nEntry\nString,Long\n();\n\n\n\n\nThe Constructor\n\n\nThe constructor is the place where we initialize the non-transient data\nstructures,\u00a0since\nconstructor is called just once per activation of an operator. With regards to Word Count\u00a0operator, we initialize the globalCounts variable in the constructor.\n\n\nglobalCounts = Maps.newHashMap();\n\n\n\n\nSetup call\n\n\nThe setup method is called only once during an operator lifetime and its purpose is to allow \nthe operator to set itself up for processing incoming streams. Transient objects in the operator are\nnot serialized and checkpointed. Hence, it is essential that such objects initialized in the setup call. \nIn case of operator failure, the operator will be redeployed (most likely on a different container). The setup method called by the Apache Apex engine allows the operator to prepare for execution in the new container.\n\n\nThe following tasks are executed as part of the setup call:\n\n\n\n\nRead the stop-word list from HDFS and store it in the\n stopWords\u00a0array\n\n\nInitialize updatedCounts\u00a0variable. This will store the updated\n counts for words in most recent tuples processed by the operator.\n As a transient variable, the value will be lost when operator fails.\n\n\n\n\nBegin Window call\n\n\nThe begin window call signals the start of an application window. With \nregards to Word Count Operator, we are expecting updated counts for the most recent window of\ndata if the sendPerTuple\u00a0is set to false. Hence, we clear the updatedCounts\u00a0variable in the begin window\ncall and start accumulating the counts till the end window call.\n\n\nProcess Tuple call\n\n\nThe processTuple\u00a0method is called by the process\u00a0method of the input\nport, input. This method defines the processing logic for the current\ntuple that is received at the input port. As part of this method, we\nidentify the words in the current tuple and update the globalCounts\u00a0and\nthe updatedCounts\u00a0variables. In addition, if the sendPerTuple\u00a0variable\nis set to true, we also emit the words\u00a0and corresponding counts in\nupdatedCounts\u00a0to the output port. Note\u00a0that in this case (sendPerTuple =\ntrue), we clear the updatedCounts\u00a0variable in every call to\nprocessTuple.\n\n\nEnd Window call\n\n\nThis call signals the end of an application window. With regards to Word\nCount Operator, we emit the updatedCounts\u00a0to the output port if the\nsendPerTuple\u00a0flag is set to false.\n\n\nTeardown call\n\n\nThis method allows the operator to gracefully shut down itself after\nreleasing the resources that it has acquired. With regards to our operator,\nwe call the shutDown\u00a0method which shuts down the operator along with any\ndownstream operators.\n\n\nTesting your Operator\n\n\nAs part of testing our operator, we test the following two facets:\n\n\n\n\nTest output of the operator after processing a single tuple\n\n\nTest output of the operator after processing of a window of tuples\n\n\n\n\nThe unit tests for the WordCount operator are available in the class\nWordCountOperatorTest.java. We simulate the behavior of the engine by\nusing the test utilities provided by Apache Apex libraries. We simulate\nthe setup, beginWindow, process\u00a0method of the input port and\nendWindow\u00a0calls and compare the output received at the simulated output\nports.\n\n\n\n\nInvoke constructor; non-transients initialized.\n\n\nCopy state from checkpoint -- initialized values from step 1 are\nreplaced.\n\n\n\n\nMalhar Operator Library\n\n\nTo see the full list of Apex Malhar operators along with related documentation, visit \nApex Malhar on Github",
"title": "Operators"
@@ -332,10 +667,25 @@
},
{
"location": "/operator_development/#design",
- "text": "Design of the operator must be finalized before starting to write an\noperator. Many aspects including the functionality, the data sources,\nthe types involved etc. need to be first finalized before writing the\noperator. Let us dive into each of these while considering the Word\nCount\u00a0operator. Functionality We can define the scope of operator functionality using the following\ntasks: Parse the input tuple to identify the words in the tuple Identify the stop-words in the tuple by looking up the stop-word\n file as configured For each non-stop-word in the tuple, count the occurrences in that\n tuple and add it to a global counts Let\u2019s consider an example. Suppose we have the following tuples flow\ninto the Word Count operator. Humpty dumpty sat on a wall Humpty dumpty had a great fall Initially counts for all words\u00a0is 0. Once the first tuple is processed,\nthe counts that must be emitted are: humpty - 1\ndumpty - 1\nsat - 1\nwall - 1 Note that we are ignoring the stop-words, \u201con\u201d and \u201ca\u201d in this case.\nAlso note that as a rule, we\u2019ll ignore the case of the words when\ncounting occurrences. Similarly, after the second tuple is processed, the counts that must be\nemitted are: humpty - 2\ndumpty - 2\ngreat - 1\nfall - 1 Again, we ignore the words \u201chad\u201d and \u201ca\u201d since these are stop-words. Note that the most recent count for any word is correct count for that\nword. In other words, any new output for a word, invalidated all the\nprevious counts for that word. Inputs As seen from the example\u00a0above, the following\u00a0inputs are expected for\nthe operator: Input stream whose tuple type is String Input HDFS file path, pointing to a file containing stop-words Only one input port is needed. The stop-word file will be small enough\nto be read completely in a single read. In addition this will be a one\ntime activity for the lifetime of the operator. This does not need a\nseparate input port. Outputs We can define the output for this operator in multiple ways. The operator may send out the set of counts for which the counts\n have changed after processing each tuple. Some applications might not need an update after every tuple, but\n only after\u00a0a certain time duration. Let us try and implement both these options depending on the\nconfiguration. Let us define a\u00a0boolean configuration parameter \u201csendPerTuple\u201d . The value of this parameter will indicate whether the\nupdated counts for words need to be emitted after processing each\ntuple\u00a0(true)\u00a0or after a certain time duration (false). The type of information the operator will be sending out on the output\nport is the same for all the cases. This will be a key, value \u00a0pair,\nwhere the key\u00a0is the word while, the value is the latest count for that\nword. This means we just need one output port on which this information\nwill go out.",
+ "text": "Design of the operator must be finalized before starting to write an\noperator. Many aspects including the functionality, the data sources,\nthe types involved etc. need to be first finalized before writing the\noperator. Let us dive into each of these while considering the Word\nCount\u00a0operator.",
"title": "Design"
},
{
+ "location": "/operator_development/#functionality",
+ "text": "We can define the scope of operator functionality using the following\ntasks: Parse the input tuple to identify the words in the tuple Identify the stop-words in the tuple by looking up the stop-word\n file as configured For each non-stop-word in the tuple, count the occurrences in that\n tuple and add it to a global counts Let\u2019s consider an example. Suppose we have the following tuples flow\ninto the Word Count operator. Humpty dumpty sat on a wall Humpty dumpty had a great fall Initially counts for all words\u00a0is 0. Once the first tuple is processed,\nthe counts that must be emitted are: humpty - 1\ndumpty - 1\nsat - 1\nwall - 1 Note that we are ignoring the stop-words, \u201con\u201d and \u201ca\u201d in this case.\nAlso note that as a rule, we\u2019ll ignore the case of the words when\ncounting occurrences. Similarly, after the second tuple is processed, the counts that must be\nemitted are: humpty - 2\ndumpty - 2\ngreat - 1\nfall - 1 Again, we ignore the words \u201chad\u201d and \u201ca\u201d since these are stop-words. Note that the most recent count for any word is correct count for that\nword. In other words, any new output for a word, invalidated all the\nprevious counts for that word.",
+ "title": "Functionality"
+ },
+ {
+ "location": "/operator_development/#inputs",
+ "text": "As seen from the example\u00a0above, the following\u00a0inputs are expected for\nthe operator: Input stream whose tuple type is String Input HDFS file path, pointing to a file containing stop-words Only one input port is needed. The stop-word file will be small enough\nto be read completely in a single read. In addition this will be a one\ntime activity for the lifetime of the operator. This does not need a\nseparate input port.",
+ "title": "Inputs"
+ },
+ {
+ "location": "/operator_development/#outputs",
+ "text": "We can define the output for this operator in multiple ways. The operator may send out the set of counts for which the counts\n have changed after processing each tuple. Some applications might not need an update after every tuple, but\n only after\u00a0a certain time duration. Let us try and implement both these options depending on the\nconfiguration. Let us define a\u00a0boolean configuration parameter \u201csendPerTuple\u201d . The value of this parameter will indicate whether the\nupdated counts for words need to be emitted after processing each\ntuple\u00a0(true)\u00a0or after a certain time duration (false). The type of information the operator will be sending out on the output\nport is the same for all the cases. This will be a key, value \u00a0pair,\nwhere the key\u00a0is the word while, the value is the latest count for that\nword. This means we just need one output port on which this information\nwill go out.",
+ "title": "Outputs"
+ },
+ {
"location": "/operator_development/#configuration",
"text": "We have the following configuration parameters: stopWordFilePath \u00a0- This parameter will store the path to the stop\n word file on HDFS as configured by the user. sendPerTuple \u00a0- This parameter decides whether we send out the\n updated counts after processing each tuple or at the end of a\n window. When set to true, the operator will send out the updated\n counts after each tuple, else it will send at the end of\n each\u00a0window.",
"title": "Configuration"
@@ -347,10 +697,50 @@
},
{
"location": "/operator_development/#operator-reference",
- "text": "The Operator Class The operator will exist physically as a class which implements the\nOperator\u00a0interface. This interface will require implementations for the\nfollowing method calls: setup(OperatorContext context) beginWindow(long windowId) endWindow() tearDown() In order to simplify the creation of an operator, Apache\u00a0Apex\nlibrary also provides a base class \u201cBaseOperator\u201d which has empty\nimplementations for these methods. Please refer to the Apex Operators \u00a0section and the Reference \u00a0section for details on these. We extend the class \u201cBaseOperator\u201d to create our own operator\n\u201cWordCountOperator\u201d. public class WordCountOperator extends BaseOperator\n{\n} Class (Operator) properties We define the following class variables: sendPerTuple \u00a0- Configures the output frequency from the operator private boolean sendPerTuple = true; // default stopWordFilePath \u00a0- Stores the path to the stop words file on HDFS private String stopWordFilePath;\u00a0// no default stopWords \u00a0- Stores the stop words read from the configured file private transient String[] stopWords; globalCounts \u00a0- A Map which stores the counts of all the words\n encountered so far. Note that this variable is non transient, which\n means that this variable is saved as part of the checkpoint and can be recovered in event of a crash. private Map String, Long globalCounts; updatedCounts \u00a0- A Map which stores the counts for only the most\n recent tuple(s). sendPerTuple configuration determines whether to store the most recent or the recent\n window worth of tuples. private transient Map String, Long updatedCounts; input - The input port for the operator. The type of this input port\n is String\u00a0which means it will only accept tuples of type String. The\n definition of an input port requires implementation of a method\n called process(String tuple), which should\u00a0have the processing logic\n for the input tuple which \u00a0arrives at this input port. We delegate\n this task to another method called processTuple(String tuple). This\n helps in keeping the operator classes extensible by overriding the\n processing logic for the input tuples. public transient DefaultInputPort String input = new \u00a0 \u00a0\nDefaultInputPort String ()\n{\n\u00a0\u00a0\u00a0\u00a0@Override\n\u00a0\u00a0\u00a0\u00a0public void process(String tuple)\n\u00a0\u00a0\u00a0\u00a0{\n \u00a0\u00a0\u00a0\u00a0processTuple(tuple);\n\u00a0\u00a0\u00a0\u00a0}\n}; output - The output port for the operator. The type of this port is\n Entry String, Long , which means the operator will emit word,\n count pairs for the updated counts. public transient DefaultOutputPort Entry String, Long output = new\nDefaultOutputPort Entry String,Long (); The Constructor The constructor is the place where we initialize the non-transient data\nstructures,\u00a0since\nconstructor is called just once per activation of an operator. With regards to Word Count\u00a0operator, we initialize the globalCounts variable in the constructor. globalCounts = Maps.newHashMap(); Setup call The setup method is called only once during an operator lifetime and its purpose is to allow \nthe operator to set itself up for processing incoming streams. Transient objects in the operator are\nnot serialized and checkpointed. Hence, it is essential that such objects initialized in the setup call. \nIn case of operator failure, the operator will be redeployed (most likely on a different container). The setup method called by the Apache Apex engine allows the operator to prepare for execution in the new container. The following tasks are executed as part of the setup call: Read the stop-word list from HDFS and store it in the\n stopWords\u00a0array Initialize updatedCounts\u00a0variable. This will store the updated\n counts for words in most recent tuples processed by the operator.\n As a transient variable, the value will be lost when operator fails. Begin Window call The begin window call signals the start of an application window. With \nregards to Word Count Operator, we are expecting updated counts for the most recent window of\ndata if the sendPerTuple\u00a0is set to false. Hence, we clear the updatedCounts\u00a0variable in the begin window\ncall and start accumulating the counts till the end window call. Process Tuple call The processTuple\u00a0method is called by the process\u00a0method of the input\nport, input. This method defines the processing logic for the current\ntuple that is received at the input port. As part of this method, we\nidentify the words in the current tuple and update the globalCounts\u00a0and\nthe updatedCounts\u00a0variables. In addition, if the sendPerTuple\u00a0variable\nis set to true, we also emit the words\u00a0and corresponding counts in\nupdatedCounts\u00a0to the output port. Note\u00a0that in this case (sendPerTuple =\ntrue), we clear the updatedCounts\u00a0variable in every call to\nprocessTuple. End Window call This call signals the end of an application window. With regards to Word\nCount Operator, we emit the updatedCounts\u00a0to the output port if the\nsendPerTuple\u00a0flag is set to false. Teardown call This method allows the operator to gracefully shut down itself after\nreleasing the resources that it has acquired. With regards to our operator,\nwe call the shutDown\u00a0method which shuts down the operator along with any\ndownstream operators.",
+ "text": "",
"title": "Operator Reference "
},
{
+ "location": "/operator_development/#the-operator-class",
+ "text": "The operator will exist physically as a class which implements the\nOperator\u00a0interface. This interface will require implementations for the\nfollowing method calls: setup(OperatorContext context) beginWindow(long windowId) endWindow() tearDown() In order to simplify the creation of an operator, Apache\u00a0Apex\nlibrary also provides a base class \u201cBaseOperator\u201d which has empty\nimplementations for these methods. Please refer to the Apex Operators \u00a0section and the Reference \u00a0section for details on these. We extend the class \u201cBaseOperator\u201d to create our own operator\n\u201cWordCountOperator\u201d. public class WordCountOperator extends BaseOperator\n{\n}",
+ "title": "The Operator Class"
+ },
+ {
+ "location": "/operator_development/#class-operator-properties",
+ "text": "We define the following class variables: sendPerTuple \u00a0- Configures the output frequency from the operator private boolean sendPerTuple = true; // default stopWordFilePath \u00a0- Stores the path to the stop words file on HDFS private String stopWordFilePath;\u00a0// no default stopWords \u00a0- Stores the stop words read from the configured file private transient String[] stopWords; globalCounts \u00a0- A Map which stores the counts of all the words\n encountered so far. Note that this variable is non transient, which\n means that this variable is saved as part of the checkpoint and can be recovered in event of a crash. private Map String, Long globalCounts; updatedCounts \u00a0- A Map which stores the counts for only the most\n recent tuple(s). sendPerTuple configuration determines whether to store the most recent or the recent\n window worth of tuples. private transient Map String, Long updatedCounts; input - The input port for the operator. The type of this input port\n is String\u00a0which means it will only accept tuples of type String. The\n definition of an input port requires implementation of a method\n called process(String tuple), which should\u00a0have the processing logic\n for the input tuple which \u00a0arrives at this input port. We delegate\n this task to another method called processTuple(String tuple). This\n helps in keeping the operator classes extensible by overriding the\n processing logic for the input tuples. public transient DefaultInputPort String input = new \u00a0 \u00a0\nDefaultInputPort String ()\n{\n\u00a0\u00a0\u00a0\u00a0@Override\n\u00a0\u00a0\u00a0\u00a0public void process(String tuple)\n\u00a0\u00a0\u00a0\u00a0{\n \u00a0\u00a0\u00a0\u00a0processTuple(tuple);\n\u00a0\u00a0\u00a0\u00a0}\n}; output - The output port for the operator. The type of this port is\n Entry String, Long , which means the operator will emit word,\n count pairs for the updated counts. public transient DefaultOutputPort Entry String, Long output = new\nDefaultOutputPort Entry String,Long ();",
+ "title": "Class (Operator) properties"
+ },
+ {
+ "location": "/operator_development/#the-constructor",
+ "text": "The constructor is the place where we initialize the non-transient data\nstructures,\u00a0since\nconstructor is called just once per activation of an operator. With regards to Word Count\u00a0operator, we initialize the globalCounts variable in the constructor. globalCounts = Maps.newHashMap();",
+ "title": "The Constructor"
+ },
+ {
+ "location": "/operator_development/#setup-call",
+ "text": "The setup method is called only once during an operator lifetime and its purpose is to allow \nthe operator to set itself up for processing incoming streams. Transient objects in the operator are\nnot serialized and checkpointed. Hence, it is essential that such objects initialized in the setup call. \nIn case of operator failure, the operator will be redeployed (most likely on a different container). The setup method called by the Apache Apex engine allows the operator to prepare for execution in the new container. The following tasks are executed as part of the setup call: Read the stop-word list from HDFS and store it in the\n stopWords\u00a0array Initialize updatedCounts\u00a0variable. This will store the updated\n counts for words in most recent tuples processed by the operator.\n As a transient variable, the value will be lost when operator fails.",
+ "title": "Setup call"
+ },
+ {
+ "location": "/operator_development/#begin-window-call",
+ "text": "The begin window call signals the start of an application window. With \nregards to Word Count Operator, we are expecting updated counts for the most recent window of\ndata if the sendPerTuple\u00a0is set to false. Hence, we clear the updatedCounts\u00a0variable in the begin window\ncall and start accumulating the counts till the end window call.",
+ "title": "Begin Window call"
+ },
+ {
+ "location": "/operator_development/#process-tuple-call",
+ "text": "The processTuple\u00a0method is called by the process\u00a0method of the input\nport, input. This method defines the processing logic for the current\ntuple that is received at the input port. As part of this method, we\nidentify the words in the current tuple and update the globalCounts\u00a0and\nthe updatedCounts\u00a0variables. In addition, if the sendPerTuple\u00a0variable\nis set to true, we also emit the words\u00a0and corresponding counts in\nupdatedCounts\u00a0to the output port. Note\u00a0that in this case (sendPerTuple =\ntrue), we clear the updatedCounts\u00a0variable in every call to\nprocessTuple.",
+ "title": "Process Tuple call"
+ },
+ {
+ "location": "/operator_development/#end-window-call",
+ "text": "This call signals the end of an application window. With regards to Word\nCount Operator, we emit the updatedCounts\u00a0to the output port if the\nsendPerTuple\u00a0flag is set to false.",
+ "title": "End Window call"
+ },
+ {
+ "location": "/operator_development/#teardown-call",
+ "text": "This method allows the operator to gracefully shut down itself after\nreleasing the resources that it has acquired. With regards to our operator,\nwe call the shutDown\u00a0method which shuts down the operator along with any\ndownstream operators.",
+ "title": "Teardown call"
+ },
+ {
"location": "/operator_development/#testing-your-operator",
"text": "As part of testing our operator, we test the following two facets: Test output of the operator after processing a single tuple Test output of the operator after processing of a window of tuples The unit tests for the WordCount operator are available in the class\nWordCountOperatorTest.java. We simulate the behavior of the engine by\nusing the test utilities provided by Apache Apex libraries. We simulate\nthe setup, beginWindow, process\u00a0method of the input port and\nendWindow\u00a0calls and compare the output received at the simulated output\nports. Invoke constructor; non-transients initialized. Copy state from checkpoint -- initialized values from step 1 are\nreplaced.",
"title": "Testing your Operator"
@@ -417,10 +807,25 @@
},
{
"location": "/dtcli/#dtcli-commands",
- "text": "dtCli can be launched by running following command dtcli Help on all commands is available via \u201chelp\u201d command in the CLI Global Commands GLOBAL COMMANDS EXCEPT WHEN CHANGING LOGICAL PLAN:\n\nalias alias-name command\n Create a command alias\n\nbegin-macro name\n Begin Macro Definition ($1...$9 to access parameters and type 'end' to end the definition)\n\nconnect app-id\n Connect to an app\n\ndump-properties-file out-file jar-file class-name\n Dump the properties file of an app class\n\necho [arg ...]\n Echo the arguments\n\nexit\n Exit the CLI\n\nget-app-info app-id\n Get the information of an app\n\nget-app-package-info app-package-file\n Get info on the app package file\n\nget-app-package-operator-properties app-package-file operator-class\n Get operator properties within the given app package\n\nget-app-package-operators [options] app-package-file [search-term]\n Get operators within the given app package\n Options:\n -parent Specify the parent class for the operators\n\nget-config-parameter [parameter-name]\n Get the configuration parameter\n\nget-jar-operator-classes [options] jar-files-comma-separated [search-term]\n List operators in a jar list\n Options:\n -parent Specify the parent class for the operators\n\nget-jar-operator-properties jar-files-comma-separated operator-class-name\n List properties in specified operator\n\nhelp [command]\n Show help\n\nkill-app app-id [app-id ...]\n Kill an app\n\n launch [options] jar-file/json-file/properties-file/app-package-file [matching-app-name]\n Launch an app\n Options:\n -apconf app package configuration file Specify an application\n configuration file\n within the app\n package if launching\n an app package.\n -archives comma separated list of archives Specify comma\n separated archives\n to be unarchived on\n the compute machines.\n -conf configuration file Specify an\n application\n configuration file.\n -D property=value Use value for given\n property.\n -exactMatch Only consider\n applications with\n exact app name\n -files comma separated list of files Specify comma\n separated files to\n be copied on the\n compute machines.\n -ignorepom Do not run maven to\n find the dependency\n -libjars comma separated list of libjars Specify comma\n separated jar files\n or other resource\n files to include in\n the classpath.\n -local Run application in\n local mode.\n -originalAppId application id Specify original\n application\n identifier for restart.\n -queue queue name Specify the queue to\n launch the application\n\nlist-application-attributes\n Lists the application attributes\nlist-apps [pattern]\n List applications\nlist-operator-attributes\n Lists the operator attributes\nlist-port-attributes\n Lists the port attributes\nset-pager on/off\n Set the pager program for output\nshow-logical-plan [options] jar-file/app-package-file [class-name]\n List apps in a jar or show logical plan of an app class\n Options:\n -exactMatch Only consider exact match\n for app name\n -ignorepom Do not run maven to find\n the dependency\n -libjars comma separated list of jars Specify comma separated\n jar/resource files to\n include in the classpath.\nshutdown-app app-id [app-id ...]\n Shutdown an app\nsource file\n Execute the commands in a file Commands after connecting to an application COMMANDS WHEN CONNECTED TO AN APP (via connect appid ) EXCEPT WHEN CHANGING LOGICAL PLAN:\n\nbegin-logical-plan-change\n Begin Logical Plan Change\ndump-properties-file out-file [jar-file] [class-name]\n Dump the properties file of an app class\nget-app-attributes [attribute-name]\n Get attributes of the connected app\nget-app-info [app-id]\n Get the information of an app\nget-operator-attributes operator-name [attribute-name]\n Get attributes of an operator\nget-operator-properties operator-name [property-name]\n Get properties of a logical operator\nget-physical-operator-properties [options] operator-id\n Get properties of a physical operator\n Options:\n -propertyName property name The name of the property whose\n value needs to be retrieved\n -waitTime wait time How long to wait to get the result\nget-port-attributes operator-name port-name [attribute-name]\n Get attributes of a port\nget-recording-info [operator-id] [start-time]\n Get tuple recording info\nkill-app [app-id ...]\n Kill an app\nkill-container container-id [container-id ...]\n Kill a container\nlist-containers\n List containers\nlist-operators [pattern]\n List operators\nset-operator-property operator-name property-name property-value\n Set a property of an operator\nset-physical-operator-property operator-id property-name property-value\n Set a property of an operator\nshow-logical-plan [options] [jar-file/app-package-file] [class-name]\n Show logical plan of an app class\n Options:\n -exactMatch Only consider exact match\n for app name\n -ignorepom Do not run maven to find\n the dependency\n -libjars comma separated list of jars Specify comma separated\n jar/resource files to\n include in the classpath.\nshow-physical-plan\n Show physical plan\nshutdown-app [app-id ...]\n Shutdown an app\nstart-recording operator-id [port-name] [num-windows]\n Start recording\nstop-recording operator-id [port-name]\n Stop recording\nwait timeout\n Wait for completion of current application Commands when changing the logical plan COMMANDS WHEN CHANGING LOGICAL PLAN (via begin-logical-plan-change):\n\nabort\n Abort the plan change\nadd-stream-sink stream-name to-operator-name to-port-name\n Add a sink to an existing stream\ncreate-operator operator-name class-name\n Create an operator\ncreate-stream stream-name from-operator-name from-port-name to-operator-name to-port-name\n Create a stream\nhelp [command]\n Show help\nremove-operator operator-name\n Remove an operator\nremove-stream stream-name\n Remove a stream\nset-operator-attribute operator-name attr-name attr-value\n Set an attribute of an operator\nset-operator-property operator-name property-name property-value\n Set a property of an operator\nset-port-attribute operator-name port-name attr-name attr-value\n Set an attribute of a port\nset-stream-attribute stream-name attr-name attr-value\n Set an attribute of a stream\nshow-queue\n Show the queue of the plan change\nsubmit\n Submit the plan change",
+ "text": "dtCli can be launched by running following command dtcli Help on all commands is available via \u201chelp\u201d command in the CLI",
"title": "dtcli Commands"
},
{
+ "location": "/dtcli/#global-commands",
+ "text": "GLOBAL COMMANDS EXCEPT WHEN CHANGING LOGICAL PLAN:\n\nalias alias-name command\n Create a command alias\n\nbegin-macro name\n Begin Macro Definition ($1...$9 to access parameters and type 'end' to end the definition)\n\nconnect app-id\n Connect to an app\n\ndump-properties-file out-file jar-file class-name\n Dump the properties file of an app class\n\necho [arg ...]\n Echo the arguments\n\nexit\n Exit the CLI\n\nget-app-info app-id\n Get the information of an app\n\nget-app-package-info app-package-file\n Get info on the app package file\n\nget-app-package-operator-properties app-package-file operator-class\n Get operator properties within the given app package\n\nget-app-package-operators [options] app-package-file [search-term]\n Get operators within the given app package\n Options:\n -parent Specify the parent class for the operators\n\nget-config-parameter [parameter-name]\n Get the configuration parameter\n\nget-jar-operator-classes [options] jar-files-comma-separated [search-term]\n List operators in a jar list\n Options:\n -parent Specify the parent class for the operators\n\nget-jar-operator-properties jar-files-comma-separated operator-class-name\n List properties in specified operator\n\nhelp [command]\n Show help\n\nkill-app app-id [app-id ...]\n Kill an app\n\n launch [options] jar-file/json-file/properties-file/app-package-file [matching-app-name]\n Launch an app\n Options:\n -apconf app package configuration file Specify an application\n configuration file\n within the app\n package if launching\n an app package.\n -archives comma separated list of archives Specify comma\n separated archives\n to be unarchived on\n the compute machines.\n -conf configuration file Specify an\n application\n configuration file.\n -D property=value Use value for given\n property.\n -exactMatch Only consider\n applications with\n exact app name\n -files comma separated list of files Specify comma\n separated files to\n be copied on the\n compute machines.\n -ignorepom Do not run maven to\n find the dependency\n -libjars comma separated list of libjars Specify comma\n separated jar files\n or other resource\n files to include in\n the classpath.\n -local Run application in\n local mode.\n -originalAppId application id Specify original\n application\n identifier for restart.\n -queue queue name Specify the queue to\n launch the application\n\nlist-application-attributes\n Lists the application attributes\nlist-apps [pattern]\n List applications\nlist-operator-attributes\n Lists the operator attributes\nlist-port-attributes\n Lists the port attributes\nset-pager on/off\n Set the pager program for output\nshow-logical-plan [options] jar-file/app-package-file [class-name]\n List apps in a jar or show logical plan of an app class\n Options:\n -exactMatch Only consider exact match\n for app name\n -ignorepom Do not run maven to find\n the dependency\n -libjars comma separated list of jars Specify comma separated\n jar/resource files to\n include in the classpath.\nshutdown-app app-id [app-id ...]\n Shutdown an app\nsource file\n Execute the commands in a file",
+ "title": "Global Commands"
+ },
+ {
+ "location": "/dtcli/#commands-after-connecting-to-an-application",
+ "text": "COMMANDS WHEN CONNECTED TO AN APP (via connect appid ) EXCEPT WHEN CHANGING LOGICAL PLAN:\n\nbegin-logical-plan-change\n Begin Logical Plan Change\ndump-properties-file out-file [jar-file] [class-name]\n Dump the properties file of an app class\nget-app-attributes [attribute-name]\n Get attributes of the connected app\nget-app-info [app-id]\n Get the information of an app\nget-operator-attributes operator-name [attribute-name]\n Get attributes of an operator\nget-operator-properties operator-name [property-name]\n Get properties of a logical operator\nget-physical-operator-properties [options] operator-id\n Get properties of a physical operator\n Options:\n -propertyName property name The name of the property whose\n value needs to be retrieved\n -waitTime wait time How long to wait to get the result\nget-port-attributes operator-name port-name [attribute-name]\n Get attributes of a port\nget-recording-info [operator-id] [start-time]\n Get tuple recording info\nkill-app [app-id ...]\n Kill an app\nkill-container container-id [container-id ...]\n Kill a container\nlist-containers\n List containers\nlist-operators [pattern]\n List operators\nset-operator-property operator-name property-name property-value\n Set a property of an operator\nset-physical-operator-property operator-id property-name property-value\n Set a property of an operator\nshow-logical-plan [options] [jar-file/app-package-file] [class-name]\n Show logical plan of an app class\n Options:\n -exactMatch Only consider exact match\n for app name\n -ignorepom Do not run maven to find\n the dependency\n -libjars comma separated list of jars Specify comma separated\n jar/resource files to\n include in the classpath.\nshow-physical-plan\n Show physical plan\nshutdown-app [app-id ...]\n Shutdown an app\nstart-recording operator-id [port-name] [num-windows]\n Start recording\nstop-recording operator-id [port-name]\n Stop recording\nwait timeout\n Wait for completion of current application",
+ "title": "Commands after connecting to an application"
+ },
+ {
+ "location": "/dtcli/#commands-when-changing-the-logical-plan",
+ "text": "COMMANDS WHEN CHANGING LOGICAL PLAN (via begin-logical-plan-change):\n\nabort\n Abort the plan change\nadd-stream-sink stream-name to-operator-name to-port-name\n Add a sink to an existing stream\ncreate-operator operator-name class-name\n Create an operator\ncreate-stream stream-name from-operator-name from-port-name to-operator-name to-port-name\n Create a stream\nhelp [command]\n Show help\nremove-operator operator-name\n Remove an operator\nremove-stream stream-name\n Remove a stream\nset-operator-attribute operator-name attr-name attr-value\n Set an attribute of an operator\nset-operator-property operator-name property-name property-value\n Set a property of an operator\nset-port-attribute operator-name port-name attr-name attr-value\n Set an attribute of a port\nset-stream-attribute stream-name attr-name attr-value\n Set an attribute of a stream\nshow-queue\n Show the queue of the plan change\nsubmit\n Submit the plan change",
+ "title": "Commands when changing the logical plan"
+ },
+ {
"location": "/dtcli/#examples",
"text": "An example of defining a custom macro. The macro updates a running application by inserting a new operator. It takes three parameters and executes a logical plan changes. dt begin-macro add-console-output\nmacro begin-logical-plan-change\nmacro create-operator $1 com.datatorrent.lib.io.ConsoleOutputOperator\nmacro create-stream stream_$1 $2 $3 $1 in\nmacro submit Then execute the add-console-output macro like this dt add-console-output xyz opername portname This macro then expands to run the following command begin-logical-plan-change\ncreate-operator xyz com.datatorrent.lib.io.ConsoleOutputOperator\ncreate-stream stream_xyz opername portname xyz in\nsubmit Note : To perform runtime logical plan changes, like ability to add new operators,\nthey must be part of the jar files that were deployed at application launch time.",
"title": "Examples"
diff --git a/docs/apex-3.3/operator_development/index.html b/docs/apex-3.3/operator_development/index.html
index 2a9527b..deef122 100644
--- a/docs/apex-3.3/operator_development/index.html
+++ b/docs/apex-3.3/operator_development/index.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>Operators - Apache Apex Documentation</title>
@@ -22,6 +25,8 @@
<script>
// Current page data
var mkdocs_page_name = "Operators";
+ var mkdocs_page_input_path = "operator_development.md";
+ var mkdocs_page_url = "/operator_development/";
</script>
<script src="../js/jquery-2.1.1.min.js"></script>
@@ -599,7 +604,7 @@
<div class="rst-footer-buttons" role="navigation" aria-label="footer navigation">
- <a href="../autometrics/" class="btn btn-neutral float-right" title="AutoMetric API"/>Next <span class="icon icon-circle-arrow-right"></span></a>
+ <a href="../autometrics/" class="btn btn-neutral float-right" title="AutoMetric API">Next <span class="icon icon-circle-arrow-right"></span></a>
<a href="../application_packages/" class="btn btn-neutral" title="Packages"><span class="icon icon-circle-arrow-left"></span> Previous</a>
diff --git a/docs/apex-3.3/search.html b/docs/apex-3.3/search.html
index 85bdec1..aaad72b 100644
--- a/docs/apex-3.3/search.html
+++ b/docs/apex-3.3/search.html
@@ -3,8 +3,11 @@
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
+ <meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
+
+
<title>Apache Apex Documentation</title>
@@ -152,7 +155,7 @@
</form>
<div id="mkdocs-search-results">
- Sorry, page not found.
+ Searching...
</div>
diff --git a/docs/apex-3.3/searchbox.html b/docs/apex-3.3/searchbox.html
new file mode 100644
index 0000000..177fcb3
--- /dev/null
+++ b/docs/apex-3.3/searchbox.html
@@ -0,0 +1,5 @@
+<div role="search">
+ <form id ="rtd-search-form" class="wy-form" action="{{ base_url }}/search.html" method="get">
+ <input type="text" name="q" placeholder="Search docs" />
+ </form>
+</div>
diff --git a/docs/apex-3.3/sitemap.xml b/docs/apex-3.3/sitemap.xml
index 14f9650..edd34f3 100644
--- a/docs/apex-3.3/sitemap.xml
+++ b/docs/apex-3.3/sitemap.xml
@@ -3,8 +3,8 @@
<url>
- <loc>None/</loc>
- <lastmod>2016-03-04</lastmod>
+ <loc>/</loc>
+ <lastmod>2016-03-09</lastmod>
<changefreq>daily</changefreq>
</url>
@@ -12,32 +12,32 @@
<url>
- <loc>None/apex_development_setup/</loc>
- <lastmod>2016-03-04</lastmod>
+ <loc>/apex_development_setup/</loc>
+ <lastmod>2016-03-09</lastmod>
<changefreq>daily</changefreq>
</url>
<url>
- <loc>None/application_development/</loc>
- <lastmod>2016-03-04</lastmod>
+ <loc>/application_development/</loc>
+ <lastmod>2016-03-09</lastmod>
<changefreq>daily</changefreq>
</url>
<url>
- <loc>None/application_packages/</loc>
- <lastmod>2016-03-04</lastmod>
+ <loc>/application_packages/</loc>
+ <lastmod>2016-03-09</lastmod>
<changefreq>daily</changefreq>
</url>
<url>
- <loc>None/operator_development/</loc>
- <lastmod>2016-03-04</lastmod>
+ <loc>/operator_development/</loc>
+ <lastmod>2016-03-09</lastmod>
<changefreq>daily</changefreq>
</url>
<url>
- <loc>None/autometrics/</loc>
- <lastmod>2016-03-04</lastmod>
+ <loc>/autometrics/</loc>
+ <lastmod>2016-03-09</lastmod>
<changefreq>daily</changefreq>
</url>
@@ -46,8 +46,8 @@
<url>
- <loc>None/dtcli/</loc>
- <lastmod>2016-03-04</lastmod>
+ <loc>/dtcli/</loc>
+ <lastmod>2016-03-09</lastmod>
<changefreq>daily</changefreq>
</url>
diff --git a/docs/apex-3.3/toc.html b/docs/apex-3.3/toc.html
new file mode 100644
index 0000000..6cd2fc9
--- /dev/null
+++ b/docs/apex-3.3/toc.html
@@ -0,0 +1,23 @@
+{% if nav_item.children %}
+ <ul class="subnav">
+ <li><span>{{ nav_item.title }}</span></li>
+
+ {% for nav_item in nav_item.children %}
+ {% include 'toc.html' %}
+ {% endfor %}
+ </ul>
+{% else %}
+ <li class="toctree-l1 {% if nav_item.active%}current{%endif%}">
+ <a class="{% if nav_item.active%}current{%endif%}" href="{{ nav_item.url }}">{{ nav_item.title }}</a>
+ {% if nav_item == current_page %}
+ <ul>
+ {% for toc_item in toc %}
+ <li class="toctree-l3"><a href="{{ toc_item.url }}">{{ toc_item.title }}</a></li>
+ {% for toc_item in toc_item.children %}
+ <li><a class="toctree-l4" href="{{ toc_item.url }}">{{ toc_item.title }}</a></li>
+ {% endfor %}
+ {% endfor %}
+ </ul>
+ {% endif %}
+ </li>
+{% endif %}
diff --git a/docs/apex-3.3/versions.html b/docs/apex-3.3/versions.html
new file mode 100644
index 0000000..d12d197
--- /dev/null
+++ b/docs/apex-3.3/versions.html
@@ -0,0 +1,15 @@
+<div class="rst-versions" role="note" style="cursor: pointer">
+ <span class="rst-current-version" data-toggle="rst-current-version">
+ {% if repo_name == 'GitHub' %}
+ <a href="{{ repo_url }}" class="icon icon-github" style="float: left; color: #fcfcfc"> GitHub</a>
+ {% elif repo_name == 'Bitbucket' %}
+ <a href="{{ repo_url }}" class="icon icon-bitbucket" style="float: left; color: #fcfcfc"> BitBucket</a>
+ {% endif %}
+ {% if previous_page %}
+ <span><a href="{{ previous_page.url }}" style="color: #fcfcfc;">« Previous</a></span>
+ {% endif %}
+ {% if next_page %}
+ <span style="margin-left: 15px"><a href="{{ next_page.url }}" style="color: #fcfcfc">Next »</a></span>
+ {% endif %}
+ </span>
+</div>
