http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/jdbcPollInputOperator/index.html ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/jdbcPollInputOperator/index.html b/content/docs/malhar-3.7/operators/jdbcPollInputOperator/index.html new file mode 100644 index 0000000..d428f4f --- /dev/null +++ b/content/docs/malhar-3.7/operators/jdbcPollInputOperator/index.html @@ -0,0 +1,678 @@ +<!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"> + + + + <title>JDBC Poller Input - Apache Apex Malhar Documentation</title> + + + <link rel="shortcut icon" href="../../favicon.ico"> + + + + <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="../../css/theme.css" type="text/css" /> + <link rel="stylesheet" href="../../css/theme_extra.css" type="text/css" /> + <link rel="stylesheet" href="../../css/highlight.css"> + + + <script> + // Current page data + var mkdocs_page_name = "JDBC Poller Input"; + var mkdocs_page_input_path = "operators/jdbcPollInputOperator.md"; + var mkdocs_page_url = "/operators/jdbcPollInputOperator/"; + </script> + + <script src="../../js/jquery-2.1.1.min.js"></script> + <script src="../../js/modernizr-2.8.3.min.js"></script> + <script type="text/javascript" src="../../js/highlight.pack.js"></script> + <script src="../../js/theme.js"></script> + + +</head> + +<body class="wy-body-for-nav" role="document"> + + <div class="wy-grid-for-nav"> + + + <nav data-toggle="wy-nav-shift" class="wy-nav-side stickynav"> + <div class="wy-side-nav-search"> + <a href="../.." class="icon icon-home"> Apache Apex Malhar Documentation</a> + <div role="search"> + <form id ="rtd-search-form" class="wy-form" action="../../search.html" method="get"> + <input type="text" name="q" placeholder="Search docs" /> + </form> +</div> + </div> + + <div class="wy-menu wy-menu-vertical" data-spy="affix" role="navigation" aria-label="main navigation"> + <ul class="current"> + + <li> + <li class="toctree-l1 "> + <a class="" href="../..">Apache Apex Malhar</a> + + </li> +<li> + + <li> + <ul class="subnav"> + <li><span>APIs</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../../apis/calcite/">SQL</a> + + </li> + + + </ul> +<li> + + <li> + <ul class="subnav"> + <li><span>Operators</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../block_reader/">Block Reader</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">CSV Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvParserOperator/">CSV Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../deduper/">Deduper</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../enricher/">Enricher</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../fsInputOperator/">File Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_output/">File Output</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_splitter/">File Splitter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../filter/">Filter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../AbstractJdbcTransactionableOutputOperator/">Jdbc Output Operator</a> + + </li> + + + + <li class="toctree-l1 current"> + <a class="current" href="./">JDBC Poller Input</a> + + <ul> + + <li class="toctree-l3"><a href="#jdbc-poller-input-operator">JDBC Poller Input Operator</a></li> + + <li><a class="toctree-l4" href="#operator-objective">Operator Objective</a></li> + + <li><a class="toctree-l4" href="#use-cases">Use cases</a></li> + + <li><a class="toctree-l4" href="#how-to-use">How to Use?</a></li> + + <li><a class="toctree-l4" href="#partitioning-of-jdbc-poller">Partitioning of JDBC Poller</a></li> + + <li><a class="toctree-l4" href="#operator-information">Operator Information</a></li> + + <li><a class="toctree-l4" href="#abstractjdbcpollinputoperator">AbstractJdbcPollInputOperator</a></li> + + <li><a class="toctree-l4" href="#abstract-methods">Abstract Methods</a></li> + + + <li class="toctree-l3"><a href="#concrete-classes">Concrete Classes</a></li> + + <li><a class="toctree-l4" href="#1-jdbcpojopollinputoperator">1. JdbcPOJOPollInputOperator</a></li> + + <li><a class="toctree-l4" href="#2-jdbcpollinputoperator">2. JdbcPollInputOperator</a></li> + + + <li class="toctree-l3"><a href="#limitations">Limitations</a></li> + + + </ul> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jmsInputOperator/">JMS Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonFormatter/">JSON Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonParser/">JSON Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../kafkaInputOperator/">Kafka Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transformer</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../windowedOperator/">Windowed Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../xmlParserOperator/">XML Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonParser/">Json Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonFormatter/">Json Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transform Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">Csv Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../s3outputmodule/">S3 Output Module</a> + + </li> + + + </ul> +<li> + + </ul> + </div> + + </nav> + + <section data-toggle="wy-nav-shift" class="wy-nav-content-wrap"> + + + <nav class="wy-nav-top" role="navigation" aria-label="top navigation"> + <i data-toggle="wy-nav-top" class="fa fa-bars"></i> + <a href="../..">Apache Apex Malhar Documentation</a> + </nav> + + + <div class="wy-nav-content"> + <div class="rst-content"> + <div role="navigation" aria-label="breadcrumbs navigation"> + <ul class="wy-breadcrumbs"> + <li><a href="../..">Docs</a> »</li> + + + + <li>Operators »</li> + + + + <li>JDBC Poller Input</li> + <li class="wy-breadcrumbs-aside"> + + </li> + </ul> + <hr/> +</div> + <div role="main"> + <div class="section"> + + <h1 id="jdbc-poller-input-operator">JDBC Poller Input Operator</h1> +<h2 id="operator-objective">Operator Objective</h2> +<p>This operator scans JDBC database table in parallel fashion. This operator is added to address common input operator problems like,</p> +<ol> +<li> +<p>As discussed in <a href="https://github.com/apache/apex-core/blob/master/docs/development_best_practices.md";>Development Best Practices</a>, + the operator callbacks such as <code>beginWindow()</code>, <code>endWindow()</code>, <code>emitTuples()</code>, etc. + (which are invoked by the main operator thread) + are required to return quickly, well within the default streaming window duration of + 500ms. This requirement can be an issue when retrieving data from slow external systems + such as databases or object stores: if the call takes too long, the platform will deem + the operator blocked and restart it. Restarting will often run into the same issue + causing an unbroken sequence of restarts.</p> +</li> +<li> +<p>When a large volume of data is available from a single store that allows reading from + arbitrary locations (such as a file or a database table), reading the data sequentially + can be throughput limiting: Having multiple readers read from non-overlapping sections + of the store allows any downstream parallelism in the DAG to be exploited better to + enhance throughput. For files, this approach is used by the file splitter and block + reader operators in the Malhar library.</p> +</li> +</ol> +<p>JDBC Poller Input operator addresses the first issue with an asynchronous worker thread which retrieves the data and adds it to an in-memory queue; the main operator thread dequeue tuples very quickly if data is available or simply returns if not. The second is addressed in a way that parallels the approach to files by having multiple partitions read records from non-overlapping areas of the table. Additional details of how this is done are described below.</p> +<h4 id="assumption">Assumption</h4> +<p>Assumption is that there is an ordered column using which range queries can be formed. That means database has a column or combination of columns which has unique constraint as well as every newly inserted record should have column value more than max value in that column, as we poll only appended records.</p> +<h2 id="use-cases">Use cases</h2> +<ol> +<li>Scan huge database tables to either copy to other database or process it using <strong>Apache Apex</strong>. An example application using this operator to copy database contents to HDFS is available in the <a href="https://github.com/DataTorrent/examples/tree/master/tutorials/jdbcIngest";>examples repository</a>. Look for "PollJdbcToHDFSApp" for example of this particular operator.</li> +</ol> +<h2 id="how-to-use">How to Use?</h2> +<p>The tuple type in the abstract class is a generic parameter. Concrete subclasses need to choose an appropriate class (such as String or an appropriate concrete java class, having no-argument constructor so that it can be serialized using kyro). Also implement a couple of abstract methods: <code>getTuple(ResultSet)</code> to convert database rows to objects of concrete class and <code>emitTuple(T)</code> to emit the tuple.</p> +<p>In principle, no ports need be defined in the rare case that the operator simply writes tuples to some external sink or merely maintains aggregated statistics. But in most common scenarios, the tuples need to be sent to one or more downstream operators for additional processing such as parsing, enrichment or aggregation; in such cases, appropriate output ports are defined and the emitTuple(T) implementation dispatches tuples to the desired output ports.</p> +<p>Couple of concrete implementations are provided in Malhar:</p> +<ol> +<li> +<p><a href="https://github.com/apache/apex-malhar/blob/master/library/src/main/java/com/datatorrent/lib/db/jdbc/JdbcPOJOPollInputOperator.java";>JdbcPOJOPollInputOperator</a>: It uses java Object for the generic parameter. This operator defines a single output port and processes each database table record one by one as a tuple object. You need to set the output port attribute TUPLE_CLASS to define your <a href="https://en.wikipedia.org/wiki/Plain_Old_Java_Object";>POJO</a> class name to define Object type. The record fetched from the database table will be parsed, using the <code>getTuple</code> method mentioned above, as an object of the configured class. Details are discussed below.</p> +</li> +<li> +<p><a href="https://github.com/apache/apex-malhar/blob/master/library/src/main/java/com/datatorrent/lib/db/jdbc/JdbcPollInputOperator.java";>JdbcPollInputOperator</a>: It uses String for the generic parameter. This operator defines a single port and processes each database table record one by one as String tuple. Details are discussed below.</p> +</li> +</ol> +<h2 id="partitioning-of-jdbc-poller">Partitioning of JDBC Poller</h2> +<h4 id="static-partitioning">Static Partitioning</h4> +<p>Only static partitioning is supported for JDBC Poller Input Operator. Configure parameter <code>partitionCount</code> to define the desired number of initial partitions (4 in this example). +<strong>Note</strong>: An additional partition will be created to poll newly added records, so the total number of partitions will always be 1 + partitionCount.</p> +<pre><code class="xml"> <property> + <name>dt.operator.{OperatorName}.prop.partitionCount</name> + <value>4</value> + </property> +</code></pre> + +<p>where {OperatorName} is the name of the JDBC Poller operator.</p> +<p>This will create 5 operator instances in all. Four of these will fetch the data which is currently in the table. We call these static non-polling partitions. The partitions will be idle after they fetch the portion of the data. An additional partition will be created which will read any newly added data. We call such a partition as a polling partition, as it "polls" for newly added data. There will be only one polling partition.</p> +<h4 id="dynamic-partitioning">Dynamic Partitioning</h4> +<p>Not supported.</p> +<h2 id="operator-information">Operator Information</h2> +<ol> +<li>Operator location: <strong><em>malhar-library</em></strong></li> +<li>Available since: <strong><em>3.5.0</em></strong></li> +<li>Operator state: <strong><em>Evolving</em></strong></li> +<li>Java Packages:<ul> +<li>Operator: <strong><em><a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/db/jdbc/AbstractJdbcPollInputOperator.html";>com.datatorrent.lib.db.jdbc.AbstractJdbcPollInputOperator</a></em></strong></li> +</ul> +</li> +</ol> +<p>JDBC Poller is <strong>idempotent</strong>, <strong>fault-tolerant</strong> and <strong>statically partitionable</strong>.</p> +<h2 id="abstractjdbcpollinputoperator">AbstractJdbcPollInputOperator</h2> +<p>This is the abstract implementation that serves as base class for polling messages from JDBC store. It can be extended to modify functionality or add new capabilities. This class doesnât have any ports, so concrete subclasses will need to provide them if necessary.</p> +<p><img alt="AbstractJdbcPollInputOperator.png" src="../images/jdbcinput/operatorsClassDiagram.png" /></p> +<h3 id="properties-of-abstractjdbcpollinputoperator"><a name="AbstractJdbcPollInputOperatorProps"></a>Properties of AbstractJdbcPollInputOperator</h3> +<p>Several properties are available to configure the behavior of this operator and they are summarized in the table below.</p> +<table> +<thead> +<tr> +<th><strong>Property</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +<th><strong>Default Value</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>store</em></td> +<td>JDBC Store for connection</td> +<td><a href="#JDBCStore">JDBCStore</a></td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>tableName</em></td> +<td>table name to be scanned</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>columnsExpression</em></td> +<td>Comma separated list of columns to select from the given table.</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>key</em></td> +<td>Primary key column name</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>partitionCount</em></td> +<td>Static partitions count</td> +<td>int</td> +<td>No</td> +<td>1</td> +</tr> +<tr> +<td><em>whereCondition</em></td> +<td>Where condition for JDBC query</td> +<td>String</td> +<td>No</td> +<td>N/A</td> +</tr> +<tr> +<td><em>fetchSize</em></td> +<td>Hint limiting the number of rows to fetch in a single call</td> +<td>int</td> +<td>No</td> +<td>20000</td> +</tr> +<tr> +<td><em>pollInterval</em></td> +<td>Interval in milliseconds to poll the database table</td> +<td>int</td> +<td>No</td> +<td>10000</td> +</tr> +<tr> +<td><em>queueCapacity</em></td> +<td>Capacity of queue which holds DB data before emiting</td> +<td>int</td> +<td>No</td> +<td>4096</td> +</tr> +<tr> +<td><em>batchSize</em></td> +<td>Maximum number of tuples to emit in a single call to the <code>emitTuples()</code> callback (see explanation above).</td> +<td>int</td> +<td>No</td> +<td>2000</td> +</tr> +</tbody> +</table> +<h4 id="properties-of-jdbc-store-backendstore"><a name="JDBCStore"></a>Properties of JDBC Store (BackendStore)</h4> +<table> +<thead> +<tr> +<th><strong>Property</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +<th><strong>Default Value</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>databaseDriver</em></td> +<td>JDBC Driver class for connection to JDBC Store. This driver should be there in class path</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>databaseUrl</em></td> +<td>Database url of the form jdbc:subprotocol:subname</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>connectionProps</em></td> +<td>Comma separated connection properties e.g. user:xyz,password:ijk</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +</tbody> +</table> +<p>Of these only <code>store</code> properties, <code>tableName</code>, <code>columnsExpression</code> and <code>key</code> are mandatory. Those properties can be set like this:</p> +<pre><code class="xml"><property> + <name>dt.operator.{OperatorName}.prop.tableName</name> + <value>mytable</value> +</property> +<property> + <name>dt.operator.{OperatorName}.prop.columnsExpression</name> + <value>column1,column2,column4</value> +</property> +<property> + <name>dt.operator.{OperatorName}.prop.key</name> + <value>keycolumn</value> +</property> +<property> + <name>dt.operator.{OperatorName}.prop.store.databaseDriver</name> + <value>com.mysql.jdbc.Driver</value> +</property> +<property> + <name>dt.operator.{OperatorName}.prop.store.databaseUrl</name> + <value>jdbc:mysql://localhost:3306/mydb</value> +</property> +<property> + <name>dt.operator.{OperatorName}.prop.store.connectionProps</name> + <value>user:myuser,password:mypassword</value> +</property> +</code></pre> + +<ul> +<li>If you need to filter the table records, set <code>whereCondition</code> which will be added to the generated SQL query.</li> +<li>If you have a table with a very large number or rows to scan set <code>partitionsCount</code> to a higher number to increase read parallelism.</li> +<li>The operator uses PreparedStatement, a precompiled SQL statement to fetch records from database table. You can set <code>fetchSize</code> as a hint to the database driver to restrict number of rows to fetch in one call. The remaining rows will be fetched in subsequent calls. Please note, some of the database drivers may not honor this hint. Please refer to database driver documentation to know recommended value.</li> +<li>The platform invokes the <code>emitTuples()</code> callback multiple time in each streaming window; within a single such call, if a large number of tuples are emitted, there is some risk that they may overwhelm the downstream operators especially if they are performing some compute intensive operation. For such cases, output can be throttled by reducing the value of the <code>batchSize</code> property. Conversely, if the downstream operators can handle the load, increase the value to enhance throughput.</li> +<li>If there is high rate of incoming records in your table and you want to process them as soon as they appear, use lower value of <code>pollInterval</code>; if they appear rarely or if some delay in processing new records is acceptable, increase it.</li> +<li>After reading the records from the table they are held in memory for some time till they are emitted to next operator. The records are kept in a blocking queue. The capacity of this blocking queue can be changed using parameter <code>queueCapacity</code>. You can use larger size of queue when your reader thread is very fast and you want to read more data in memory to keep it ready for emission.</li> +</ul> +<p><strong>Note</strong>: Please set right store object instance to JDBC Poller Input Operator using your application code. It's recommended to use <a href="https://github.com/apache/apex-malhar/blob/master/library/src/main/java/com/datatorrent/lib/db/jdbc/JdbcStore.java";>JdbcStore</a> for this operator.</p> +<h3 id="abstract-methods">Abstract Methods</h3> +<p><code>void emitTuple(T tuple)</code>: Abstract method that emits tuple extracted from JDBC store.</p> +<p><code>T getTuple(ResultSet result)</code>: Abstract method to extract the tuple from the JDBC ResultSet object and convert it to the required type (T).</p> +<h2 id="concrete-classes">Concrete Classes</h2> +<h3 id="1-jdbcpojopollinputoperator">1. JdbcPOJOPollInputOperator</h3> +<p>This implementation converts JDBC store records to <a href="https://en.wikipedia.org/wiki/Plain_Old_Java_Object";>POJO</a> and emits POJO on output port.</p> +<h4 id="properties-of-jdbcpojopollinputoperator"><a name="JdbcPOJOPollInputOperatorProps"></a>Properties of JdbcPOJOPollInputOperator</h4> +<p>This operator defines following additional properties beyond those defined in the <a href="#AbstractJdbcPollInputOperatorProps">parent class</a>.</p> +<table> +<thead> +<tr> +<th><strong>Property</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +<th><strong>Default Value</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>fieldInfos</em></td> +<td><a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/util/FieldInfo.html";>FieldInfo</a> maps a store column to a POJO field name.</td> +<td>List</td> +<td>Yes</td> +<td>N/A</td> +</tr> +</tbody> +</table> +<h4 id="platform-attributes-that-influence-operator-behavior">Platform Attributes that influence operator behavior</h4> +<table> +<thead> +<tr> +<th><strong>Attribute</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>outputPort.TUPLE_CLASS</em></td> +<td>TUPLE_CLASS attribute on output port which tells operator the class of POJO which need to be emitted</td> +<td>Class or FQCN (Fully Qualified Class Name)</td> +<td>Yes</td> +</tr> +</tbody> +</table> +<h4 id="ports">Ports</h4> +<table> +<thead> +<tr> +<th><strong>Port</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>outputPort</em></td> +<td>Tuples that are read from JDBC store are emitted from on this port</td> +<td>Object (POJO)</td> +<td>No</td> +</tr> +</tbody> +</table> +<h3 id="2-jdbcpollinputoperator">2. JdbcPollInputOperator</h3> +<p>This implementation converts JDBC store records to comma separated CSV records. This operator is normally used when you just want to copy the data from database to somewhere else and don't want to do much of processing.</p> +<h4 id="properties-of-jdbcpollinputoperator"><a name="props"></a>Properties of JdbcPollInputOperator</h4> +<p>This operator defines no additional properties beyond those defined in the <a href="#AbstractJdbcPollInputOperatorProps">parent class</a>.</p> +<h4 id="ports_1">Ports</h4> +<table> +<thead> +<tr> +<th><strong>Port</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>outputPort</em></td> +<td>Tuples that are read from JDBC store are emitted on this port</td> +<td>String</td> +<td>No</td> +</tr> +</tbody> +</table> +<h2 id="limitations">Limitations</h2> +<p>Out of order insertion/deletion won't be supported.</p> + + </div> + </div> + <footer> + + <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation"> + + <a href="../jmsInputOperator/" class="btn btn-neutral float-right" title="JMS Input">Next <span class="icon icon-circle-arrow-right"></span></a> + + + <a href="../AbstractJdbcTransactionableOutputOperator/" class="btn btn-neutral" title="Jdbc Output Operator"><span class="icon icon-circle-arrow-left"></span> Previous</a> + + </div> + + + <hr/> + + <div role="contentinfo"> + <!-- Copyright etc --> + + </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> + + </div> + </div> + + </section> + + </div> + +<div class="rst-versions" role="note" style="cursor: pointer"> + <span class="rst-current-version" data-toggle="rst-current-version"> + + + <span><a href="../AbstractJdbcTransactionableOutputOperator/" style="color: #fcfcfc;">« Previous</a></span> + + + <span style="margin-left: 15px"><a href="../jmsInputOperator/" style="color: #fcfcfc">Next »</a></span> + + </span> +</div> + +</body> +</html>
http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/jmsInputOperator/index.html ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/jmsInputOperator/index.html b/content/docs/malhar-3.7/operators/jmsInputOperator/index.html new file mode 100644 index 0000000..b3bfb4d --- /dev/null +++ b/content/docs/malhar-3.7/operators/jmsInputOperator/index.html @@ -0,0 +1,539 @@ +<!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"> + + + + <title>JMS Input - Apache Apex Malhar Documentation</title> + + + <link rel="shortcut icon" href="../../favicon.ico"> + + + + <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="../../css/theme.css" type="text/css" /> + <link rel="stylesheet" href="../../css/theme_extra.css" type="text/css" /> + <link rel="stylesheet" href="../../css/highlight.css"> + + + <script> + // Current page data + var mkdocs_page_name = "JMS Input"; + var mkdocs_page_input_path = "operators/jmsInputOperator.md"; + var mkdocs_page_url = "/operators/jmsInputOperator/"; + </script> + + <script src="../../js/jquery-2.1.1.min.js"></script> + <script src="../../js/modernizr-2.8.3.min.js"></script> + <script type="text/javascript" src="../../js/highlight.pack.js"></script> + <script src="../../js/theme.js"></script> + + +</head> + +<body class="wy-body-for-nav" role="document"> + + <div class="wy-grid-for-nav"> + + + <nav data-toggle="wy-nav-shift" class="wy-nav-side stickynav"> + <div class="wy-side-nav-search"> + <a href="../.." class="icon icon-home"> Apache Apex Malhar Documentation</a> + <div role="search"> + <form id ="rtd-search-form" class="wy-form" action="../../search.html" method="get"> + <input type="text" name="q" placeholder="Search docs" /> + </form> +</div> + </div> + + <div class="wy-menu wy-menu-vertical" data-spy="affix" role="navigation" aria-label="main navigation"> + <ul class="current"> + + <li> + <li class="toctree-l1 "> + <a class="" href="../..">Apache Apex Malhar</a> + + </li> +<li> + + <li> + <ul class="subnav"> + <li><span>APIs</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../../apis/calcite/">SQL</a> + + </li> + + + </ul> +<li> + + <li> + <ul class="subnav"> + <li><span>Operators</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../block_reader/">Block Reader</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">CSV Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvParserOperator/">CSV Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../deduper/">Deduper</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../enricher/">Enricher</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../fsInputOperator/">File Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_output/">File Output</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_splitter/">File Splitter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../filter/">Filter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../AbstractJdbcTransactionableOutputOperator/">Jdbc Output Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jdbcPollInputOperator/">JDBC Poller Input</a> + + </li> + + + + <li class="toctree-l1 current"> + <a class="current" href="./">JMS Input</a> + + <ul> + + <li class="toctree-l3"><a href="#jms-input-operator">JMS INPUT OPERATOR</a></li> + + <li><a class="toctree-l4" href="#introduction-about-the-jms-input-operator">Introduction: About the JMS Input Operator</a></li> + + <li><a class="toctree-l4" href="#why-is-it-needed">Why is it needed ?</a></li> + + <li><a class="toctree-l4" href="#jmsbase">JMSBase</a></li> + + <li><a class="toctree-l4" href="#abstractjmsinputoperator">AbstractJMSInputOperator</a></li> + + <li><a class="toctree-l4" href="#concrete-classes">Concrete Classes</a></li> + + <li><a class="toctree-l4" href="#application-examples">Application Examples</a></li> + + + </ul> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonFormatter/">JSON Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonParser/">JSON Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../kafkaInputOperator/">Kafka Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transformer</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../windowedOperator/">Windowed Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../xmlParserOperator/">XML Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonParser/">Json Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonFormatter/">Json Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transform Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">Csv Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../s3outputmodule/">S3 Output Module</a> + + </li> + + + </ul> +<li> + + </ul> + </div> + + </nav> + + <section data-toggle="wy-nav-shift" class="wy-nav-content-wrap"> + + + <nav class="wy-nav-top" role="navigation" aria-label="top navigation"> + <i data-toggle="wy-nav-top" class="fa fa-bars"></i> + <a href="../..">Apache Apex Malhar Documentation</a> + </nav> + + + <div class="wy-nav-content"> + <div class="rst-content"> + <div role="navigation" aria-label="breadcrumbs navigation"> + <ul class="wy-breadcrumbs"> + <li><a href="../..">Docs</a> »</li> + + + + <li>Operators »</li> + + + + <li>JMS Input</li> + <li class="wy-breadcrumbs-aside"> + + </li> + </ul> + <hr/> +</div> + <div role="main"> + <div class="section"> + + <h1 id="jms-input-operator">JMS INPUT OPERATOR</h1> +<h3 id="introduction-about-the-jms-input-operator">Introduction: About the JMS Input Operator</h3> +<p>The JMS input operator consumes data from a messaging system using the JMS client API. JMS not being a communication protocol, the operator needs an underlying JMS client API library to talk to a messaging system. Currently the operator has been tested with the Amazon SQS and Apache ActiveMQ System brokers via their respective JMS client API libraries.</p> +<h3 id="why-is-it-needed">Why is it needed ?</h3> +<p>You will need the operator to read data from a messaging system (e.g. Apache ActiveMQ) via the JMS client API. The operator supports both the publish-subscribe (topics) and point-to-point (queues) modes. The operator currently does not support partitioning and dynamic scalability.</p> +<h3 id="jmsbase">JMSBase</h3> +<p>This class encapsulates various JMS properties and behaviors and maintains connections with the JMS broker. This is the base class for JMS input and output adaptor operators. Operators should not directly subclass JMSBase but one of the JMS input or output operators.</p> +<h3 id="abstractjmsinputoperator">AbstractJMSInputOperator</h3> +<p>This abstract implementation serves as the base class for consuming generic messages from an external messaging system. Concrete subclasses implement conversion and emit methods to emit tuples for a concrete type. JMSStringInputOperator is one such subclass in the library used for String messages. JMSObjectInputOperator is another one used for multiple message types where the user has the ability to get String, byte array, Map or POJO messages on the respective output ports.</p> +<h4 id="configuration-parameters">Configuration Parameters</h4> +<p>Common configuration parameters are described here. +<table> +<col width="25%" /> +<col width="75%" /> +<tbody> +<tr class="odd"> +<td align="left"><p>Parameter</p></td> +<td align="left"><p>Description</p></td> +</tr> +<tr class="even"> +<td align="left"><p>windowDataManager</p></td> +<td align="left"><p>This is an instance of <code>WindowDataManager</code> that implements idempotency. Idempotency ensures that an operator will process the same set of messages in a window before and after a failure. For example, say the operator completed window 10 and failed before or during window 11. If the operator gets restored at window 10, it will replay the messages of window 10 which were saved from the previous run before the failure. Although important, idempotency comes at a price because an operator needs to persist some state at the end of each window. Default Value = <code>org.apache.apex.malhar.lib.wal.FSWindowDataManager</code></p></td> +</tr> +<tr class="odd"> +<td align="left"><p>connectionFactoryBuilder</p></td> +<td align="left"><p>The operator uses the builder pattern that requires the user to specify an instance of <code>com.datatorrent.lib.io.jms.JMSBase.ConnectionFactoryBuilder</code>. This builder creates the connection factory that encapsulates the underlying JMS client API library (e.g. ActiveMQ or Amazon SQS). By default the operator uses <code>com.datatorrent.lib.io.jms.JMSBase.DefaultConnectionFactoryBuilder</code> which is used for ActiveMQ. One of the examples below describes the Amazon SQS use-case. </td> +</tr> +</tbody> +</table></p> +<h4 id="abstract-methods">Abstract Methods</h4> +<p>The following abstract methods need to be implemented by concrete subclasses.</p> +<p>T convert(Message message): This method converts a JMS Message object to type T.</p> +<p>void emit(T payload): This method emits a tuple given the payload extracted from a JMS message.</p> +<h3 id="concrete-classes">Concrete Classes</h3> +<ol> +<li> +<p>JMSStringInputOperator : +This class extends AbstractJMSInputOperator to deliver String payloads in the tuple.</p> +</li> +<li> +<p>JMSObjectInputOperator: +This class extends AbstractJMSInputOperator to deliver String, byte array, Map or POJO payloads in the tuple.</p> +</li> +</ol> +<h3 id="application-examples">Application Examples</h3> +<h4 id="activemq-example">ActiveMQ Example</h4> +<p>The source code for the tutorial can be found here:</p> +<p><a href="https://github.com/DataTorrent/examples/tree/master/tutorials/jmsActiveMQ";>https://github.com/DataTorrent/examples/tree/master/tutorials/jmsActiveMQ</a></p> +<p>The following code snippet from the example illustrates how the DAG is created:</p> +<pre><code class="java"> @Override + public void populateDAG(DAG dag, Configuration conf) + { + JMSStringInputOperator amqInput = dag.addOperator("amqIn", + new JMSStringInputOperator()); + + LineOutputOperator out = dag.addOperator("fileOut", new LineOutputOperator()); + + dag.addStream("data", amqInput.output, out.input); + } +</code></pre> + +<p>The DAG consists of only 2 operators: the <code>JMSStringInputOperator</code> which is the input operator that feeds received ActiveMQ messages into the output operator <code>LineOutputOperator</code> which outputs these messages into a file or files.</p> +<p>The default connectionFactoryBuilder supports ActiveMQ so there is no need to set this value. However the following ActiveMQ related values need to be set either from properties files or using the appropriate setter methods in the code:</p> +<table> +<col width="25%" /> +<col width="75%" /> +<tbody> +<tr class="odd"> +<td align="left"><p>Value</p></td> +<td align="left"><p>Description</p></td> +</tr> +<tr class="even"> +<td align="left"><p>connectionFactoryProperties</p></td> +<td align="left"><p>This is a Map of key and value strings and can be set directly from configuration as in the example above. The table below describes the most important properties.</p></td> +</tr> +<tr class="odd"> +<td align="left"><p>topic</p></td> +<td align="left"><p>This boolean value is set to true for the publish-subscribe case and false for the PTP (point-to-point) case.</p></td> +</tr> +<tr class="even"> +<td align="left"><p>subject</p></td> +<td align="left"><p>This is the queue name for PTP (point-to-point) use-case and topic name for the publish-subscribe use case.</p></td> +</tr> +<tr class="odd"> +<td align="left"><p>durable</p></td> +<td align="left"><p>This boolean value is set to true for durable subscriptionss, false otherwise. Durable subscriptions save messages to persistent storage until consumed. Used only when the clientId (see below) is set.</p></td> +</tr> +<tr class="even"> +<td align="left"><p>clientId</p></td> +<td align="left"><p>The client-ID for this ActiveMQ consumer in the durable subscription mode as described above.</p></td> +</tr> +<tr class="odd"> +<td align="left"><p>transacted</p></td> +<td align="left"><p>This boolean value is set to true for transacted JMS sessions as described in +<a href="https://docs.oracle.com/javaee/7/api/javax/jms/Session.html";>Session</a>.</p></td> +</tr> +<tr class="even"> +<td align="left"><p>ackMode</p></td> +<td align="left"><p>This string value sets the acknowledgement mode as described in +<a href="https://docs.oracle.com/javaee/7/api/javax/jms/Session.html#field.summary";>Session fields</a>.</p></td> +</tr> +</tbody> +</table> + +<p>The following table describes the string properties to be set in the map that is passed in the connectionFactoryProperties value described above.</p> +<table> +<col width="25%" /> +<col width="75%" /> +<tbody> +<tr class="odd"> +<td align="left"><p>Property Name</p></td> +<td align="left"><p>Description</p></td> +</tr> +<tr class="even"> +<td align="left"><p>brokerURL</p></td> +<td align="left"><p>The <a href="http://activemq.apache.org/configuring-transports.html";>connection URL</a> +used to connect to the ActiveMQ broker</p></td></tr> +<tr class="even"> +<td align="left"><p>userName</p></td> +<td align="left"><p>The JMS userName used by connections created by this factory (optional when anonymous access is used)</p></td> +</tr> +<tr class="even"> +<td align="left"><p>password</p></td> +<td align="left"><p>The JMS password used for connections created from this factory (optional when anonymous access is used)</p></td> +</tr> +</tbody> +</table> + +<p>These properties can be set from the properties.xml file as shown below +(from the example <a href="https://github.com/DataTorrent/examples/tree/master/tutorials/jmsActiveMQ";>https://github.com/DataTorrent/examples/tree/master/tutorials/jmsActiveMQ</a> ).</p> +<pre><code class="xml"><configuration> + <property> + <name>dt.operator.amqIn.prop.connectionFactoryProperties.brokerURL</name> + <value>vm://localhost</value> + </property> + <property> + <name>dt.operator.amqIn.prop.subject</name> + <value>jms4Amq</value> + </property> +</configuration> +</code></pre> + +<h4 id="sqs-example">SQS Example</h4> +<p>The source code for the tutorial can be found here:</p> +<p><a href="https://github.com/DataTorrent/examples/tree/master/tutorials/jmsSqs";>https://github.com/DataTorrent/examples/tree/master/tutorials/jmsSqs</a></p> +<p>The following code snippet from the example illustrates how the DAG is created:</p> +<pre><code class="java"> @Override + public void populateDAG(DAG dag, Configuration conf) + { + + JMSStringInputOperator sqsInput = dag.addOperator("sqsIn", + new JMSStringInputOperator()); + + MyConnectionFactoryBuilder factoryBuilder = new MyConnectionFactoryBuilder(); + + factoryBuilder.sqsDevCredsFilename = conf.get(SQSDEV_CREDS_FILENAME_PROPERTY); + + sqsInput.setConnectionFactoryBuilder(factoryBuilder); + + LineOutputOperator out = dag.addOperator("fileOut", new LineOutputOperator()); + + dag.addStream("data", sqsInput.output, out.input); + } +</code></pre> + +<p>The DAG consists of only 2 operators: the <code>JMSStringInputOperator</code> which is the input operator that feeds received SQS messages into the output operator <code>LineOutputOperator</code> which outputs these messages into a file or files. The code also shows how the AWS/SQS credentials are initialized in the factory builder. </p> +<p>For SQS you will have to provide a custom connectionFactoryBuilder as shown in the example above and in <a href="https://github.com/awslabs/amazon-sqs-java-messaging-lib/blob/master/src/main/java/com/amazon/sqs/javamessaging/SQSConnectionFactory.java";>SQSConnectionFactory.java</a>. The builder is typically used to supply AWS region and credential information that cannot be supplied via any JMS interfaces.</p> +<p>The following code snippet shows a typical Builder implementation that can be supplied to the operator. The AWS credentials are supplied via a <a href="http://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/PropertiesFileCredentialsProvider.html";>PropertiesFileCredentialsProvider</a> object in which sqsCredsFilename is the fully qualified path to a properties file from which the AWS security credentials are to be loaded. For example <code>/etc/somewhere/credentials.properties</code></p> +<pre><code class="java">static class MyConnectionFactoryBuilder implements JMSBase.ConnectionFactoryBuilder { + +String sqsCredsFilename; + +MyConnectionFactoryBuilder() +{ +} + +@Override +public ConnectionFactory buildConnectionFactory() +{ + // Create the connection factory using the properties file credential provider. + // Connections this factory creates can talk to the queues in us-east-1 region. + SQSConnectionFactory connectionFactory = + SQSConnectionFactory.builder() + .withRegion(Region.getRegion(Regions.US_EAST_1)) + .withAWSCredentialsProvider(new PropertiesFileCredentialsProvider(sqsCredsFilename)) + .build(); + return connectionFactory; + } +} +</code></pre> + + </div> + </div> + <footer> + + <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation"> + + <a href="../jsonFormatter/" class="btn btn-neutral float-right" title="JSON Formatter">Next <span class="icon icon-circle-arrow-right"></span></a> + + + <a href="../jdbcPollInputOperator/" class="btn btn-neutral" title="JDBC Poller Input"><span class="icon icon-circle-arrow-left"></span> Previous</a> + + </div> + + + <hr/> + + <div role="contentinfo"> + <!-- Copyright etc --> + + </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> + + </div> + </div> + + </section> + + </div> + +<div class="rst-versions" role="note" style="cursor: pointer"> + <span class="rst-current-version" data-toggle="rst-current-version"> + + + <span><a href="../jdbcPollInputOperator/" style="color: #fcfcfc;">« Previous</a></span> + + + <span style="margin-left: 15px"><a href="../jsonFormatter/" style="color: #fcfcfc">Next »</a></span> + + </span> +</div> + +</body> +</html> http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/jsonFormatter/index.html ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/jsonFormatter/index.html b/content/docs/malhar-3.7/operators/jsonFormatter/index.html new file mode 100644 index 0000000..8831017 --- /dev/null +++ b/content/docs/malhar-3.7/operators/jsonFormatter/index.html @@ -0,0 +1,485 @@ +<!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"> + + + + <title>Json Formatter - Apache Apex Malhar Documentation</title> + + + <link rel="shortcut icon" href="../../favicon.ico"> + + + + <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="../../css/theme.css" type="text/css" /> + <link rel="stylesheet" href="../../css/theme_extra.css" type="text/css" /> + <link rel="stylesheet" href="../../css/highlight.css"> + + + <script> + // Current page data + var mkdocs_page_name = "Json Formatter"; + var mkdocs_page_input_path = "operators/jsonFormatter.md"; + var mkdocs_page_url = "/operators/jsonFormatter/"; + </script> + + <script src="../../js/jquery-2.1.1.min.js"></script> + <script src="../../js/modernizr-2.8.3.min.js"></script> + <script type="text/javascript" src="../../js/highlight.pack.js"></script> + <script src="../../js/theme.js"></script> + + +</head> + +<body class="wy-body-for-nav" role="document"> + + <div class="wy-grid-for-nav"> + + + <nav data-toggle="wy-nav-shift" class="wy-nav-side stickynav"> + <div class="wy-side-nav-search"> + <a href="../.." class="icon icon-home"> Apache Apex Malhar Documentation</a> + <div role="search"> + <form id ="rtd-search-form" class="wy-form" action="../../search.html" method="get"> + <input type="text" name="q" placeholder="Search docs" /> + </form> +</div> + </div> + + <div class="wy-menu wy-menu-vertical" data-spy="affix" role="navigation" aria-label="main navigation"> + <ul class="current"> + + <li> + <li class="toctree-l1 "> + <a class="" href="../..">Apache Apex Malhar</a> + + </li> +<li> + + <li> + <ul class="subnav"> + <li><span>APIs</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../../apis/calcite/">SQL</a> + + </li> + + + </ul> +<li> + + <li> + <ul class="subnav"> + <li><span>Operators</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../block_reader/">Block Reader</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">CSV Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvParserOperator/">CSV Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../deduper/">Deduper</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../enricher/">Enricher</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../fsInputOperator/">File Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_output/">File Output</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_splitter/">File Splitter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../filter/">Filter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../AbstractJdbcTransactionableOutputOperator/">Jdbc Output Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jdbcPollInputOperator/">JDBC Poller Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jmsInputOperator/">JMS Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="./">JSON Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonParser/">JSON Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../kafkaInputOperator/">Kafka Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transformer</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../windowedOperator/">Windowed Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../xmlParserOperator/">XML Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonParser/">Json Parser</a> + + </li> + + + + <li class="toctree-l1 current"> + <a class="current" href="./">Json Formatter</a> + + <ul> + + <li class="toctree-l3"><a href="#json-formatter">Json Formatter</a></li> + + <li><a class="toctree-l4" href="#operator-objective">Operator Objective</a></li> + + <li><a class="toctree-l4" href="#class-diagram">Class Diagram</a></li> + + <li><a class="toctree-l4" href="#operator-information">Operator Information</a></li> + + <li><a class="toctree-l4" href="#properties-attributes-and-ports">Properties, Attributes and Ports</a></li> + + <li><a class="toctree-l4" href="#partitioning">Partitioning</a></li> + + <li><a class="toctree-l4" href="#example">Example</a></li> + + <li><a class="toctree-l4" href="#advance-features">Advance Features</a></li> + + + </ul> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transform Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">Csv Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../s3outputmodule/">S3 Output Module</a> + + </li> + + + </ul> +<li> + + </ul> + </div> + + </nav> + + <section data-toggle="wy-nav-shift" class="wy-nav-content-wrap"> + + + <nav class="wy-nav-top" role="navigation" aria-label="top navigation"> + <i data-toggle="wy-nav-top" class="fa fa-bars"></i> + <a href="../..">Apache Apex Malhar Documentation</a> + </nav> + + + <div class="wy-nav-content"> + <div class="rst-content"> + <div role="navigation" aria-label="breadcrumbs navigation"> + <ul class="wy-breadcrumbs"> + <li><a href="../..">Docs</a> »</li> + + + + <li>Operators »</li> + + + + <li>Json Formatter</li> + <li class="wy-breadcrumbs-aside"> + + </li> + </ul> + <hr/> +</div> + <div role="main"> + <div class="section"> + + <h1 id="json-formatter">Json Formatter</h1> +<h2 id="operator-objective">Operator Objective</h2> +<p>Purpose of JsonFormatter is to consume Plain Old Java Object ("POJO") and write them as JSON. +Json Formatter is <strong>idempotent</strong>, <strong>fault-tolerance</strong> & <strong>statically/dynamically partitionable</strong>.</p> +<h2 id="class-diagram">Class Diagram</h2> +<p><img alt="JsonFormatter" src="../images/jsonFormatter/JsonFormatter.png" /></p> +<h2 id="operator-information">Operator Information</h2> +<ol> +<li>Operator location:<strong>_malhar-library</strong></li> +<li>Available since:<strong><em>3.2.0</em></strong></li> +<li>Operator state:<strong><em>Evolving</em></strong></li> +<li>Java Package:<a href="https://github.com/apache/apex-malhar/blob/master/library/src/main/java/com/datatorrent/lib/formatter/JsonFormatter.java";>com.datatorrent.lib.formatter.JsonFormatter</a></li> +</ol> +<h2 id="properties-attributes-and-ports">Properties, Attributes and Ports</h2> +<h3 id="platform-attributes-that-influences-operator-behavior">Platform Attributes that influences operator behavior</h3> +<table> +<thead> +<tr> +<th><strong>Attribute</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>in.TUPLE_CLASS</em></td> +<td>TUPLE_CLASS attribute on input port which tells operator the class of incoming POJO</td> +<td>Class or FQCN</td> +<td>Yes</td> +</tr> +</tbody> +</table> +<h3 id="ports">Ports</h3> +<table> +<thead> +<tr> +<th><strong>Port</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>in</em></td> +<td>Tuples that needs to be formatted are recieved on this port</td> +<td>Object (POJO)</td> +<td>Yes</td> +</tr> +<tr> +<td><em>out</em></td> +<td>Valid Tuples that are emitted as JSON</td> +<td>String</td> +<td>No</td> +</tr> +<tr> +<td><em>err</em></td> +<td>Invalid Tuples are emitted on this port</td> +<td>Object</td> +<td>No</td> +</tr> +</tbody> +</table> +<h2 id="partitioning">Partitioning</h2> +<p>JSON Formatter is both statically and dynamically partitionable.</p> +<h3 id="static-partitioning">Static Partitioning</h3> +<p>This can be achieved in 2 ways</p> +<ol> +<li>Specifying the partitioner and number of partitions in the populateDAG() method</li> +</ol> +<pre><code class="java">JsonFormatter jsonFormatter = dag.addOperator("jsonFormatter", JsonFormatter.class); +StatelessPartitioner<JsonFormatter> partitioner1 = new StatelessPartitioner<JsonFormatter>(2); +dag.setAttribute(jsonFormatter, Context.OperatorContext.PARTITIONER, partitioner1 ); +</code></pre> + +<ol> +<li>Specifying the partitioner in properties file.</li> +</ol> +<pre><code class="xml"> <property> + <name>dt.operator.{OperatorName}.attr.PARTITIONER</name> + <value>com.datatorrent.common.partitioner.StatelessPartitioner:2</value> + </property> +</code></pre> + +<p>where {OperatorName} is the name of the JsonFormatter operator. + Above lines will partition JsonFormatter statically 2 times. Above value can be changed accordingly to change the number of static partitions.</p> +<h3 id="dynamic-paritioning">Dynamic Paritioning</h3> +<p>JsonFormatter can be dynamically partitioned using an out-of-the-box partitioner:</p> +<h4 id="throughput-based">Throughput based</h4> +<p>Following code can be added to populateDAG method of application to dynamically partition JsonFormatter:</p> +<pre><code class="java">JsonFormatter jsonFormatter = dag.addOperator("jsonFormatter", JsonFormatter.class); +StatelessThroughputBasedPartitioner<JsonFormatter> partitioner = new StatelessThroughputBasedPartitioner<>(); +partitioner.setCooldownMillis(conf.getLong(COOL_DOWN_MILLIS, 10000)); +partitioner.setMaximumEvents(conf.getLong(MAX_THROUGHPUT, 30000)); +partitioner.setMinimumEvents(conf.getLong(MIN_THROUGHPUT, 10000)); +dag.setAttribute(JsonFormatter, OperatorContext.STATS_LISTENERS, Arrays.asList(new StatsListener[]{partitioner})); +dag.setAttribute(JsonFormatter, OperatorContext.PARTITIONER, partitioner); +</code></pre> + +<p>Above code will dynamically partition JsonFormatter when the throughput changes. +If the overall throughput of JsonFormatter goes beyond 30000 or less than 10000, the platform will repartition JsonFormatter +to balance throughput of a single partition to be between 10000 and 30000. +CooldownMillis of 10000 will be used as the threshold time for which the throughput change is observed.</p> +<h2 id="example">Example</h2> +<p>Example for Json Formatter can be found at: <a href="https://github.com/DataTorrent/examples/tree/master/tutorials/parser";>https://github.com/DataTorrent/examples/tree/master/tutorials/parser</a></p> +<h2 id="advance-features">Advance Features</h2> +<p>JsonFormatter is based on <a href="https://github.com/FasterXML/jackson-databind";>jackson-databind</a> and so users can make use of <a href="https://github.com/FasterXML/jackson-annotations";>annotations</a> in POJO class. Here are few annotations that are relavant while using JsonFormatter +1. <strong>@JsonProperty</strong> : Sometimes POJOs contain properties that has different name from incoming POJOs.You can specify names as:</p> +<pre><code class="java">public class Ad{ + @JsonProperty("desc") + public String description; + public List<String> sizes; +} +</code></pre> + +<ol> +<li><strong>@JsonIgnore</strong> : Sometimes POJOs contain properties that you do not want to write out, so you can do:</li> +</ol> +<pre><code class="java">public class Value { + public int value; + @JsonIgnore + public int internalValue; +} +</code></pre> + +<ol> +<li><strong>@JsonFormat</strong> : Sometimes Date fields need to be printed in custom format, so you can do:</li> +</ol> +<pre><code class="java">public class Ad{ + @JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "EEE, d MMM yyyy HH:mm:ss") + public Date startDate; +} +</code></pre> + + </div> + </div> + <footer> + + <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation"> + + <a href="../transform/" class="btn btn-neutral float-right" title="Transform Operator">Next <span class="icon icon-circle-arrow-right"></span></a> + + + <a href="../jsonParser/" class="btn btn-neutral" title="Json Parser"><span class="icon icon-circle-arrow-left"></span> Previous</a> + + </div> + + + <hr/> + + <div role="contentinfo"> + <!-- Copyright etc --> + + </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> + + </div> + </div> + + </section> + + </div> + +<div class="rst-versions" role="note" style="cursor: pointer"> + <span class="rst-current-version" data-toggle="rst-current-version"> + + + <span><a href="../jsonParser/" style="color: #fcfcfc;">« Previous</a></span> + + + <span style="margin-left: 15px"><a href="../transform/" style="color: #fcfcfc">Next »</a></span> + + </span> +</div> + +</body> +</html> http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/jsonParser/index.html ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/jsonParser/index.html b/content/docs/malhar-3.7/operators/jsonParser/index.html new file mode 100644 index 0000000..9bd2099 --- /dev/null +++ b/content/docs/malhar-3.7/operators/jsonParser/index.html @@ -0,0 +1,482 @@ +<!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"> + + + + <title>Json Parser - Apache Apex Malhar Documentation</title> + + + <link rel="shortcut icon" href="../../favicon.ico"> + + + + <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="../../css/theme.css" type="text/css" /> + <link rel="stylesheet" href="../../css/theme_extra.css" type="text/css" /> + <link rel="stylesheet" href="../../css/highlight.css"> + + + <script> + // Current page data + var mkdocs_page_name = "Json Parser"; + var mkdocs_page_input_path = "operators/jsonParser.md"; + var mkdocs_page_url = "/operators/jsonParser/"; + </script> + + <script src="../../js/jquery-2.1.1.min.js"></script> + <script src="../../js/modernizr-2.8.3.min.js"></script> + <script type="text/javascript" src="../../js/highlight.pack.js"></script> + <script src="../../js/theme.js"></script> + + +</head> + +<body class="wy-body-for-nav" role="document"> + + <div class="wy-grid-for-nav"> + + + <nav data-toggle="wy-nav-shift" class="wy-nav-side stickynav"> + <div class="wy-side-nav-search"> + <a href="../.." class="icon icon-home"> Apache Apex Malhar Documentation</a> + <div role="search"> + <form id ="rtd-search-form" class="wy-form" action="../../search.html" method="get"> + <input type="text" name="q" placeholder="Search docs" /> + </form> +</div> + </div> + + <div class="wy-menu wy-menu-vertical" data-spy="affix" role="navigation" aria-label="main navigation"> + <ul class="current"> + + <li> + <li class="toctree-l1 "> + <a class="" href="../..">Apache Apex Malhar</a> + + </li> +<li> + + <li> + <ul class="subnav"> + <li><span>APIs</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../../apis/calcite/">SQL</a> + + </li> + + + </ul> +<li> + + <li> + <ul class="subnav"> + <li><span>Operators</span></li> + + + + <li class="toctree-l1 "> + <a class="" href="../block_reader/">Block Reader</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">CSV Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvParserOperator/">CSV Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../deduper/">Deduper</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../enricher/">Enricher</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../fsInputOperator/">File Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_output/">File Output</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../file_splitter/">File Splitter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../filter/">Filter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../AbstractJdbcTransactionableOutputOperator/">Jdbc Output Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jdbcPollInputOperator/">JDBC Poller Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jmsInputOperator/">JMS Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonFormatter/">JSON Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="./">JSON Parser</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../kafkaInputOperator/">Kafka Input</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transformer</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../windowedOperator/">Windowed Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../xmlParserOperator/">XML Parser</a> + + </li> + + + + <li class="toctree-l1 current"> + <a class="current" href="./">Json Parser</a> + + <ul> + + <li class="toctree-l3"><a href="#json-parser">Json Parser</a></li> + + <li><a class="toctree-l4" href="#operator-objective">Operator Objective</a></li> + + <li><a class="toctree-l4" href="#class-diagram">Class Diagram</a></li> + + <li><a class="toctree-l4" href="#operator-information">Operator Information</a></li> + + <li><a class="toctree-l4" href="#properties-attributes-and-ports">Properties, Attributes and Ports</a></li> + + <li><a class="toctree-l4" href="#partitioning">Partitioning</a></li> + + <li><a class="toctree-l4" href="#example">Example</a></li> + + + </ul> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../jsonFormatter/">Json Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../transform/">Transform Operator</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../csvformatter/">Csv Formatter</a> + + </li> + + + + <li class="toctree-l1 "> + <a class="" href="../s3outputmodule/">S3 Output Module</a> + + </li> + + + </ul> +<li> + + </ul> + </div> + + </nav> + + <section data-toggle="wy-nav-shift" class="wy-nav-content-wrap"> + + + <nav class="wy-nav-top" role="navigation" aria-label="top navigation"> + <i data-toggle="wy-nav-top" class="fa fa-bars"></i> + <a href="../..">Apache Apex Malhar Documentation</a> + </nav> + + + <div class="wy-nav-content"> + <div class="rst-content"> + <div role="navigation" aria-label="breadcrumbs navigation"> + <ul class="wy-breadcrumbs"> + <li><a href="../..">Docs</a> »</li> + + + + <li>Operators »</li> + + + + <li>Json Parser</li> + <li class="wy-breadcrumbs-aside"> + + </li> + </ul> + <hr/> +</div> + <div role="main"> + <div class="section"> + + <h1 id="json-parser">Json Parser</h1> +<h2 id="operator-objective">Operator Objective</h2> +<p>Purpose of JsonParser operator is to parse JSON records and construct a Plain Old Java Object ("POJO") out of it. The operator also emits each record as JSONObject if the relevant output port is connected. User can also provide a schema describing JSON data to validate incoming JSON records. Valid records will be emitted as POJO / JSONObject while invalid ones are emitted on error port with the error message if the error port is connected.</p> +<p>Json Parser is <strong>idempotent</strong>, <strong>fault-tolerant</strong> & <strong>statically/dynamically partitionable</strong>.</p> +<h2 id="class-diagram">Class Diagram</h2> +<p><img alt="" src="../images/jsonParser/JsonParser.png" /></p> +<h2 id="operator-information">Operator Information</h2> +<ol> +<li>Operator location:<strong><em>malhar-contrib</em></strong></li> +<li>Available since:<strong><em>3.2.0</em></strong></li> +<li>Operator state:<strong><em>Evolving</em></strong></li> +<li>Java Package:<a href="https://github.com/apache/apex-malhar/blob/master/contrib/src/main/java/com/datatorrent/contrib/parser/JsonParser.java";>com.datatorrent.contrib.parser.JsonParser</a></li> +</ol> +<h2 id="properties-attributes-and-ports">Properties, Attributes and Ports</h2> +<h3 id="properties-of-json-parser"><a name="props"></a>Properties of Json Parser</h3> +<table> +<thead> +<tr> +<th><strong>Property</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +<th><strong>Default Value</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>jsonSchema</em></td> +<td><a href="http://json-schema.org/";>Schema</a> describing JSON data. Incoming records can be validated using the jsonSchema. If the data is not as per the requirements specified in jsonSchema, they are emitted on the error port.This is an optional property. If the schema is not provided, incoming tuples are simply converted to POJO or JSONObject without any validations</td> +<td>String</td> +<td>NO</td> +<td>N/A</td> +</tr> +</tbody> +</table> +<h3 id="platform-attributes-that-influences-operator-behavior">Platform Attributes that influences operator behavior</h3> +<table> +<thead> +<tr> +<th><strong>Attribute</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>out.TUPLE_CLASS</em></td> +<td>TUPLE_CLASS attribute on output port which tells operator the class of POJO which needs to be emitted. The name of the field members of the class must match with the names in incoming POJO. The operator ignores unknown properties.</td> +<td>Class or FQCN</td> +<td>Yes</td> +</tr> +</tbody> +</table> +<h3 id="ports">Ports</h3> +<table> +<thead> +<tr> +<th><strong>Port</strong></th> +<th><strong>Description</strong></th> +<th><strong>Type</strong></th> +<th><strong>Mandatory</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>in</em></td> +<td>Tuples that needs to be parsed are recieved on this port</td> +<td>byte[]</td> +<td>Yes</td> +</tr> +<tr> +<td><em>out</em></td> +<td>Valid Tuples that are emitted as pojo. Tuples are converted to POJO only if the port is connected.</td> +<td>Object (POJO)</td> +<td>No</td> +</tr> +<tr> +<td><em>parsedOutput</em></td> +<td>Valid Tuples that are emitted as JSONObject. Tuples are converted to JSONObject only if the port is connected.</td> +<td>JSONObject</td> +<td>No</td> +</tr> +<tr> +<td><em>err</em></td> +<td>Invalid Tuples are emitted with error message. Invaid tuples are discarded if the port is not connected.</td> +<td>KeyValPair <String, String></td> +<td>No</td> +</tr> +</tbody> +</table> +<h2 id="partitioning">Partitioning</h2> +<p>JSON Parser is both statically and dynamically partitionable.</p> +<h3 id="static-partitioning">Static Partitioning</h3> +<p>This can be achieved in 2 ways</p> +<ol> +<li>Specifying the partitioner and number of partitions in the populateDAG() method</li> +</ol> +<pre><code class="java">JsonParser jsonParser = dag.addOperator("jsonParser", JsonParser.class); +StatelessPartitioner<JsonParser> partitioner1 = new StatelessPartitioner<JsonParser>(2); +dag.setAttribute(jsonParser, Context.OperatorContext.PARTITIONER, partitioner1 ); +</code></pre> + +<ol> +<li>Specifying the partitioner in properties file.</li> +</ol> +<pre><code class="xml"> <property> + <name>dt.operator.{OperatorName}.attr.PARTITIONER</name> + <value>com.datatorrent.common.partitioner.StatelessPartitioner:2</value> + </property> +</code></pre> + +<p>where {OperatorName} is the name of the JsonParser operator. + Above lines will partition JsonParser statically 2 times. Above value can be changed accordingly to change the number of static partitions.</p> +<h3 id="dynamic-paritioning">Dynamic Paritioning</h3> +<p>JsonParser can be dynamically partitioned using an out-of-the-box partitioner:</p> +<h4 id="throughput-based">Throughput based</h4> +<p>Following code can be added to populateDAG method of application to dynamically partition JsonParser:</p> +<pre><code class="java">JsonParser jsonParser = dag.addOperator("jsonParser", JsonParser.class); +StatelessThroughputBasedPartitioner<JsonParser> partitioner = new StatelessThroughputBasedPartitioner<>(); +partitioner.setCooldownMillis(conf.getLong(COOL_DOWN_MILLIS, 10000)); +partitioner.setMaximumEvents(conf.getLong(MAX_THROUGHPUT, 30000)); +partitioner.setMinimumEvents(conf.getLong(MIN_THROUGHPUT, 10000)); +dag.setAttribute(jsonParser, OperatorContext.STATS_LISTENERS, Arrays.asList(new StatsListener[]{partitioner})); +dag.setAttribute(jsonParser, OperatorContext.PARTITIONER, partitioner); +</code></pre> + +<p>Above code will dynamically partition JsonParser when the throughput changes. +If the overall throughput of JsonParser goes beyond 30000 or less than 10000, the platform will repartition JsonParser +to balance throughput of a single partition to be between 10000 and 30000. +CooldownMillis of 10000 will be used as the threshold time for which the throughput change is observed.</p> +<h2 id="example">Example</h2> +<p>Example for Json Parser can be found at: <a href="https://github.com/DataTorrent/examples/tree/master/tutorials/parser";>https://github.com/DataTorrent/examples/tree/master/tutorials/parser</a></p> + + </div> + </div> + <footer> + + <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation"> + + <a href="../jsonFormatter/" class="btn btn-neutral float-right" title="Json Formatter">Next <span class="icon icon-circle-arrow-right"></span></a> + + + <a href="../xmlParserOperator/" class="btn btn-neutral" title="XML Parser"><span class="icon icon-circle-arrow-left"></span> Previous</a> + + </div> + + + <hr/> + + <div role="contentinfo"> + <!-- Copyright etc --> + + </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> + + </div> + </div> + + </section> + + </div> + +<div class="rst-versions" role="note" style="cursor: pointer"> + <span class="rst-current-version" data-toggle="rst-current-version"> + + + <span><a href="../xmlParserOperator/" style="color: #fcfcfc;">« Previous</a></span> + + + <span style="margin-left: 15px"><a href="../jsonFormatter/" style="color: #fcfcfc">Next »</a></span> + + </span> +</div> + +</body> +</html>
