<commit_before>.. :Author: Continuuity, Inc.
   :Description: Advanced Reactor Features

=====================================
Advanced Continuuity Reactor Features
=====================================

Building Big Data Applications

.. reST Editor: .. section-numbering::
.. reST Editor: .. contents::


Flow System
===========
**Flows** are user-implemented real-time stream processors. They are comprised of one or more **Flowlets** that are wired together into a directed acyclic graph or DAG. Flowlets pass data between one another; each Flowlet is able to perform custom logic and execute data operations for each individual data object it processes.

A Flowlet processes the data objects from its input one by one. If a Flowlet has multiple inputs, they are consumed in a round-robin fashion. When processing a single input object, all operations, including the removal of the object from the input, and emission of data to the outputs, are executed in a transaction. This provides us with Atomicity, Consistency, Isolation, and Durability (ACID) properties, and helps assure a unique and core property of the Flow system: it guarantees atomic and "exactly-once" processing of each input object by each Flowlet in the DAG.

Batch Execution
---------------
By default, a Flowlet processes a single data object at a time within a single transaction. To increase throughput, you can also process a batch of data objects within the same transaction::

	@Batch(100)
	@ProcessInput
	public void process(String words) {
	  ...

For the batch example above, the **process** method will be called up to 100 times with different data objects read from the input per transaction.

If you are interested to know when a batch begins and ends, you can use **Iterator** as the method argument::

	@Batch(100)
	@ProcessInput
	public void process(Iterator<String> words) {
	  ...<commit_msg>Update Flowlet @Batch documentation based on review comment.<commit_after>.. :Author: Continuuity, Inc.
   :Description: Advanced Reactor Features

=====================================
Advanced Continuuity Reactor Features
=====================================

Building Big Data Applications

.. reST Editor: .. section-numbering::
.. reST Editor: .. contents::


Flow System
===========
**Flows** are user-implemented real-time stream processors. They are comprised of one or more **Flowlets** that are wired together into a directed acyclic graph or DAG. Flowlets pass data between one another; each Flowlet is able to perform custom logic and execute data operations for each individual data object it processes.

A Flowlet processes the data objects from its input one by one. If a Flowlet has multiple inputs, they are consumed in a round-robin fashion. When processing a single input object, all operations, including the removal of the object from the input, and emission of data to the outputs, are executed in a transaction. This provides us with Atomicity, Consistency, Isolation, and Durability (ACID) properties, and helps assure a unique and core property of the Flow system: it guarantees atomic and "exactly-once" processing of each input object by each Flowlet in the DAG.

Batch Execution
---------------
By default, a Flowlet processes a single data object at a time within a single transaction. To increase throughput, you can also process a batch of data objects within the same transaction::

	@Batch(100)
	@ProcessInput
	public void process(String words) {
	  ...

For the above batch example, the **process** method will be called up to 100 times per transaction, with different data objects read from the input each time it is called.

If you are interested in knowing when a batch begins and ends, you can use an **Iterator** as the method argument::

	@Batch(100)
	@ProcessInput
	public void process(Iterator<String> words) {
	  ...
{'fraction_non_alphanumeric': 0.10486599011189175, 'fraction_numerical': 0.00468384074941452, 'mean_word_length': 4.51506456241033, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6527200', 'n_tokens_mistral': 927, 'n_tokens_neox': 878, 'n_words': 529}