apex-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Pramod Immaneni (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (APEXCORE-348) Load based stream partitioning
Date Wed, 17 Feb 2016 20:53:18 GMT

    [ https://issues.apache.org/jira/browse/APEXCORE-348?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15151156#comment-15151156
] 

Pramod Immaneni commented on APEXCORE-348:
------------------------------------------

Conversation from email

Pramod Immaneni <pramod@datatorrent.com>
Inline

On Thu, Feb 11, 2016 at 4:32 PM, Timothy Farkas <tim@datatorrent.com> wrote:
Comments inline

+1 Overall as well provided Apex-339 is implemented first and it is
documented that the mechanism should not be used with some stateful
operators.

On Thu, Feb 11, 2016 at 4:20 PM, Pramod Immaneni <pramod@datatorrent.com>
wrote:

> Comments inline
>
> On Thu, Feb 11, 2016 at 4:13 PM, Timothy Farkas <tim@datatorrent.com>
> wrote:
>
> > Hey Pramod,
> >
> > I agree if APEX-339 is in place then it would work without redeploying
> > containers for operators that are Stateless, or a subset of Stateful
> > operators.
> >
> > Addressing your previous questions.
> >
> > - The StatsListener can be used to see how far behind operators are. You
> > could determine what window the operator is on, or the number of tuples
> > it's processed so far, or how long
> > it takes it to complete a window.
> >
>
> What if tuples are different sizes and number of tuples processed doesn't
> reflect how far ahead or behind a downstream partition is? How is the
> information from StatsListener made available to the upstream partition
> codecs.
>
What is the information Buffer Server can provide that the StatsListener
cannot?

The stats information would have to be relayed down to the upstream operators. It's possible.
 

The StatsListener can trigger a repartition. The information in the
StatsListener can be shared
with the partitioner by setting the same object for both in populate Dag.
The partitioner can then
compute the new Stream Codec. The mechanism by which the upstream would be
updated with the new
Stream Codec would have to be implemented as it's currently not there.

>
>
> >
> > - Some examples of Stateful operators that require repartitioning of
> state
> > are the following:
> >       - Deduper
> >            In this case after updating the stream codec the operator may
> > allow a previously seen value to pass because the partition didn't
> receive
> > that value with the previous stream codec.
> >       - A key value store that holds aggregations for each key.
> >            In this case multiple partitions would hold partial
> aggregations
> > for a key, when they are expecting to hold the complete aggregation.
> >
>
> Agreed for deduper. For the second case a unifier is a better approach so
> that you are not affected by key skew in general.
>
This is not always possible. We can discuss this offline, since it won't
add much to the discussion here to go into the details.


Yes not always.


> Load based stream partitioning
> ------------------------------
>
>                 Key: APEXCORE-348
>                 URL: https://issues.apache.org/jira/browse/APEXCORE-348
>             Project: Apache Apex Core
>          Issue Type: Improvement
>            Reporter: Pramod Immaneni
>            Assignee: Pramod Immaneni
>
> There are scenarios where the downstream partitions of an upstream operator are generally
not performing uniformly resulting in an overall sub-optimal performance dictated by the slowest
partitions. The reasons could be data related such as some partitions are receiving more data
to process than the others or could be environment related such as some partitions are running
slower than others because they are on heavily loaded nodes.
> A solution based on currently available functionality in the engine would be to write
a StreamCodec implementation to distribute data among the partitions such that each partition
is receiving similar amount of data to process. We should consider adding StreamCodecs like
these to the library but these however do not solve the problem when it is environment related.
> For that a better and more comprehensive approach would be look at how data is being
consumed by the downstream partitions from the BufferServer and use that information to make
decisions on how to send future data. If some partitions are behind others in consuming data
then data can be directed to the other partitions. One way to do this would be to relay this
type of statistical and positional information from BufferServer to the upstream publishers.
The publishers can use this information in ways such as making it available to StreamCodecs
to affect destination of future data.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message