<commit_before>.. meta::
    :author: Cask Data, Inc.
    :copyright: Copyright © 2014-2015 Cask Data, Inc.

.. _workflows:

=========
Workflows
=========

*Workflows* are used to automate the execution of a series of :ref:`MapReduce
<mapreduce>`, :ref:`Spark <spark>` or :ref:`custom actions <workflow-custom-actions>`. It
allows for both sequential and :ref:`parallel execution <workflow_parallel>` of programs.

The workflow system allows specifying, executing, scheduling, and monitoring complex
series of jobs and tasks in CDAP. The system can manage thousands of workflows and
maintain millions of historic workflow logs. 

Overview
========

A *Workflow* is given a sequence of programs that follow each other, with an optional
schedule to run the workflow periodically. Upon successful execution of a program, the
control is transferred to the next program in the sequence until the last program in the
sequence is executed. Upon failure, the execution is stopped at the failed program and no
subsequent programs in the sequence are executed.

The control flow of a workflow can be described as a directed, acyclic graph (DAG) of actions.
To be more precise, we require that it be a series-parallel graph. This is a graph with a
single start node and a single finish node. As described below, execution can be either
sequential (the default) or :ref:`concurrent <workflow-concurrent>`, and the graph can be 
either a simple series of nodes or a more complicated :ref:`parallel workflow <workflow_parallel>`.

Workflows can be controlled by the :ref:`CDAP CLI <cli>` and the :ref:`Lifecycle HTTP
RESTful API <http-restful-api-lifecycle>`. The :ref:`status of a workflow
<http-restful-api-lifecycle-status>` can be retrieved, workflows
:ref:`started <http-restful-api-lifecycle-start>` or
:ref:`stopped <http-restful-api-lifecycle-stop>`, and
individual runs of a workflow :ref:`suspended or resumed 
<http-restful-api-lifecycle-workflow-runs-suspend-resume>`. 

A workflow can have one or more than one :ref:`schedules` that call upon it.
These schedules are in a *suspended* state when the application is first deployed.
Each schedule needs to be *resumed*, changing its status to *scheduled*, in order for the
workflow to become executed following the schedule.

Executing MapReduce or Spark Programs
-------------------------------------
To execute MapReduce or Spark programs in a workflow, you will need to add them in your
application along with the workflow. You can optionally add a :ref:`schedule <schedules>` 
to the workflow using the `java api 
<../../reference-manual/javadocs/co/cask/cdap/api/app/AbstractApplication.html#scheduleWorkflow(co.cask.cdap.api.schedule.Schedule,%20java.lang.String)>`__::

  public void configure() {
    ...
    addMapReduce(new MyMapReduce());
    addMapReduce(new AnotherMapReduce());
    addSpark(new MySpark());
    addWorkflow(new MyWorkflow());
    scheduleWorkflow(Schedules.builder("FiveHourSchedule")
                       .setDescription("Schedule running every 5 hours")
                       .createTimeSchedule("0 */5 * * *"),
                     "MyWorkflow");
    ...
  }

You'll then extend the ``AbstractWorkflow`` class and implement the ``configure()``
method. Inside ``configure``, you can add multiple MapReduce, Spark programs or custom
actions to the workflow. The programs will be executed in the order they are specified in
the ``configure`` method::

  public static class MyWorkflow extends AbstractWorkflow {

    @Override
    public void configure() {
        setName("MyWorkflow");
        setDescription("MyWorkflow description");
        addMapReduce("MyMapReduce");
        addSpark("MySpark");
        addMapReduce("AnotherMapReduce");
        addAction(new MyAction());
    }
  }

In this example, the ``MyWorkflow`` will be executed every 5 hours. During each execution
of the workflow, the ``MyMapReduce``, ``MySpark``, and ``AnotherMapReduce`` programs and
the ``MyAction`` custom action will be executed in order.

.. _workflow-custom-actions:

Workflow Custom Action
----------------------
In addition to MapReduce and Spark programs, workflow can also execute custom actions.
Custom actions are implemented in Java and can perform tasks such as sending an email. To
define a custom action, you will need to extend the ``AbstractWorkflowAction`` and
implement the ``run()`` method::

  public static class MyAction extends AbstractWorkflowAction {

    @Override
    public void run() {
      // your code goes here
    }
  }

The custom action then can be added to the workflow using the ``addAction()`` method as
shown above.

.. _workflow-unique-names:

Assigning Unique Names
----------------------
It's important to assign unique names to each component of the workflow, especially when
you use  multiple instances of the same program in the same workflow.

These unique names can be set when the Workflow is first configured, passed to the
instance of the program, and then be used when the program performs its own configuration.

An example of this is the :ref:`Wikipedia Pipeline <examples-wikipedia-data-pipeline>` example, and
its use of the *StreamToDataset* MapReduce program multiple times::

  public class StreamToDataset extends AbstractMapReduce {
    ...
    private final String name;
  
    public StreamToDataset(String name) {
      this.name = name;
    }
  
    @Override
    public void configure() {
      setName(name);
      ...
    }

In its declaration of the application, the example specifies the unique names for each instance::

  public class WikipediaPipelineApp extends AbstractApplication {
    ...
    static final String LIKES_TO_DATASET_MR_NAME = "likesToDataset";
    static final String WIKIPEDIA_TO_DATASET_MR_NAME = "wikiDataToDataset";

    @Override
    public void configure() {
      ...
      addMapReduce(new StreamToDataset(LIKES_TO_DATASET_MR_NAME));
      addMapReduce(new StreamToDataset(WIKIPEDIA_TO_DATASET_MR_NAME));
      ...
      addWorkflow(new WikipediaPipelineWorkflow());
    }
  }

The workflow itself uses the same names in its configuration::

  public class WikipediaPipelineWorkflow extends AbstractWorkflow {

    public static final String NAME = WikipediaPipelineWorkflow.class.getSimpleName();

    @Override
    protected void configure() {
      setName(NAME);
      setDescription("A workflow that demonstrates a typical data pipeline to process Wikipedia data.");
      addMapReduce(WikipediaPipelineApp.LIKES_TO_DATASET_MR_NAME);
      condition(new EnoughDataToProceed())
        .condition(new IsWikipediaSourceOnline())
          .addAction(new DownloadWikiDataAction())
        .otherwise()
          .addMapReduce(WikipediaPipelineApp.WIKIPEDIA_TO_DATASET_MR_NAME)
        .end()
        .addMapReduce(WikiContentValidatorAndNormalizer.NAME)
        .fork()
          .addSpark(SparkWikipediaAnalyzer.NAME)
        .also()
          .addMapReduce(TopNMapReduce.NAME)
        .join()
      .otherwise()
      .end();
    }

.. _workflow-concurrent:

Concurrent Workflows
--------------------
By default, a workflow runs sequentially. Multiple instances of a workflow can be run
concurrently. To enable concurrent runs for a workflow, set its runtime argument
``concurrent.runs.enabled`` to ``true``.


.. _workflow_token:

Workflow Tokens
===============

In addition to passing the control flow from one node to the next, a **workflow token** is
passed, available to each of the programs in the workflow. This allows programs to:

- pass custom data (such as a counter, a status, or an error code) from one program in the 
  workflow to subsequent programs;
- query and set the data in the token;
- fetch the data from the token which was set by a specific node; and
- alter the job configuration based on a key in the token; for example, set a different
  mapper/reducer class or a different input/output dataset for a Spark or MapReduce program.
  
The API is intended to allow appropriate action to be taken in response to the token, including
logging and modifying the conditional execution of the workflow based on the token.
<commit_msg>Update default to concurrent workflows.<commit_after>.. meta::
    :author: Cask Data, Inc.
    :copyright: Copyright © 2014-2015 Cask Data, Inc.

.. _workflows:

=========
Workflows
=========

*Workflows* are used to automate the execution of a series of :ref:`MapReduce
<mapreduce>`, :ref:`Spark <spark>` or :ref:`custom actions <workflow-custom-actions>`. It
allows for both sequential and :ref:`parallel execution <workflow_parallel>` of programs.

The workflow system allows specifying, executing, scheduling, and monitoring complex
series of jobs and tasks in CDAP. The system can manage thousands of workflows and
maintain millions of historic workflow logs. 

Overview
========

A *Workflow* is given a sequence of programs that follow each other, with an optional
schedule to run the workflow periodically. Upon successful execution of a program, the
control is transferred to the next program in the sequence until the last program in the
sequence is executed. Upon failure, the execution is stopped at the failed program and no
subsequent programs in the sequence are executed.

The control flow of a workflow can be described as a directed, acyclic graph (DAG) of actions.
To be more precise, we require that it be a series-parallel graph. This is a graph with a
single start node and a single finish node. As described below, execution can be either
sequential or :ref:`concurrent <workflow-concurrent>` (the default), and the graph can be 
either a simple series of nodes or a more complicated :ref:`parallel workflow <workflow_parallel>`.

Workflows can be controlled by the :ref:`CDAP CLI <cli>` and the :ref:`Lifecycle HTTP
RESTful API <http-restful-api-lifecycle>`. The :ref:`status of a workflow
<http-restful-api-lifecycle-status>` can be retrieved, workflows
:ref:`started <http-restful-api-lifecycle-start>` or
:ref:`stopped <http-restful-api-lifecycle-stop>`, and
individual runs of a workflow :ref:`suspended or resumed 
<http-restful-api-lifecycle-workflow-runs-suspend-resume>`. 

A workflow can have one or more than one :ref:`schedules` that call upon it.
These schedules are in a *suspended* state when the application is first deployed.
Each schedule needs to be *resumed*, changing its status to *scheduled*, in order for the
workflow to become executed following the schedule.

Executing MapReduce or Spark Programs
-------------------------------------
To execute MapReduce or Spark programs in a workflow, you will need to add them in your
application along with the workflow. You can optionally add a :ref:`schedule <schedules>` 
to the workflow using the `java api 
<../../reference-manual/javadocs/co/cask/cdap/api/app/AbstractApplication.html#scheduleWorkflow(co.cask.cdap.api.schedule.Schedule,%20java.lang.String)>`__::

  public void configure() {
    ...
    addMapReduce(new MyMapReduce());
    addMapReduce(new AnotherMapReduce());
    addSpark(new MySpark());
    addWorkflow(new MyWorkflow());
    scheduleWorkflow(Schedules.builder("FiveHourSchedule")
                       .setDescription("Schedule running every 5 hours")
                       .createTimeSchedule("0 */5 * * *"),
                     "MyWorkflow");
    ...
  }

You'll then extend the ``AbstractWorkflow`` class and implement the ``configure()``
method. Inside ``configure``, you can add multiple MapReduce, Spark programs or custom
actions to the workflow. The programs will be executed in the order they are specified in
the ``configure`` method::

  public static class MyWorkflow extends AbstractWorkflow {

    @Override
    public void configure() {
        setName("MyWorkflow");
        setDescription("MyWorkflow description");
        addMapReduce("MyMapReduce");
        addSpark("MySpark");
        addMapReduce("AnotherMapReduce");
        addAction(new MyAction());
    }
  }

In this example, the ``MyWorkflow`` will be executed every 5 hours. During each execution
of the workflow, the ``MyMapReduce``, ``MySpark``, and ``AnotherMapReduce`` programs and
the ``MyAction`` custom action will be executed in order.

.. _workflow-custom-actions:

Workflow Custom Action
----------------------
In addition to MapReduce and Spark programs, workflow can also execute custom actions.
Custom actions are implemented in Java and can perform tasks such as sending an email. To
define a custom action, you will need to extend the ``AbstractWorkflowAction`` and
implement the ``run()`` method::

  public static class MyAction extends AbstractWorkflowAction {

    @Override
    public void run() {
      // your code goes here
    }
  }

The custom action then can be added to the workflow using the ``addAction()`` method as
shown above.

.. _workflow-unique-names:

Assigning Unique Names
----------------------
It's important to assign unique names to each component of the workflow, especially when
you use  multiple instances of the same program in the same workflow.

These unique names can be set when the Workflow is first configured, passed to the
instance of the program, and then be used when the program performs its own configuration.

An example of this is the :ref:`Wikipedia Pipeline <examples-wikipedia-data-pipeline>` example, and
its use of the *StreamToDataset* MapReduce program multiple times::

  public class StreamToDataset extends AbstractMapReduce {
    ...
    private final String name;
  
    public StreamToDataset(String name) {
      this.name = name;
    }
  
    @Override
    public void configure() {
      setName(name);
      ...
    }

In its declaration of the application, the example specifies the unique names for each instance::

  public class WikipediaPipelineApp extends AbstractApplication {
    ...
    static final String LIKES_TO_DATASET_MR_NAME = "likesToDataset";
    static final String WIKIPEDIA_TO_DATASET_MR_NAME = "wikiDataToDataset";

    @Override
    public void configure() {
      ...
      addMapReduce(new StreamToDataset(LIKES_TO_DATASET_MR_NAME));
      addMapReduce(new StreamToDataset(WIKIPEDIA_TO_DATASET_MR_NAME));
      ...
      addWorkflow(new WikipediaPipelineWorkflow());
    }
  }

The workflow itself uses the same names in its configuration::

  public class WikipediaPipelineWorkflow extends AbstractWorkflow {

    public static final String NAME = WikipediaPipelineWorkflow.class.getSimpleName();

    @Override
    protected void configure() {
      setName(NAME);
      setDescription("A workflow that demonstrates a typical data pipeline to process Wikipedia data.");
      addMapReduce(WikipediaPipelineApp.LIKES_TO_DATASET_MR_NAME);
      condition(new EnoughDataToProceed())
        .condition(new IsWikipediaSourceOnline())
          .addAction(new DownloadWikiDataAction())
        .otherwise()
          .addMapReduce(WikipediaPipelineApp.WIKIPEDIA_TO_DATASET_MR_NAME)
        .end()
        .addMapReduce(WikiContentValidatorAndNormalizer.NAME)
        .fork()
          .addSpark(SparkWikipediaAnalyzer.NAME)
        .also()
          .addMapReduce(TopNMapReduce.NAME)
        .join()
      .otherwise()
      .end();
    }

.. _workflow-concurrent:

Concurrent Workflows
--------------------
By default, a workflow runs concurrently, allowing multiple instances of a workflow to be run
simultaneously. To disable concurrent runs for a workflow, set its runtime argument
``concurrent.runs.enabled`` to ``false``.


.. _workflow_token:

Workflow Tokens
===============

In addition to passing the control flow from one node to the next, a **workflow token** is
passed, available to each of the programs in the workflow. This allows programs to:

- pass custom data (such as a counter, a status, or an error code) from one program in the 
  workflow to subsequent programs;
- query and set the data in the token;
- fetch the data from the token which was set by a specific node; and
- alter the job configuration based on a key in the token; for example, set a different
  mapper/reducer class or a different input/output dataset for a Spark or MapReduce program.
  
The API is intended to allow appropriate action to be taken in response to the token, including
logging and modifying the conditional execution of the workflow based on the token.

{'fraction_non_alphanumeric': 0.091930729550479, 'fraction_numerical': 0.0017194792434291329, 'mean_word_length': 3.70121247113164, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '137769', 'n_tokens_mistral': 4442, 'n_tokens_neox': 4218, 'n_words': 1815}