http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/file_splitter/index.html ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/file_splitter/index.html b/content/docs/malhar-3.7/operators/file_splitter/index.html new file mode 100644 index 0000000..12e0f38 --- /dev/null +++ b/content/docs/malhar-3.7/operators/file_splitter/index.html @@ -0,0 +1,500 @@ +<!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>File Splitter - 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 = "File Splitter"; + var mkdocs_page_input_path = "operators/file_splitter.md"; + var mkdocs_page_url = "/operators/file_splitter/"; + </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 current"> + <a class="current" href="./">File Splitter</a> + + <ul> + + <li class="toctree-l3"><a href="#file-splitter">File Splitter</a></li> + + <li><a class="toctree-l4" href="#why-is-it-needed">Why is it needed?</a></li> + + <li><a class="toctree-l4" href="#class-diagram">Class Diagram</a></li> + + <li><a class="toctree-l4" href="#abstractfilesplitter">AbstractFileSplitter</a></li> + + <li><a class="toctree-l4" href="#filesplitterbase">FileSplitterBase</a></li> + + <li><a class="toctree-l4" href="#filesplitterinput">FileSplitterInput</a></li> + + <li><a class="toctree-l4" href="#handling-of-split-records">Handling of split records</a></li> + + + </ul> + + </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="../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>File Splitter</li> + <li class="wy-breadcrumbs-aside"> + + </li> + </ul> + <hr/> +</div> + <div role="main"> + <div class="section"> + + <h1 id="file-splitter">File Splitter</h1> +<p>This is a simple operator whose main function is to split a file virtually and create metadata describing the files and the splits. </p> +<h2 id="why-is-it-needed">Why is it needed?</h2> +<p>It is a common operation to read a file and parse it. This operation can be parallelized by having multiple partitions of such operators and each partition operating on different files. However, at times when a file is large then a single partition reading it can become a bottleneck. +In these cases, throughput can be increased if instances of the partitioned operator can read and parse non-overlapping sets of file blocks. This is where file splitter comes in handy. It creates metadata of blocks of file which serves as tasks handed out to downstream operator partitions. +The downstream partitions can read/parse the block without the need of interacting with other partitions.</p> +<h2 id="class-diagram">Class Diagram</h2> +<p><img alt="FileSplitter class dierarchy" src="../images/filesplitter/classdiagram.png" /></p> +<h2 id="abstractfilesplitter">AbstractFileSplitter</h2> +<p>The abstract implementation defines the logic of processing <code>FileInfo</code>. This comprises the following tasks - </p> +<ul> +<li> +<p>building <code>FileMetadata</code> per file and emitting it. This metadata contains the file information such as filepath, no. of blocks in it, length of the file, all the block ids, etc.</p> +</li> +<li> +<p>creating <code>BlockMetadataIterator</code> from <code>FileMetadata</code>. The iterator lazy-loads the block metadata when needed. We use an iterator because the no. of blocks in a file can be huge if the block size is small and loading all of them at once in memory may cause out of memory errors.</p> +</li> +<li> +<p>retrieving <code>BlockMetadata.FileBlockMetadata</code> from the block metadata iterator and emitting it. The FileBlockMetadata contains the block id, start offset of the block, length of file in the block, etc. The number of block metadata emitted per window are controlled by <code>blocksThreshold</code> setting which by default is 1. </p> +</li> +</ul> +<p>The main utility method that performs all the above tasks is the <a href="#process_method"><code>process()</code></a> method. Concrete implementations can invoke this method whenever they have data to process.</p> +<h3 id="ports">Ports</h3> +<p>Declares only output ports on which file metadata and block metadata are emitted.</p> +<ul> +<li>filesMetadataOutput: metadata for each file is emitted on this port. </li> +<li>blocksMetadataOutput: metadata for each block is emitted on this port. </li> +</ul> +<h3 id="process-method"><a name="process_method"></a><code>process()</code> method</h3> +<p>When process() is invoked, any pending blocks from the current file are emitted on the 'blocksMetadataOutput' port. If the threshold for blocks per window is still not met then a new input file is processed - corresponding metadata is emitted on 'filesMetadataOutput' and more of its blocks are emitted. This operation is repeated until the <code>blocksThreshold</code> is reached or there are no more new files.</p> +<pre><code class="java"> protected void process() + { + if (blockMetadataIterator != null && blockCount < blocksThreshold) { + emitBlockMetadata(); + } + + FileInfo fileInfo; + while (blockCount < blocksThreshold && (fileInfo = getFileInfo()) != null) { + if (!processFileInfo(fileInfo)) { + break; + } + } + } +</code></pre> + +<h3 id="abstract-methods">Abstract methods</h3> +<ul> +<li> +<p><code>FileInfo getFileInfo()</code>: called from within the <code>process()</code> and provides the next file to process.</p> +</li> +<li> +<p><code>long getDefaultBlockSize()</code>: provides the block size which is used when user hasn't configured the size.</p> +</li> +<li> +<p><code>FileStatus getFileStatus(Path path)</code>: provides the <code>org.apache.hadoop.fs.FileStatus</code> instance for a path. </p> +</li> +</ul> +<h3 id="configuration">Configuration</h3> +<ol> +<li><strong>blockSize</strong>: size of a block.</li> +<li><strong>blocksThreshold</strong><a name="blocksThreshold"></a>: threshold on the number of blocks emitted by file splitter every window. This setting is used for throttling the work for downstream operators.</li> +</ol> +<h2 id="filesplitterbase">FileSplitterBase</h2> +<p>Simple operator that receives tuples of type <code>FileInfo</code> on its <code>input</code> port. <code>FileInfo</code> contains the information (currently just the file path) about the file which this operator uses to create file metadata and block metadata.</p> +<h3 id="example-application">Example application</h3> +<p>This is a simple sub-dag that demonstrates how FileSplitterBase can be plugged into an application. +<img alt="Application with FileSplitterBase" src="../images/filesplitter/baseexample.png" /></p> +<p>The upstream operator emits tuples of type <code>FileInfo</code> on its output port which is connected to splitter input port. The downstream receives tuples of type <code>BlockMetadata.FileBlockMetadata</code> from the splitter's block metadata output port.</p> +<pre><code class="java">public class ApplicationWithBaseSplitter implements StreamingApplication +{ + @Override + public void populateDAG(DAG dag, Configuration configuration) + { + JMSInput input = dag.addOperator("Input", new JMSInput()); + FileSplitterBase splitter = dag.addOperator("Splitter", new FileSplitterBase()); + FSSliceReader blockReader = dag.addOperator("BlockReader", new FSSliceReader()); + ... + dag.addStream("file-info", input.output, splitter.input); + dag.addStream("block-metadata", splitter.blocksMetadataOutput, blockReader.blocksMetadataInput); + ... + } + + public static class JMSInput extends AbstractJMSInputOperator<AbstractFileSplitter.FileInfo> + { + + public final transient DefaultOutputPort<AbstractFileSplitter.FileInfo> output = new DefaultOutputPort<>(); + + @Override + protected AbstractFileSplitter.FileInfo convert(Message message) throws JMSException + { + //assuming the message is a text message containing the absolute path of the file. + return new AbstractFileSplitter.FileInfo(null, ((TextMessage)message).getText()); + } + + @Override + protected void emit(AbstractFileSplitter.FileInfo payload) + { + output.emit(payload); + } + } +} +</code></pre> + +<h3 id="ports_1">Ports</h3> +<p>Declares an input port on which it receives tuples from the upstream operator. Output ports are inherited from AbstractFileSplitter.</p> +<ul> +<li>input: non optional port on which tuples of type <code>FileInfo</code> are received.</li> +</ul> +<h3 id="configuration_1">Configuration</h3> +<ol> +<li><strong>file</strong>: path of the file from which the filesystem is inferred. FileSplitter creates an instance of <code>org.apache.hadoop.fs.FileSystem</code> which is why this path is needed. </li> +</ol> +<pre><code>FileSystem.newInstance(new Path(file).toUri(), new Configuration()); +</code></pre> + +<p>The fs instance is then used to fetch the default block size and <code>org.apache.hadoop.fs.FileStatus</code> for each file path.</p> +<h2 id="filesplitterinput">FileSplitterInput</h2> +<p>This is an input operator that discovers files itself. The scanning of the directories for new files is asynchronous which is handled by <code>TimeBasedDirectoryScanner</code>. The function of TimeBasedDirectoryScanner is to periodically scan specified directories and find files which were newly added or modified. The interaction between the operator and the scanner is depicted in the diagram below.</p> +<p><img alt="Interaction between operator and scanner" src="../images/filesplitter/sequence.png" /></p> +<h3 id="example-application_1">Example application</h3> +<p>This is a simple sub-dag that demonstrates how FileSplitterInput can be plugged into an application.</p> +<p><img alt="Application with FileSplitterInput" src="../images/filesplitter/inputexample.png" /></p> +<p>Splitter is the input operator here that sends block metadata to the downstream BlockReader.</p> +<pre><code class="java"> @Override + public void populateDAG(DAG dag, Configuration configuration) + { + FileSplitterInput input = dag.addOperator("Input", new FileSplitterInput()); + FSSliceReader reader = dag.addOperator("Block Reader", new FSSliceReader()); + ... + dag.addStream("block-metadata", input.blocksMetadataOutput, reader.blocksMetadataInput); + ... + } + +</code></pre> + +<h3 id="ports_2">Ports</h3> +<p>Since it is an input operator there are no input ports and output ports are inherited from AbstractFileSplitter.</p> +<h3 id="configuration_2">Configuration</h3> +<ol> +<li><strong>scanner</strong>: the component that scans directories asynchronously. It is of type <code>com.datatorrent.lib.io.fs.FileSplitter.TimeBasedDirectoryScanner</code>. The basic implementation of TimeBasedDirectoryScanner can be customized by users. </li> +</ol> +<p>a. <strong>files</strong>: comma separated list of directories to scan. </p> +<p>b. <strong>recursive</strong>: flag that controls whether the directories should be scanned recursively. </p> +<p>c. <strong>scanIntervalMillis</strong>: interval specified in milliseconds after which another scan iteration is triggered. </p> +<p>d. <strong>filePatternRegularExp</strong>: regular expression for accepted file names. </p> +<p>e. <strong>trigger</strong>: a flag that triggers a scan iteration instantly. If the scanner thread is idling then it will initiate a scan immediately otherwise if a scan is in progress, then the new iteration will be triggered immediately after the completion of current one. +2. <strong>idempotentStorageManager</strong>: by default FileSplitterInput is idempotent. +Idempotency ensures that the operator will process the same set of files/blocks in a window if it has seen that window previously, i.e., before a failure. For example, let's say the operator completed window 10 and failed somewhere between window 11. If the operator gets restored at window 10 then it will process the same file/block again in window 10 which it did in the previous run before the failure. Idempotency is important but comes with higher cost because at the end of each window the operator needs to persist some state with respect to that window. Therefore, if one doesn't care about idempotency then they can set this property to be an instance of <code>com.datatorrent.lib.io.IdempotentStorageManager.NoopIdempotentStorageManager</code>.</p> +<h2 id="handling-of-split-records">Handling of split records</h2> +<p>Splitting of files to create tasks for downstream operator needs to be a simple operation that doesn't consume a lot of resources and is fast. This is why the file splitter doesn't open files to read. The downside of that is if the file contains records then a record may split across adjacent blocks. Handling of this is left to the downstream operator.</p> +<p>We have created Block readers in Apex-malhar library that handle line splits efficiently. The 2 line readers- <code>AbstractFSLineReader</code> and <code>AbstractFSReadAheadLineReader</code> can be found here <a href="https://github.com/apache/incubator-apex-malhar/blob/master/library/src/main/java/com/datatorrent/lib/io/block/AbstractFSBlockReader.java";>AbstractFSBlockReader</a>.</p> + + </div> + </div> + <footer> + + <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation"> + + <a href="../filter/" class="btn btn-neutral float-right" title="Filter">Next <span class="icon icon-circle-arrow-right"></span></a> + + + <a href="../file_output/" class="btn btn-neutral" title="File Output"><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="../file_output/" style="color: #fcfcfc;">« Previous</a></span> + + + <span style="margin-left: 15px"><a href="../filter/" 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/filter/index.html ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/filter/index.html b/content/docs/malhar-3.7/operators/filter/index.html new file mode 100644 index 0000000..19ea1d3 --- /dev/null +++ b/content/docs/malhar-3.7/operators/filter/index.html @@ -0,0 +1,456 @@ +<!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>Filter - 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 = "Filter"; + var mkdocs_page_input_path = "operators/filter.md"; + var mkdocs_page_url = "/operators/filter/"; + </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 current"> + <a class="current" href="./">Filter</a> + + <ul> + + <li class="toctree-l3"><a href="#filter">Filter</a></li> + + <li><a class="toctree-l4" href="#operator-objective">Operator Objective</a></li> + + <li><a class="toctree-l4" href="#operator-usecase">Operator Usecase</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="#limitations">Limitations</a></li> + + <li><a class="toctree-l4" href="#example">Example</a></li> + + + </ul> + + </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="../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>Filter</li> + <li class="wy-breadcrumbs-aside"> + + </li> + </ul> + <hr/> +</div> + <div role="main"> + <div class="section"> + + <h1 id="filter">Filter</h1> +<h2 id="operator-objective">Operator Objective</h2> +<p>This operator receives an POJO (<a href="https://en.wikipedia.org/wiki/Plain_Old_Java_Object";>Plain Old Java Object</a>) as an incoming tuple +and based on the filter condition it emits filtered tuples on one output port and rest on another output port.</p> +<p>Filter operator supports quasi Java expressions to specify filter rule.</p> +<p>Filter operator does not hold any state and is <strong>idempotent</strong>, <strong>fault-tolerant</strong> and <strong>statically/dynamically partitionable</strong>.</p> +<h2 id="operator-usecase">Operator Usecase</h2> +<ol> +<li><strong><em>Customer data</em></strong> usually contains a field customer category/segment. One wants some analysis to be done for specific customer segment. One could use this filter operator to filter the records based on segment for some analysis for specific customer segment. </li> +<li><strong><em>Log data</em></strong> processing pipeline may want to filter logs from specific machine/router/switch.</li> +</ol> +<h2 id="operator-information">Operator Information</h2> +<ol> +<li>Operator location: <strong><em><a href="https://github.com/apache/apex-malhar/tree/master/library";>malhar-library</a></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/filter/FilterOperator.html";>com.datatorrent.lib.filter.FilterOperator</a></em></strong></li> +</ul> +</li> +</ol> +<h2 id="properties-attributes-and-ports">Properties, Attributes and Ports</h2> +<h3 id="properties-of-filteroperator"><a name="props"></a>Properties of FilterOperator</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>condition</em></td> +<td>condition/expression with which Filtering is done.</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>additionalExpressionFunctions</em></td> +<td>List of import classes/method that should be made statically available to expression to use.</td> +<td><code>List<String></code></td> +<td>No</td> +<td>Empty List</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>port.input.attr.TUPLE_CLASS</em></td> +<td>TUPLE_CLASS attribute on input port indicates the class of POJO which incoming tuple</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>Connection Required</strong></th> +</tr> +</thead> +<tbody> +<tr> +<td><em>input</em></td> +<td>Tuple which needs to be filtered are received on this port</td> +<td>Object (POJO)</td> +<td>Yes</td> +</tr> +<tr> +<td><em>truePort</em></td> +<td>Tuples which satisfies <a href="#props">condition</a> are emitted on this port</td> +<td>Object (POJO)</td> +<td>No</td> +</tr> +<tr> +<td><em>falsePort</em></td> +<td>Tuples which does not satisfy <a href="#props">condition</a> are emitted on this port</td> +<td>Object (POJO)</td> +<td>No</td> +</tr> +</tbody> +</table> +<h2 id="limitations">Limitations</h2> +<p>Current <code>FilterOperator</code> has following limitation:</p> +<ol> +<li><a href="https://issues.apache.org/jira/browse/APEXMALHAR-2175";>APEXMALHAR-2175</a> : Filter condition is not able to correctly handle java reserved words.</li> +</ol> +<h2 id="example">Example</h2> +<p>Example for <code>FilterOperator</code> can be found at: <a href="https://github.com/DataTorrent/examples/tree/master/tutorials/filter";>https://github.com/DataTorrent/examples/tree/master/tutorials/filter</a></p> + + </div> + </div> + <footer> + + <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation"> + + <a href="../AbstractJdbcTransactionableOutputOperator/" class="btn btn-neutral float-right" title="Jdbc Output Operator">Next <span class="icon icon-circle-arrow-right"></span></a> + + + <a href="../file_splitter/" class="btn btn-neutral" title="File Splitter"><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="../file_splitter/" style="color: #fcfcfc;">« Previous</a></span> + + + <span style="margin-left: 15px"><a href="../AbstractJdbcTransactionableOutputOperator/" 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/fsInputOperator/index.html ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/fsInputOperator/index.html b/content/docs/malhar-3.7/operators/fsInputOperator/index.html new file mode 100644 index 0000000..89e0ebe --- /dev/null +++ b/content/docs/malhar-3.7/operators/fsInputOperator/index.html @@ -0,0 +1,832 @@ +<!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>File 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 = "File Input"; + var mkdocs_page_input_path = "operators/fsInputOperator.md"; + var mkdocs_page_url = "/operators/fsInputOperator/"; + </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 current"> + <a class="current" href="./">File Input</a> + + <ul> + + <li class="toctree-l3"><a href="#file-input-operator">File Input Operator</a></li> + + <li><a class="toctree-l4" href="#operator-objective">Operator Objective</a></li> + + <li><a class="toctree-l4" href="#overview">Overview</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">Partitioning</a></li> + + <li><a class="toctree-l4" href="#operator-information">Operator Information</a></li> + + <li><a class="toctree-l4" href="#ports">Ports</a></li> + + + <li class="toctree-l3"><a href="#abstract-methods">Abstract Methods</a></li> + + + <li class="toctree-l3"><a href="#derived-classes">Derived Classes</a></li> + + <li><a class="toctree-l4" href="#1-abstractftpinputoperator">1. AbstractFTPInputOperator</a></li> + + <li><a class="toctree-l4" href="#2-ftpstringinputoperator">2. FTPStringInputOperator</a></li> + + <li><a class="toctree-l4" href="#3-abstractparquetfilereader">3. AbstractParquetFileReader</a></li> + + <li><a class="toctree-l4" href="#4-abstractthroughputfileinputoperator">4. AbstractThroughputFileInputOperator</a></li> + + <li><a class="toctree-l4" href="#5-linebylinefileinputoperator">5. LineByLineFileInputOperator</a></li> + + + <li class="toctree-l3"><a href="#example-implementation-using-a-custom-character-encoding">Example Implementation Using a Custom Character Encoding</a></li> + + + <li class="toctree-l3"><a href="#common-implementation-scenarios">Common Implementation Scenarios</a></li> + + + </ul> + + </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="../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>File Input</li> + <li class="wy-breadcrumbs-aside"> + + </li> + </ul> + <hr/> +</div> + <div role="main"> + <div class="section"> + + <h1 id="file-input-operator">File Input Operator</h1> +<h2 id="operator-objective">Operator Objective</h2> +<p>This operator is designed to scan a directory for files, read and split file content into tuples +such as lines or a block of bytes, and finally emit them on output ports defined in concrete +subclasses for further processing by downstream operators. +It can be used with any filesystem supported by Hadoop like HDFS, S3, ftp, NFS etc.</p> +<h2 id="overview">Overview</h2> +<p>The operator is <strong>idempotent</strong>, <strong>fault-tolerant</strong> and <strong>partitionable</strong>.</p> +<p>Logic for directory scanning is encapsulated in the <code>DirectoryScanner</code> static inner class +which provides functions such as matching file names against a regular expression, tracking files +that have already been processed (so that they are not processed again), filtering files based +on the hashcode of the file names in the presence of partitioning so that each file is +processed by a unique partition. This class can be extended if necessary to provide +additional capabilities such as scanning multiple directories.</p> +<p>It tracks the current file offset as part of checkpoint state. It it fails and is restarted +by the platform, it will seek to the saved offset to avoid duplicate processing. Exactly once processing +for fault tolerance is handled using window data manager. For more details check the blog about <a href="https://www.datatorrent.com/blog/fault-tolerant-file-processing/";>Fault-Tolerant File Processing</a>. +It supports both static and dynamic partitioning.</p> +<h2 id="use-cases">Use Cases</h2> +<p>This operator is suitable for use in an environment where small to medium sized files are +deposited in a specific directory on a regular basis. For very large files a better alternative +is the <code>FileSplitter</code> and <code>BlockReader</code> combination since they allow such files to be processed +by multiple partitions to achieve higher throughput. Additionally, files which are continually +modified by other processes are not suitable for processing with this operator since they may +yield unpredictable results.</p> +<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 <code>String</code> or <code>byte[]</code>) for the +generic parameter and also implement a couple of abstract methods: <code>readEntity()</code> to read +the next tuple from the currently open file and <code>emit()</code> to process the next 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 <code>emit()</code> implementation dispatches tuples to the +desired output ports.</p> +<p>A simple concrete implementation is provided in Malhar: <code>LineByLineFileInputOperator</code>. +It uses <code>String</code> for the generic parameter, defines a single output port and processes each +line of the input file as a tuple. It is discussed further below.</p> +<h2 id="partitioning">Partitioning</h2> +<h4 id="static-partitioning">Static Partitioning</h4> +<p>Configure parameter <code>partitionCount</code> to define the desired number of initial partitions +(4 in this example).</p> +<pre><code class="xml"><property> + <name>dt.operator.{OperatorName}.prop.partitionCount</name> + <value>4</value> +</property> +</code></pre> + +<p>where <em>{OperatorName}</em> is the name of the input operator.</p> +<h4 id="dynamic-partitioning">Dynamic Partitioning</h4> +<p>Dynamic partitioning -- changing the number of partitions of one or more operators +in a running application -- can be achieved in multiple ways: +- Use the command line tool <code>apex</code> or the UI console to change the value of the + <code>partitionCount</code> property of the running operator. This change is detected in + <code>processStats()</code> (which is invoked periodically by the platform) where, if the + current partition count (<code>currentPartitions</code>) and the desired partition count + (<code>partitionCount</code>) differ, the <code>repartitionRequired</code> flag in the response is set. + This causes the platform to invoke <code>definePartitions()</code> to create a new set of + partitions with the desired count. +- Override <code>processStats()</code> and within it, based on the statistics in the + incoming parameter or any other factors, define a new desired value of + <code>partitionCount</code> and finally, if this value differs from the current partition + count, set the <code>repartitionRequired</code> flag in the response.</p> +<p>The details of actually creating the new set of partitions can be customized by overriding +the <code>definePartitions()</code> method. There are a couple of things to keep in mind when doing this. +The first is that repartitioning needs some care when the operator has state (as is the +case here): Existing state from current operator partitions needs to redistributed to the +new partitions in a logically consistent way. The second is that some or all of the +current set of partitions, which is an input parameter to <code>definePartitions()</code>, can be +copied over to the new set; such partitions will continue running and will not be +restarted. Any existing partitions that are not present in the new set will be shutdown. +The current re-partitioning logic does not preserve any existing partitions, so upon +a repartition event, all existing partitions are shutdown and the new ones started.</p> +<h2 id="operator-information">Operator Information</h2> +<ol> +<li>Operator location: <strong><em>malhar-library</em></strong></li> +<li>Available since: <strong><em>1.0.2</em></strong></li> +<li>Operator state: <strong><em>Stable</em></strong></li> +<li>Java Packages:<ul> +<li>Operator: <strong><em><a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/io/fs/AbstractFileInputOperator.html";>com.datatorrent.lib.io.fs.AbstractFileInputOperator</a></em></strong></li> +</ul> +</li> +</ol> +<h3 id="abstractfileinputoperator">AbstractFileInputOperator</h3> +<p>This is the abstract implementation that, as noted above, scans a single directory. +It can be extended to modify functionality or add new capabilities. For example, the +directory scanner can be overriden to monitor multiple directories. <a href="https://github.com/DataTorrent/examples/tree/master/tutorials/fileIO-multiDir";>This</a> example demonstrates how to do that. +As noted in the overview above, this class has no ports, so concrete subclasses will need to +provide them if necessary.</p> +<p><img alt="AbstractFileInputOperator.png" src="../images/fsInput/operatorsClassDiagram.png" /></p> +<h3 id="properties-of-abstractfileinputoperator"><a name="AbstractFileInputOperatorProps"></a>Properties of AbstractFileInputOperator</h3> +<p>Several properties are available to configure the behavior of this operator and they are +summarized in the table below. Of these, only <code>directory</code> is required: it specifies +the path of the monitored directory. It can be set like this:</p> +<pre><code class="xml"><property> + <name>dt.operator.{OperatorName}.prop.directory</name> + <value>/tmp/fileInput</value> +</property> +</code></pre> + +<p>If new files appear with high frequency in this directory +and they need to be processed as soon as they appear, reduce the value of <code>scanIntervalMillis</code>; +if they appear rarely or if some delay in processing a new file is acceptable, increase it. +Obviously, smaller values will result in greater IO activity for the corresponding filesystem.</p> +<p>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>emitBatchSize</code> property. Conversely, if the downstream operators can handle the load, increase +the value to enhance throughput.</p> +<p>The <code>partitionCount</code> parameter has already been discussed above.</p> +<p>Occasionally, some files get into a bad state and cause errors when an attempt is made to +read from them. The causes vary depending on the filesystem type ranging from corrupted +filesystems to network issues. In such cases, the operator will retry reading from such +files a limited number of times before blacklisting those files. This retry count is +defined by the <code>maxRetryCount</code> property.</p> +<p>Finally, the specific scanner class used to monitor the input directories can be configured +by setting the <code>scanner</code> property.</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>directory</em></td> +<td>absolute path of directory to be scanned</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>scanIntervalMillis</em></td> +<td>Interval in milliseconds after which directory should be scanned for new files</td> +<td>int</td> +<td>No</td> +<td>5000</td> +</tr> +<tr> +<td><em>emitBatchSize</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>1000</td> +</tr> +<tr> +<td><em>partitionCount</em></td> +<td>Desired number of partitions</td> +<td>int</td> +<td>No</td> +<td>1</td> +</tr> +<tr> +<td><em>maxRetryCount</em></td> +<td>Maximum number of times the operator will attempt to process a file</td> +<td>int</td> +<td>No</td> +<td>5</td> +</tr> +<tr> +<td><em>scanner</em></td> +<td>Scanner to scan new files in directory</td> +<td><a href="#DirectoryScanner">DirectoryScanner</a></td> +<td>No</td> +<td>DirectoryScanner</td> +</tr> +</tbody> +</table> +<h4 id="properties-of-directoryscanner"><a name="DirectoryScanner"></a>Properties of DirectoryScanner</h4> +<p>The directory scanner has one optional property: a regular expression to filter files +of interest. If absent, all files in the source directory are processed. It can be +set like this:</p> +<pre><code class="xml"><property> + <name>dt.operator.{OperatorName}.prop.scanner.filePatternRegexp</name> + <value>/tmp/fileInput</value> +</property> +</code></pre> + +<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>filePatternRegexp</em></td> +<td>regex to select files from input directory</td> +<td>String</td> +<td>No</td> +<td>N/A</td> +</tr> +</tbody> +</table> +<h3 id="ports">Ports</h3> +<p>This operator has no ports.</p> +<h2 id="abstract-methods">Abstract Methods</h2> +<p>As described above, concrete subclasses need to provide implementations for these two +methods:</p> +<pre><code class="java">void emit(T tuple); +T readEntity(); +</code></pre> + +<p>Examples of implementations are in the <code>LineByLineFileInputOperator</code> operator and also in +the example at the end of this guide.</p> +<h2 id="derived-classes">Derived Classes</h2> +<h3 id="1-abstractftpinputoperator">1. AbstractFTPInputOperator</h3> +<p>The class is used to read files from FTP file system. As for the above abstract class, concrete +subclasses need to implement the +<a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/io/fs/AbstractFileInputOperator.html#readEntity";>readEntity</a> and +<a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/io/fs/AbstractFileInputOperator.html#emit";>emit</a> methods.</p> +<h4 id="properties"><a name="AbstractFTPInputOperatorProps"></a>Properties</h4> +<p>This operator defines following additional properties beyond those defined in the +<a href="#AbstractFileInputOperatorProps">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>host</em></td> +<td>Hostname of ftp server.</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +<tr> +<td><em>port</em></td> +<td>Port of ftp server.</td> +<td>int</td> +<td>No</td> +<td>21 (default ftp port)</td> +</tr> +<tr> +<td><em>userName</em></td> +<td>Username which is used for login to the server.</td> +<td>String</td> +<td>No</td> +<td>anonymous</td> +</tr> +<tr> +<td><em>password</em></td> +<td>Password which is used for login to the server.</td> +<td>String</td> +<td>No</td> +<td>gues</td> +</tr> +</tbody> +</table> +<h4 id="ports_1">Ports</h4> +<p>This operator has no ports.</p> +<h3 id="2-ftpstringinputoperator">2. FTPStringInputOperator</h3> +<p>This class extends AbstractFTPInputOperator and implements abstract methods to read files available on FTP file system line by line.</p> +<h4 id="properties_1"><a name="FTPStringInputOperatorProps"></a>Properties</h4> +<p>This operator defines no additional properties beyond those defined in the +<a href="#AbstractFTPInputOperatorProps">parent class</a>.</p> +<h4 id="ports_2">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>output</em></td> +<td>Tuples that are read from file are emitted on this port</td> +<td>String</td> +<td>Yes</td> +</tr> +</tbody> +</table> +<h3 id="3-abstractparquetfilereader">3. AbstractParquetFileReader</h3> +<p>Reads Parquet files from input directory using GroupReadSupport. Derived classes need to implement <a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/contrib/parquet/AbstractParquetFileReader.html#convertGroup(Group)">convertGroup(Group)</a> method to convert Group to other type. Also it should implement <a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/io/fs/AbstractFileInputOperator.html#readEntity()">readEntity()</a> and <a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/io/fs/AbstractFileInputOperator.html#emit(T)">emit(T)</a> methods.</p> +<h4 id="properties-of-abstractparquetfilereader"><a name="AbstractParquetFileReaderProps"></a>Properties of AbstractParquetFileReader</h4> +<p>This operator defines following additional properties beyond those defined in the +<a href="#AbstractFileInputOperatorProps">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>parquetSchema</em></td> +<td>Parquet Schema to parse record.</td> +<td>String</td> +<td>Yes</td> +<td>N/A</td> +</tr> +</tbody> +</table> +<h4 id="ports_3">Ports</h4> +<p>This operator has no ports.</p> +<h3 id="4-abstractthroughputfileinputoperator">4. AbstractThroughputFileInputOperator</h3> +<p>This operator extends <code>AbstractFileInputOperator</code> by providing the capability to partition +dynamically based the file backlog. The user can set the preferred number of pending files per operator as well as the maximum number of operators and define a re-partition interval. If a physical operator runs out of files to process and an amount of time greater than or equal to the repartition interval has passed then a new number of operators are created to accommodate the remaining pending files. Derived classes need to implement <a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/io/fs/AbstractFileInputOperator.html#readEntity()">readEntity()</a> and <a href="https://www.datatorrent.com/docs/apidocs/com/datatorrent/lib/io/fs/AbstractFileInputOperator.html#emit(T)">emit(T)</a> methods.</p> +<h4 id="properties-of-abstractthroughputfileinputoperator"><a name="AbstractThroughputFileInputOperatorProps"></a>Properties of AbstractThroughputFileInputOperator</h4> +<p>This operator defines following additional properties beyond those defined in the +<a href="#AbstractFileInputOperatorProps">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>repartitionInterval</em></td> +<td>The minimum amount of time that must pass in milliseconds before the operator can be repartitioned.</td> +<td>long</td> +<td>No</td> +<td>5 minutes</td> +</tr> +<tr> +<td><em>preferredMaxPendingFilesPerOperator</em></td> +<td>the preferred number of pending files per operator.</td> +<td>int</td> +<td>No</td> +<td>10</td> +</tr> +<tr> +<td><em>partitionCount</em></td> +<td>the maximum number of partitions for the operator.</td> +<td>int</td> +<td>No</td> +<td>1</td> +</tr> +</tbody> +</table> +<h4 id="ports_4">Ports</h4> +<p>This operator has no ports.</p> +<h3 id="5-linebylinefileinputoperator">5. LineByLineFileInputOperator</h3> +<p>As mentioned in the overview above, this operator defines a single output port; it reads files +as lines and emits them as Java Strings on the output port. The output port <em>must</em> be connected. +Lines are extracted using the Java <code>BufferedReader</code> class and the default character encoding. +An example illustrating the use of a custom encoding (such as UTF_8) is provided below</p> +<h4 id="properties_2">Properties</h4> +<p>This operator defines no additional properties beyond those defined in the +<a href="#AbstractFileInputOperatorProps">parent class</a>.</p> +<h4 id="ports_5">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>output</em></td> +<td>Tuples that are read from file are emitted on this port</td> +<td>String</td> +<td>Yes</td> +</tr> +</tbody> +</table> +<h2 id="example-implementation-using-a-custom-character-encoding">Example Implementation Using a Custom Character Encoding</h2> +<p>This example demonstrates how to extend the <code>AbstractFileInputOperator</code> to read +UTF-8 encoded data.</p> +<pre><code>public class EncodedDataReader extends AbstractFileInputOperator<String> +{ + public final transient DefaultOutputPort<String> output = new DefaultOutputPort<>(); + protected transient BufferedReader br; + + protected InputStream openFile(Path path) throws IOException + { + InputStream is = super.openFile(path); + br = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8)); + return is; + } + + @Override + protected void closeFile(InputStream is) throws IOException + { + super.closeFile(is); + br.close(); + br = null; + } + + @Override + protected String readEntity() throws IOException + { + return br.readLine(); + } + + @Override + protected void emit(String tuple) + { + output.emit(tuple); + } +} + +</code></pre> + +<h2 id="common-implementation-scenarios">Common Implementation Scenarios</h2> +<p>Sometimes, downstream operators need to know which file each tuple came from; there are a +number of ways of achieving this, each with its own tradeoffs. Some alternatives:</p> +<ul> +<li>If the generic tuple type is a String, each tuple can be prefixed with the file name + with a suitable separator, for example: <code>foo.txt: first line</code>. This works but + has obvious additional costs in both processing (to parse out the two pieces of each + tuple) and network bandwidth utilization.</li> +<li>Define a custom tuple class with two fields: one for the file name and one for tuple data. + The costs are similar to the previous approach though the code is simpler since + parsing is handled behind the scenes by the serialization process.</li> +<li>Define the tuple type to be <code>Object</code> and emit either a custom <code>Tuple</code> object for actual + tuple data or <strong>BOF</strong>/<strong>EOF</strong> objects with the name of the file when a new file begins + or the current file ends. Here, the additional bandwidth consumed is + minimal (just 2 additional tuples at file boundaries) but the type of each tuple needs + to be checked using <code>instanceof</code> in the downstream operators which has some runtime cost.</li> +<li>Similar to the previous approach but define an additional control port dedicated to + the BOF/EOF control tuples. This approach eliminates the runtime cost of using <code>instanceof</code> + but some care is needed because (a) the order of tuples arriving at multiple input ports + in downstream operators cannot be guaranteed -- for example, the BOF/EOF control tuples + may arrive before some of the actual data tuples; and (b) since the operator may read + more than one file in a single streaming window, the downstream operator may not be + able to tell which tuples belong to which file. One way of dealing with this is to + stop emitting data tuples until the next <code>endWindow()</code> callback when an EOF is detected + for the current file; that way, if the downstream operator receives an EOF control tuple, + it has the guarantee that all the data tuples received in the same window belong to the + current file.</li> +</ul> +<p>Of course, other strategies are possible depending on the needs of the particular situation.</p> +<p>When used in a long-running application where a very large number of files are are processed +over time, the internal state (consisting of properties like <code>processedFiles</code>) may grow +correspondingly and this may have some performance impact since each checkpoint saves the +entire operator state. In such situations, it is useful to explore options such as moving +processed files to another directory and trimming operator state variables suitably.</p> + + </div> + </div> + <footer> + + <div class="rst-footer-buttons" role="navigation" aria-label="footer navigation"> + + <a href="../file_output/" class="btn btn-neutral float-right" title="File Output">Next <span class="icon icon-circle-arrow-right"></span></a> + + + <a href="../enricher/" class="btn btn-neutral" title="Enricher"><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="../enricher/" style="color: #fcfcfc;">« Previous</a></span> + + + <span style="margin-left: 15px"><a href="../file_output/" 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/images/blockreader/classdiagram.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/blockreader/classdiagram.png b/content/docs/malhar-3.7/operators/images/blockreader/classdiagram.png new file mode 100644 index 0000000..8fbd6fc Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/blockreader/classdiagram.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/blockreader/flowdiagram.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/blockreader/flowdiagram.png b/content/docs/malhar-3.7/operators/images/blockreader/flowdiagram.png new file mode 100644 index 0000000..1b2897d Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/blockreader/flowdiagram.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/blockreader/fsreaderexample.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/blockreader/fsreaderexample.png b/content/docs/malhar-3.7/operators/images/blockreader/fsreaderexample.png new file mode 100644 index 0000000..571b60a Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/blockreader/fsreaderexample.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/blockreader/totalBacklogProcessing.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/blockreader/totalBacklogProcessing.png b/content/docs/malhar-3.7/operators/images/blockreader/totalBacklogProcessing.png new file mode 100644 index 0000000..2ed481f Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/blockreader/totalBacklogProcessing.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/csvParser/CSVParser.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/csvParser/CSVParser.png b/content/docs/malhar-3.7/operators/images/csvParser/CSVParser.png new file mode 100644 index 0000000..523ba0b Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/csvParser/CSVParser.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/deduper/image00.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/deduper/image00.png b/content/docs/malhar-3.7/operators/images/deduper/image00.png new file mode 100644 index 0000000..ec3e292 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/deduper/image00.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/deduper/image01.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/deduper/image01.png b/content/docs/malhar-3.7/operators/images/deduper/image01.png new file mode 100644 index 0000000..b9e35a9 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/deduper/image01.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/deduper/image02.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/deduper/image02.png b/content/docs/malhar-3.7/operators/images/deduper/image02.png new file mode 100644 index 0000000..689bdfe Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/deduper/image02.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/deduper/image03.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/deduper/image03.png b/content/docs/malhar-3.7/operators/images/deduper/image03.png new file mode 100644 index 0000000..087a0b1 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/deduper/image03.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/deduper/image04.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/deduper/image04.png b/content/docs/malhar-3.7/operators/images/deduper/image04.png new file mode 100644 index 0000000..4d3bd32 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/deduper/image04.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/fileoutput/FileRotation.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/fileoutput/FileRotation.png b/content/docs/malhar-3.7/operators/images/fileoutput/FileRotation.png new file mode 100644 index 0000000..624c96e Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/fileoutput/FileRotation.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/fileoutput/diagram1.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/fileoutput/diagram1.png b/content/docs/malhar-3.7/operators/images/fileoutput/diagram1.png new file mode 100644 index 0000000..0a260de Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/fileoutput/diagram1.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/filesplitter/baseexample.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/filesplitter/baseexample.png b/content/docs/malhar-3.7/operators/images/filesplitter/baseexample.png new file mode 100644 index 0000000..6af2b44 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/filesplitter/baseexample.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/filesplitter/classdiagram.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/filesplitter/classdiagram.png b/content/docs/malhar-3.7/operators/images/filesplitter/classdiagram.png new file mode 100644 index 0000000..6490368 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/filesplitter/classdiagram.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/filesplitter/inputexample.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/filesplitter/inputexample.png b/content/docs/malhar-3.7/operators/images/filesplitter/inputexample.png new file mode 100644 index 0000000..65e199f Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/filesplitter/inputexample.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/filesplitter/sequence.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/filesplitter/sequence.png b/content/docs/malhar-3.7/operators/images/filesplitter/sequence.png new file mode 100644 index 0000000..85cf702 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/filesplitter/sequence.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/fsInput/operatorsClassDiagram.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/fsInput/operatorsClassDiagram.png b/content/docs/malhar-3.7/operators/images/fsInput/operatorsClassDiagram.png new file mode 100644 index 0000000..31c7a0d Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/fsInput/operatorsClassDiagram.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/jdbcinput/operatorsClassDiagram.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/jdbcinput/operatorsClassDiagram.png b/content/docs/malhar-3.7/operators/images/jdbcinput/operatorsClassDiagram.png new file mode 100644 index 0000000..4b0432d Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/jdbcinput/operatorsClassDiagram.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/jdbcoutput/operatorsClassDiagrams.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/jdbcoutput/operatorsClassDiagrams.png b/content/docs/malhar-3.7/operators/images/jdbcoutput/operatorsClassDiagrams.png new file mode 100644 index 0000000..ae7ab42 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/jdbcoutput/operatorsClassDiagrams.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/jsonFormatter/JsonFormatter.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/jsonFormatter/JsonFormatter.png b/content/docs/malhar-3.7/operators/images/jsonFormatter/JsonFormatter.png new file mode 100644 index 0000000..f457c68 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/jsonFormatter/JsonFormatter.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/jsonParser/JsonParser.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/jsonParser/JsonParser.png b/content/docs/malhar-3.7/operators/images/jsonParser/JsonParser.png new file mode 100644 index 0000000..7235efc Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/jsonParser/JsonParser.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/kafkainput/image00.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/kafkainput/image00.png b/content/docs/malhar-3.7/operators/images/kafkainput/image00.png new file mode 100644 index 0000000..0fa00e8 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/kafkainput/image00.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/s3output/s3outputmodule.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/s3output/s3outputmodule.png b/content/docs/malhar-3.7/operators/images/s3output/s3outputmodule.png new file mode 100644 index 0000000..f9c686b Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/s3output/s3outputmodule.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/windowedOperator/allowed-lateness.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/windowedOperator/allowed-lateness.png b/content/docs/malhar-3.7/operators/images/windowedOperator/allowed-lateness.png new file mode 100644 index 0000000..34153ca Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/windowedOperator/allowed-lateness.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-1.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-1.png b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-1.png new file mode 100644 index 0000000..78712e8 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-1.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-2.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-2.png b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-2.png new file mode 100644 index 0000000..19bfab0 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-2.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-3.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-3.png b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-3.png new file mode 100644 index 0000000..e7810eb Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-3.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-4.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-4.png b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-4.png new file mode 100644 index 0000000..43838c8 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/windowedOperator/session-windows-4.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/windowedOperator/sliding-time-windows.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/windowedOperator/sliding-time-windows.png b/content/docs/malhar-3.7/operators/images/windowedOperator/sliding-time-windows.png new file mode 100644 index 0000000..dc55e1a Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/windowedOperator/sliding-time-windows.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/windowedOperator/time-windows.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/windowedOperator/time-windows.png b/content/docs/malhar-3.7/operators/images/windowedOperator/time-windows.png new file mode 100644 index 0000000..cba471c Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/windowedOperator/time-windows.png differ http://git-wip-us.apache.org/repos/asf/apex-site/blob/afbb4705/content/docs/malhar-3.7/operators/images/xmlParser/XmlParser.png ---------------------------------------------------------------------- diff --git a/content/docs/malhar-3.7/operators/images/xmlParser/XmlParser.png b/content/docs/malhar-3.7/operators/images/xmlParser/XmlParser.png new file mode 100644 index 0000000..3964da9 Binary files /dev/null and b/content/docs/malhar-3.7/operators/images/xmlParser/XmlParser.png differ
