Mailing-List: contact dev-help@apex.incubator.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@apex.incubator.apache.org
MIME-Version: 1.0
Date: Wed, 16 Sep 2015 13:44:26 -0700
Message-ID: 
 <CAMqituPxN1f0ZQ9h5PmRg+SBKSa2Wdk3KQ8UY5O7NPRnw5Jvcw@mail.gmail.com>
Subject: Supporting iterations in Apex
From: David Yan <david@datatorrent.com>
To: dev@apex.incubator.apache.org
Content-Type: multipart/alternative; boundary=001a113433d64303f5051fe35da3

--001a113433d64303f5051fe35da3
Content-Type: text/plain; charset=UTF-8

Hi all,

One current disadvantage of Apex is the inability to do iterations and
machine learning algorithms because we don't allow loops in the application
DAG (hence the name DAG).  I am proposing that we allow loops in the DAG if
the loop advances the window ID by a configured amount.  A JIRA ticket has
been created:

https://malhar.atlassian.net/browse/APEX-60

I have started this work in my fork at
https://github.com/davidyan74/incubator-apex-core/tree/APEX-60.

The current progress is that a simple test case works.  Major work still
needs to be done with respect to recovery and partitioning.

The value ITERATION_WINDOW_COUNT is an attribute to an input port of an
operator.  If the value of the attribute is greater than or equal to 1, any
tuples sent to the input port are treated to be ITERATION_WINDOW_COUNT
windows ahead of what they are.

For recovery, we will need to checkpoint all the tuples between ports with
the to replay the looped tuples.  During the recovery, if the operator has
an input port, with ITERATION_WINDOW_COUNT=2, is recovering from checkpoint
window 14, the tuples for that input port from window 13 and window 14 need
to be replayed to be treated as window 15 and window 16 respectively (13+2
and 14+2).

In other words, we need to store all the tuples from window with ID
committedWindowId minus ITERATION_WINDOW_COUNT for recovery and purge the
tuples earlier than that window.
We can optimize this by only storing the tuples for ITERATION_WINDOW_COUNT
windows prior to any checkpoint.

For that, we need a storage mechanism for the tuples.  Chandni already has
something that fits this usage case in Apex Malhar.  The class is
IdempotentStorageManager.  In order for this to be used in Apex core, we
need to deprecate the class in Apex Malhar and move it to Apex Core.

A JIRA ticket has been created for this particular work:

https://malhar.atlassian.net/browse/APEX-128

Some of the above has been discussed among Thomas, Chetan, Chandni, and
myself.

For partitioning, we have not started any discussion or brainstorming.  We
appreciate any feedback on this and any other aspect related to supporting
iterations in general.

Thanks!

David

--001a113433d64303f5051fe35da3--