You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@apex.apache.org by Bhupesh Chawda <bh...@datatorrent.com> on 2015/09/15 09:38:29 UTC
Sequencing of operator calls
Hi All,
Is it safe to assume that the checkpointed() and the beginWindow() calls
are sequenced?
In other words, are these calls part of the same thread and may not run in
parallel?
Thanks.
--
-Bhupesh
Re: Sequencing of operator calls
Posted by Bhupesh Chawda <bh...@datatorrent.com>.
Just saw the reply from Chetan.
Thanks, I think that makes everything a lot clearer :-)
-Bhupesh
Chetan's answer provides a good explanation as well as clarifying that
the difference can be more than 1.
Since checkpointing (i.e. "commit notification" as Thomas refers to
it) is asynchronous, I'm curious
about whether the window ids in the checkpointed call are guaranteed
to be sequential or if they could
be out of order, i.e. can the checkpointed call see window id 101
before it sees 100 ?
Ram
On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
<bh...@datatorrent.com> wrote:
> Hi Tim,
> Thanks for the detailed explanation.
> I understand that the sequence would be
> beginWindow (x) -> endWindow (x) -> checkpointed (x) -> beginWindow
> (x+1)
>
> However when I try to print out the window ids in beginWindow, endWindow
> and checkpointed calls, I see x and x-1 respectively.
> I.e. If the window just before checkpoint is 100, I see that the
> checkpointed call had window id 99.
>
> Note: This is observed in the local mode of Apex.
>
> Thanks
> -Bhupesh
> On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com> wrote:
>
>> Hi Bhupesh,
>>
>> The sequencing of checkpoint called in relation to beginWindow and
>> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are not
>> configured to be the same then it is possible that checkpointed is called
>> within an application window. So the sequence of events would be this:
>>
>> beginWindow -> checkpointed -> endWindow
>>
>> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the same
>> (default). Then the sequence of calls would be this:
>>
>> beginWindow (100) -> endWindow (100) -> checkpointed (100) ->
beginWindow
>> (101)
>>
>> The numbers in the sequence represent possible streaming window Ids that
>> each call would be associated with.
>>
>> The StateMachine which calls these callbacks for non-input operators is
in
>> GenericNode.java.
>>
>> Tim
>>
>> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <bh...@datatorrent.com>
>> wrote:
>>
>> > Hi Chetan / Community,
>> >
>> > Can someone please elaborate on why the window id supplied to
>> > CheckpointListener and the Operator would differ.
>> > I tried looking at the window ids of checkpointed() and the
beginWindow()
>> > calls and they differ by 1. Don't know why this should be the case.
>> >
>> > Thanks.
>> > -Bhupesh
>> >
>> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <chetan@datatorrent.com
>
>> > wrote:
>> >
>> > > Short answer is yes.
>> > >
>> > > All the control tuples are scheduled to be delivered outside of the
>> > window.
>> > > As checkpointed callback is triggered because of CHECKPOINT control
>> > tuple,
>> > > it will happen after endWindow and before the next beginWindow.
>> > >
>> > > The windowId supplied to CheckpointListener and the one provided to
>> > > Operator need not match even though the sequence is defined. So I am
>> > > curious how you intend to use this knowledge.
>> > >
>> > > --
>> > > Chetan
>> > >
>> > >
>> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <thomas@datatorrent.com
>
>> > > wrote:
>> > >
>> > > > It has not changed the operator execution model. State
serialization
>> is
>> > > > still synchronous, write to HDFS is taken out of the operator
thread.
>> > > >
>> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
>> > > wrote:
>> > > >
>> > > > >
>> > > > > Sent too soon. Has asynchronous checkpointing changed this?
>> > > > >
>> > > > > Amol
>> > > > >
>> > > > > Sent from my iPhone
>> > > > >
>> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>> > > bhupesh@datatorrent.com>
>> > > > > wrote:
>> > > > > >
>> > > > > > Hi All,
>> > > > > >
>> > > > > > Is it safe to assume that the checkpointed() and the
>> beginWindow()
>> > > > calls
>> > > > > > are sequenced?
>> > > > > > In other words, are these calls part of the same thread and may
>> not
>> > > run
>> > > > > in
>> > > > > > parallel?
>> > > > > >
>> > > > > > Thanks.
>> > > > > >
>> > > > > > --
>> > > > > > -Bhupesh
>> > > > >
>> > > >
>> > >
>> >
>>
Re: Sequencing of operator calls
Posted by "Chetan Narsude (cnarsude)" <cn...@cisco.com>.
Checkpoint notification is different from commit notification.
Checkpointed is called when the state is preserved. Committed is called
when the preserved state is no longer needed. That happens when the
checkpointed windowId is processed by all the operators in the application.
The checkpoint windowIds supplied to an operator are guaranteed to be
monotonically increasing during operator instances life cycle (as the
windowIds themselves are). If there is a failure, a new instance life
cycle begins in which case there would be one time rewind on windowId and
hence everything that depends on it.
‹
Chetan
On 11/10/15, 10:33 AM, "Munagala Ramanath" <ra...@datatorrent.com> wrote:
>Chetan's answer provides a good explanation as well as clarifying that
>the difference can be more than 1.
>
>Since checkpointing (i.e. "commit notification" as Thomas refers to
>it) is asynchronous, I'm curious
>about whether the window ids in the checkpointed call are guaranteed
>to be sequential or if they could
>be out of order, i.e. can the checkpointed call see window id 101
>before it sees 100 ?
>
>Ram
>
>On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
><bh...@datatorrent.com> wrote:
>> Hi Tim,
>> Thanks for the detailed explanation.
>> I understand that the sequence would be
>> beginWindow (x) -> endWindow (x) -> checkpointed (x) -> beginWindow
>> (x+1)
>>
>> However when I try to print out the window ids in beginWindow, endWindow
>> and checkpointed calls, I see x and x-1 respectively.
>> I.e. If the window just before checkpoint is 100, I see that the
>> checkpointed call had window id 99.
>>
>> Note: This is observed in the local mode of Apex.
>>
>> Thanks
>> -Bhupesh
>> On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com> wrote:
>>
>>> Hi Bhupesh,
>>>
>>> The sequencing of checkpoint called in relation to beginWindow and
>>> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>>> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are
>>>not
>>> configured to be the same then it is possible that checkpointed is
>>>called
>>> within an application window. So the sequence of events would be this:
>>>
>>> beginWindow -> checkpointed -> endWindow
>>>
>>> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the
>>>same
>>> (default). Then the sequence of calls would be this:
>>>
>>> beginWindow (100) -> endWindow (100) -> checkpointed (100) ->
>>>beginWindow
>>> (101)
>>>
>>> The numbers in the sequence represent possible streaming window Ids
>>>that
>>> each call would be associated with.
>>>
>>> The StateMachine which calls these callbacks for non-input operators
>>>is in
>>> GenericNode.java.
>>>
>>> Tim
>>>
>>> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda
>>><bh...@datatorrent.com>
>>> wrote:
>>>
>>> > Hi Chetan / Community,
>>> >
>>> > Can someone please elaborate on why the window id supplied to
>>> > CheckpointListener and the Operator would differ.
>>> > I tried looking at the window ids of checkpointed() and the
>>>beginWindow()
>>> > calls and they differ by 1. Don't know why this should be the case.
>>> >
>>> > Thanks.
>>> > -Bhupesh
>>> >
>>> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude
>>><ch...@datatorrent.com>
>>> > wrote:
>>> >
>>> > > Short answer is yes.
>>> > >
>>> > > All the control tuples are scheduled to be delivered outside of the
>>> > window.
>>> > > As checkpointed callback is triggered because of CHECKPOINT control
>>> > tuple,
>>> > > it will happen after endWindow and before the next beginWindow.
>>> > >
>>> > > The windowId supplied to CheckpointListener and the one provided to
>>> > > Operator need not match even though the sequence is defined. So I
>>>am
>>> > > curious how you intend to use this knowledge.
>>> > >
>>> > > --
>>> > > Chetan
>>> > >
>>> > >
>>> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise
>>><th...@datatorrent.com>
>>> > > wrote:
>>> > >
>>> > > > It has not changed the operator execution model. State
>>>serialization
>>> is
>>> > > > still synchronous, write to HDFS is taken out of the operator
>>>thread.
>>> > > >
>>> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre
>>><am...@datatorrent.com>
>>> > > wrote:
>>> > > >
>>> > > > >
>>> > > > > Sent too soon. Has asynchronous checkpointing changed this?
>>> > > > >
>>> > > > > Amol
>>> > > > >
>>> > > > > Sent from my iPhone
>>> > > > >
>>> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>>> > > bhupesh@datatorrent.com>
>>> > > > > wrote:
>>> > > > > >
>>> > > > > > Hi All,
>>> > > > > >
>>> > > > > > Is it safe to assume that the checkpointed() and the
>>> beginWindow()
>>> > > > calls
>>> > > > > > are sequenced?
>>> > > > > > In other words, are these calls part of the same thread and
>>>may
>>> not
>>> > > run
>>> > > > > in
>>> > > > > > parallel?
>>> > > > > >
>>> > > > > > Thanks.
>>> > > > > >
>>> > > > > > --
>>> > > > > > -Bhupesh
>>> > > > >
>>> > > >
>>> > >
>>> >
>>>
Re: Sequencing of operator calls
Posted by Bhupesh Chawda <bh...@datatorrent.com>.
Yes, there are situations where we can do it in endWindow.
However there are also cases where we would like to do the processing only
in the endWindow which is just before the checkpoint. Such cases, I assume
would benefit from https://malhar.atlassian.net/browse/APEX-78.
Thanks.
-Bhupesh
On 11-Nov-2015 1:03 pm, "Thomas Weise" <th...@datatorrent.com> wrote:
> What you are looking for would require the guarantee for checkpointed to be
> called before next beginWindow.
>
> Why don't you record the list in endWindow? Is it a performance concern?
>
> This would then benefit from https://malhar.atlassian.net/browse/APEX-78
>
>
>
> On Tue, Nov 10, 2015 at 11:20 PM, Bhupesh Chawda <bh...@datatorrent.com>
> wrote:
>
> > I was trying to do some processing between two checkpoint windows.
> >
> > So, for example, an operator deletes some files from hdfs as part of
> normal
> > processing within an application window. However, since hdfs deletes
> cannot
> > be rolled back by the platform, I am trying NOT to delete the files from
> > hdfs but just record them in a list. Now when a checkpoint happens, I
> need
> > to take a snapshot of the list *as of the checkpoint* and later delete
> them
> > asynchronously. This process of taking the snapshot is what I need to do
> > between two checkpoint windows. This can be done anytime between the
> > endWindow just before the checkpoint and beginWindow just after the
> > checkpoint.
> >
> > So the question is, can this be done in the checkpointed callback?
> >
> > If the callback is asynchronous, i. e. It may be called after the next
> > window starts, then probably checkpointed is not the right place to do
> it.
> > On 11-Nov-2015 11:20 am, "Thomas Weise" <th...@datatorrent.com> wrote:
> >
> > > There does not seem to be a problem with the order of the checkpointed
> > > callbacks.
> > >
> > > I would however question why the callback is delayed until write to
> > storage
> > > is complete. From operator perspective, it can be called as soon as
> > > serialization is complete, which is always synchronous.
> > >
> > > The checkpoint cannot be reported to StrAM until the file copy is done,
> > as
> > > currently implemented.
> > >
> > > Bhupesh, what problem were you looking to solve through the
> checkpointed
> > > callback?
> > >
> > >
> > >
> > >
> > > On Tue, Nov 10, 2015 at 3:57 PM, Chetan Narsude <ch...@apache.org>
> > wrote:
> > >
> > > > With async checkpointing the Node.reportStats is reporting it back
> and
> > it
> > > > looks like the code is checking for the top of the queue to see if
> it’s
> > > > done (or not reporting at all). So I do not see a reason why they
> will
> > be
> > > > reported out of order.
> > > >
> > > > Chandni, I read your subsequent responses. Valid point in the last
> > email
> > > > about documentation. It amazes me how much the documentation of our
> API
> > > > can be improved (self guilty probably the most). One comment that I
> > > wanted
> > > > to make even before that is that the semantics of copyToHdfs is that
> it
> > > > either succeeds or throws exception in which case the recovery kicks
> in
> > > > and nothing is reported as checkpointed.
> > > >
> > > > Tim, we need the unit test, man!
> > > >
> > > > —
> > > > Chetan
> > > >
> > > >
> > > > On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <
> cnarsude@cisco.com>
> > > > wrote:
> > > >
> > > > >There are a lot of things which are different when it comes to async
> > > > >checkpointing. I was evaluating it in the morning and expect that
> > > either I
> > > > >am able to explain or open jira issues. With my partial observation
> is
> > > > >that with Async checkpointing, checkpointed is not issued (chandni,
> > the
> > > > >last statement in the if block is ³return²). I am digging into it
> but
> > > feel
> > > > >free to chime in if someone else is able to find that.
> > > > >
> > > > >Also I realized that my morning email applies as it is to committed
> > but
> > > > >checkpointed has deviated a little bit from that. Will post the
> > revised
> > > > >response soon.
> > > > >
> > > > >‹
> > > > >Chetan
> > > > >
> > > > >
> > > > >
> > > > >
> > > > >On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com>
> > wrote:
> > > > >
> > > > >>Chetan,
> > > > >>
> > > > >>Looking at the checkpoint(windowId) in Node.java, I don't think the
> > > steps
> > > > >>you mentioned are followed.
> > > > >>
> > > > >>*if (using AsyncFSStorageAgent) {*
> > > > >>* asyncFSStorageAgent.copyToHdfs(...)*
> > > > >>*}*
> > > > >>*operator.checkpointed(windowId);*
> > > > >>
> > > > >>This means even copyToHdfs fails the operator is notified that the
> > > window
> > > > >>is check-pointed.
> > > > >>
> > > > >>Are we saying that copyToHdfs will never fail with
> > AsyncFSStorageAgent
> > > > >>for
> > > > >>a window since the operator is notified that the window is
> > > checkpointed?
> > > > >>
> > > > >>Chandni
> > > > >>
> > > > >>On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <
> > tim@datatorrent.com>
> > > > >>wrote:
> > > > >>
> > > > >>> Will do
> > > > >>>
> > > > >>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
> > > > >>><pr...@datatorrent.com>
> > > > >>> wrote:
> > > > >>>
> > > > >>> > Is there a unit test covering it? Otherwise can you write one
> to
> > > test
> > > > >>>the
> > > > >>> > hypothesis.
> > > > >>> >
> > > > >>> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
> > > > >>><ti...@datatorrent.com>
> > > > >>> > wrote:
> > > > >>> >
> > > > >>> > > That is what it is looking like to me. The task is submitted
> > > > >>> > > GenericNode#checkpoint line 504, then at the end of the
> > > > >>> > > GenericNode#checkpoint line 531 checkpointed is called. I am
> > > likely
> > > > >>> > missing
> > > > >>> > > something, just would like to know what :)
> > > > >>> > >
> > > > >>> > > Tim
> > > > >>> > >
> > > > >>> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> > > > >>> > pramod@datatorrent.com>
> > > > >>> > > wrote:
> > > > >>> > >
> > > > >>> > > > Tim,
> > > > >>> > > >
> > > > >>> > > > Are you suggesting that checkpointed is called before the
> > > > >>>checkpoint
> > > > >>> is
> > > > >>> > > > completely persisted in the storage.
> > > > >>> > > >
> > > > >>> > > > Thanks
> > > > >>> > > >
> > > > >>> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> > > > >>> tim@datatorrent.com>
> > > > >>> > > > wrote:
> > > > >>> > > >
> > > > >>> > > > > Chetan,
> > > > >>> > > > >
> > > > >>> > > > > I do not see the process of reporting the checkpoint to
> > > stram,
> > > > >>> > > receiving
> > > > >>> > > > > the ack, and then calling checkpointed. The logic I'm
> > seeing
> > > in
> > > > >>> > > > GenericNode
> > > > >>> > > > > line 484 is that the checkpoint method is called, it
> spawns
> > > > >>>another
> > > > >>> > > > thread
> > > > >>> > > > > that writes to hdfs, and then checkpointed is immediately
> > > > >>>called
> > > > >>> > > > > afterwards. I am missing something, can you give me some
> > > > >>>pointers
> > > > >>> so
> > > > >>> > > > that I
> > > > >>> > > > > can better understand the flow?
> > > > >>> > > > >
> > > > >>> > > > > Tim
> > > > >>> > > > >
> > > > >>> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> > > > >>> > > ram@datatorrent.com
> > > > >>> > > > >
> > > > >>> > > > > wrote:
> > > > >>> > > > >
> > > > >>> > > > > > Chetan's answer provides a good explanation as well as
> > > > >>>clarifying
> > > > >>> > > that
> > > > >>> > > > > > the difference can be more than 1.
> > > > >>> > > > > >
> > > > >>> > > > > > Since checkpointing (i.e. "commit notification" as
> Thomas
> > > > >>>refers
> > > > >>> to
> > > > >>> > > > > > it) is asynchronous, I'm curious
> > > > >>> > > > > > about whether the window ids in the checkpointed call
> are
> > > > >>> > guaranteed
> > > > >>> > > > > > to be sequential or if they could
> > > > >>> > > > > > be out of order, i.e. can the checkpointed call see
> > window
> > > id
> > > > >>>101
> > > > >>> > > > > > before it sees 100 ?
> > > > >>> > > > > >
> > > > >>> > > > > > Ram
> > > > >>> > > > > >
> > > > >>> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > > > >>> > > > > > <bh...@datatorrent.com> wrote:
> > > > >>> > > > > > > Hi Tim,
> > > > >>> > > > > > > Thanks for the detailed explanation.
> > > > >>> > > > > > > I understand that the sequence would be
> > > > >>> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x)
> > ->
> > > > >>> > > beginWindow
> > > > >>> > > > > > > (x+1)
> > > > >>> > > > > > >
> > > > >>> > > > > > > However when I try to print out the window ids in
> > > > >>>beginWindow,
> > > > >>> > > > > endWindow
> > > > >>> > > > > > > and checkpointed calls, I see x and x-1
> respectively.
> > > > >>> > > > > > > I.e. If the window just before checkpoint is 100, I
> see
> > > > >>>that
> > > > >>> the
> > > > >>> > > > > > > checkpointed call had window id 99.
> > > > >>> > > > > > >
> > > > >>> > > > > > > Note: This is observed in the local mode of Apex.
> > > > >>> > > > > > >
> > > > >>> > > > > > > Thanks
> > > > >>> > > > > > > -Bhupesh
> > > > >>> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
> > > > >>><tim@datatorrent.com
> > > > >>> >
> > > > >>> > > > wrote:
> > > > >>> > > > > > >
> > > > >>> > > > > > >> Hi Bhupesh,
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> The sequencing of checkpoint called in relation to
> > > > >>>beginWindow
> > > > >>> > and
> > > > >>> > > > > > >> endWindow depends on how your
> APPLICATION_WINDOW_COUNT
> > > and
> > > > >>> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
> > > > >>> WINDOW_COUNTs
> > > > >>> > > are
> > > > >>> > > > > not
> > > > >>> > > > > > >> configured to be the same then it is possible that
> > > > >>> checkpointed
> > > > >>> > is
> > > > >>> > > > > > called
> > > > >>> > > > > > >> within an application window. So the sequence of
> > events
> > > > >>>would
> > > > >>> be
> > > > >>> > > > this:
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> beginWindow -> checkpointed -> endWindow
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> If the APPLICATION_WINDOW_COUNT and
> > > > >>>CHECKPOINT_WINDOW_COUNT
> > > > >>> are
> > > > >>> > > the
> > > > >>> > > > > same
> > > > >>> > > > > > >> (default). Then the sequence of calls would be this:
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> beginWindow (100) -> endWindow (100) ->
> checkpointed
> > > > >>>(100)
> > > > >>> ->
> > > > >>> > > > > > beginWindow
> > > > >>> > > > > > >> (101)
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> The numbers in the sequence represent possible
> > streaming
> > > > >>> window
> > > > >>> > > Ids
> > > > >>> > > > > that
> > > > >>> > > > > > >> each call would be associated with.
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> The StateMachine which calls these callbacks for
> > > non-input
> > > > >>> > > operators
> > > > >>> > > > > is
> > > > >>> > > > > > in
> > > > >>> > > > > > >> GenericNode.java.
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> Tim
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > > > >>> > > > > > bhupesh@datatorrent.com>
> > > > >>> > > > > > >> wrote:
> > > > >>> > > > > > >>
> > > > >>> > > > > > >> > Hi Chetan / Community,
> > > > >>> > > > > > >> >
> > > > >>> > > > > > >> > Can someone please elaborate on why the window id
> > > > >>>supplied
> > > > >>> to
> > > > >>> > > > > > >> > CheckpointListener and the Operator would differ.
> > > > >>> > > > > > >> > I tried looking at the window ids of
> checkpointed()
> > > and
> > > > >>>the
> > > > >>> > > > > > beginWindow()
> > > > >>> > > > > > >> > calls and they differ by 1. Don't know why this
> > should
> > > > >>>be
> > > > >>> the
> > > > >>> > > > case.
> > > > >>> > > > > > >> >
> > > > >>> > > > > > >> > Thanks.
> > > > >>> > > > > > >> > -Bhupesh
> > > > >>> > > > > > >> >
> > > > >>> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > > > >>> > > > > > chetan@datatorrent.com>
> > > > >>> > > > > > >> > wrote:
> > > > >>> > > > > > >> >
> > > > >>> > > > > > >> > > Short answer is yes.
> > > > >>> > > > > > >> > >
> > > > >>> > > > > > >> > > All the control tuples are scheduled to be
> > delivered
> > > > >>> outside
> > > > >>> > > of
> > > > >>> > > > > the
> > > > >>> > > > > > >> > window.
> > > > >>> > > > > > >> > > As checkpointed callback is triggered because of
> > > > >>> CHECKPOINT
> > > > >>> > > > > control
> > > > >>> > > > > > >> > tuple,
> > > > >>> > > > > > >> > > it will happen after endWindow and before the
> next
> > > > >>> > > beginWindow.
> > > > >>> > > > > > >> > >
> > > > >>> > > > > > >> > > The windowId supplied to CheckpointListener and
> > the
> > > > >>>one
> > > > >>> > > provided
> > > > >>> > > > > to
> > > > >>> > > > > > >> > > Operator need not match even though the sequence
> > is
> > > > >>> defined.
> > > > >>> > > So
> > > > >>> > > > I
> > > > >>> > > > > am
> > > > >>> > > > > > >> > > curious how you intend to use this knowledge.
> > > > >>> > > > > > >> > >
> > > > >>> > > > > > >> > > --
> > > > >>> > > > > > >> > > Chetan
> > > > >>> > > > > > >> > >
> > > > >>> > > > > > >> > >
> > > > >>> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > > > >>> > > > > > thomas@datatorrent.com>
> > > > >>> > > > > > >> > > wrote:
> > > > >>> > > > > > >> > >
> > > > >>> > > > > > >> > > > It has not changed the operator execution
> model.
> > > > >>>State
> > > > >>> > > > > > serialization
> > > > >>> > > > > > >> is
> > > > >>> > > > > > >> > > > still synchronous, write to HDFS is taken out
> of
> > > the
> > > > >>> > > operator
> > > > >>> > > > > > thread.
> > > > >>> > > > > > >> > > >
> > > > >>> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > > > >>> > > > > amol@datatorrent.com
> > > > >>> > > > > > >
> > > > >>> > > > > > >> > > wrote:
> > > > >>> > > > > > >> > > >
> > > > >>> > > > > > >> > > > >
> > > > >>> > > > > > >> > > > > Sent too soon. Has asynchronous
> checkpointing
> > > > >>>changed
> > > > >>> > > this?
> > > > >>> > > > > > >> > > > >
> > > > >>> > > > > > >> > > > > Amol
> > > > >>> > > > > > >> > > > >
> > > > >>> > > > > > >> > > > > Sent from my iPhone
> > > > >>> > > > > > >> > > > >
> > > > >>> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh
> > Chawda <
> > > > >>> > > > > > >> > > bhupesh@datatorrent.com>
> > > > >>> > > > > > >> > > > > wrote:
> > > > >>> > > > > > >> > > > > >
> > > > >>> > > > > > >> > > > > > Hi All,
> > > > >>> > > > > > >> > > > > >
> > > > >>> > > > > > >> > > > > > Is it safe to assume that the
> checkpointed()
> > > and
> > > > >>>the
> > > > >>> > > > > > >> beginWindow()
> > > > >>> > > > > > >> > > > calls
> > > > >>> > > > > > >> > > > > > are sequenced?
> > > > >>> > > > > > >> > > > > > In other words, are these calls part of
> the
> > > same
> > > > >>> > thread
> > > > >>> > > > and
> > > > >>> > > > > > may
> > > > >>> > > > > > >> not
> > > > >>> > > > > > >> > > run
> > > > >>> > > > > > >> > > > > in
> > > > >>> > > > > > >> > > > > > parallel?
> > > > >>> > > > > > >> > > > > >
> > > > >>> > > > > > >> > > > > > Thanks.
> > > > >>> > > > > > >> > > > > >
> > > > >>> > > > > > >> > > > > > --
> > > > >>> > > > > > >> > > > > > -Bhupesh
> > > > >>> > > > > > >> > > > >
> > > > >>> > > > > > >> > > >
> > > > >>> > > > > > >> > >
> > > > >>> > > > > > >> >
> > > > >>> > > > > > >>
> > > > >>> > > > > >
> > > > >>> > > > >
> > > > >>> > > >
> > > > >>> > >
> > > > >>> >
> > > > >>>
> > > > >
> > > >
> > > >
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Thomas Weise <th...@datatorrent.com>.
What you are looking for would require the guarantee for checkpointed to be
called before next beginWindow.
Why don't you record the list in endWindow? Is it a performance concern?
This would then benefit from https://malhar.atlassian.net/browse/APEX-78
On Tue, Nov 10, 2015 at 11:20 PM, Bhupesh Chawda <bh...@datatorrent.com>
wrote:
> I was trying to do some processing between two checkpoint windows.
>
> So, for example, an operator deletes some files from hdfs as part of normal
> processing within an application window. However, since hdfs deletes cannot
> be rolled back by the platform, I am trying NOT to delete the files from
> hdfs but just record them in a list. Now when a checkpoint happens, I need
> to take a snapshot of the list *as of the checkpoint* and later delete them
> asynchronously. This process of taking the snapshot is what I need to do
> between two checkpoint windows. This can be done anytime between the
> endWindow just before the checkpoint and beginWindow just after the
> checkpoint.
>
> So the question is, can this be done in the checkpointed callback?
>
> If the callback is asynchronous, i. e. It may be called after the next
> window starts, then probably checkpointed is not the right place to do it.
> On 11-Nov-2015 11:20 am, "Thomas Weise" <th...@datatorrent.com> wrote:
>
> > There does not seem to be a problem with the order of the checkpointed
> > callbacks.
> >
> > I would however question why the callback is delayed until write to
> storage
> > is complete. From operator perspective, it can be called as soon as
> > serialization is complete, which is always synchronous.
> >
> > The checkpoint cannot be reported to StrAM until the file copy is done,
> as
> > currently implemented.
> >
> > Bhupesh, what problem were you looking to solve through the checkpointed
> > callback?
> >
> >
> >
> >
> > On Tue, Nov 10, 2015 at 3:57 PM, Chetan Narsude <ch...@apache.org>
> wrote:
> >
> > > With async checkpointing the Node.reportStats is reporting it back and
> it
> > > looks like the code is checking for the top of the queue to see if it’s
> > > done (or not reporting at all). So I do not see a reason why they will
> be
> > > reported out of order.
> > >
> > > Chandni, I read your subsequent responses. Valid point in the last
> email
> > > about documentation. It amazes me how much the documentation of our API
> > > can be improved (self guilty probably the most). One comment that I
> > wanted
> > > to make even before that is that the semantics of copyToHdfs is that it
> > > either succeeds or throws exception in which case the recovery kicks in
> > > and nothing is reported as checkpointed.
> > >
> > > Tim, we need the unit test, man!
> > >
> > > —
> > > Chetan
> > >
> > >
> > > On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <cn...@cisco.com>
> > > wrote:
> > >
> > > >There are a lot of things which are different when it comes to async
> > > >checkpointing. I was evaluating it in the morning and expect that
> > either I
> > > >am able to explain or open jira issues. With my partial observation is
> > > >that with Async checkpointing, checkpointed is not issued (chandni,
> the
> > > >last statement in the if block is ³return²). I am digging into it but
> > feel
> > > >free to chime in if someone else is able to find that.
> > > >
> > > >Also I realized that my morning email applies as it is to committed
> but
> > > >checkpointed has deviated a little bit from that. Will post the
> revised
> > > >response soon.
> > > >
> > > >‹
> > > >Chetan
> > > >
> > > >
> > > >
> > > >
> > > >On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com>
> wrote:
> > > >
> > > >>Chetan,
> > > >>
> > > >>Looking at the checkpoint(windowId) in Node.java, I don't think the
> > steps
> > > >>you mentioned are followed.
> > > >>
> > > >>*if (using AsyncFSStorageAgent) {*
> > > >>* asyncFSStorageAgent.copyToHdfs(...)*
> > > >>*}*
> > > >>*operator.checkpointed(windowId);*
> > > >>
> > > >>This means even copyToHdfs fails the operator is notified that the
> > window
> > > >>is check-pointed.
> > > >>
> > > >>Are we saying that copyToHdfs will never fail with
> AsyncFSStorageAgent
> > > >>for
> > > >>a window since the operator is notified that the window is
> > checkpointed?
> > > >>
> > > >>Chandni
> > > >>
> > > >>On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <
> tim@datatorrent.com>
> > > >>wrote:
> > > >>
> > > >>> Will do
> > > >>>
> > > >>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
> > > >>><pr...@datatorrent.com>
> > > >>> wrote:
> > > >>>
> > > >>> > Is there a unit test covering it? Otherwise can you write one to
> > test
> > > >>>the
> > > >>> > hypothesis.
> > > >>> >
> > > >>> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
> > > >>><ti...@datatorrent.com>
> > > >>> > wrote:
> > > >>> >
> > > >>> > > That is what it is looking like to me. The task is submitted
> > > >>> > > GenericNode#checkpoint line 504, then at the end of the
> > > >>> > > GenericNode#checkpoint line 531 checkpointed is called. I am
> > likely
> > > >>> > missing
> > > >>> > > something, just would like to know what :)
> > > >>> > >
> > > >>> > > Tim
> > > >>> > >
> > > >>> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> > > >>> > pramod@datatorrent.com>
> > > >>> > > wrote:
> > > >>> > >
> > > >>> > > > Tim,
> > > >>> > > >
> > > >>> > > > Are you suggesting that checkpointed is called before the
> > > >>>checkpoint
> > > >>> is
> > > >>> > > > completely persisted in the storage.
> > > >>> > > >
> > > >>> > > > Thanks
> > > >>> > > >
> > > >>> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> > > >>> tim@datatorrent.com>
> > > >>> > > > wrote:
> > > >>> > > >
> > > >>> > > > > Chetan,
> > > >>> > > > >
> > > >>> > > > > I do not see the process of reporting the checkpoint to
> > stram,
> > > >>> > > receiving
> > > >>> > > > > the ack, and then calling checkpointed. The logic I'm
> seeing
> > in
> > > >>> > > > GenericNode
> > > >>> > > > > line 484 is that the checkpoint method is called, it spawns
> > > >>>another
> > > >>> > > > thread
> > > >>> > > > > that writes to hdfs, and then checkpointed is immediately
> > > >>>called
> > > >>> > > > > afterwards. I am missing something, can you give me some
> > > >>>pointers
> > > >>> so
> > > >>> > > > that I
> > > >>> > > > > can better understand the flow?
> > > >>> > > > >
> > > >>> > > > > Tim
> > > >>> > > > >
> > > >>> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> > > >>> > > ram@datatorrent.com
> > > >>> > > > >
> > > >>> > > > > wrote:
> > > >>> > > > >
> > > >>> > > > > > Chetan's answer provides a good explanation as well as
> > > >>>clarifying
> > > >>> > > that
> > > >>> > > > > > the difference can be more than 1.
> > > >>> > > > > >
> > > >>> > > > > > Since checkpointing (i.e. "commit notification" as Thomas
> > > >>>refers
> > > >>> to
> > > >>> > > > > > it) is asynchronous, I'm curious
> > > >>> > > > > > about whether the window ids in the checkpointed call are
> > > >>> > guaranteed
> > > >>> > > > > > to be sequential or if they could
> > > >>> > > > > > be out of order, i.e. can the checkpointed call see
> window
> > id
> > > >>>101
> > > >>> > > > > > before it sees 100 ?
> > > >>> > > > > >
> > > >>> > > > > > Ram
> > > >>> > > > > >
> > > >>> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > > >>> > > > > > <bh...@datatorrent.com> wrote:
> > > >>> > > > > > > Hi Tim,
> > > >>> > > > > > > Thanks for the detailed explanation.
> > > >>> > > > > > > I understand that the sequence would be
> > > >>> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x)
> ->
> > > >>> > > beginWindow
> > > >>> > > > > > > (x+1)
> > > >>> > > > > > >
> > > >>> > > > > > > However when I try to print out the window ids in
> > > >>>beginWindow,
> > > >>> > > > > endWindow
> > > >>> > > > > > > and checkpointed calls, I see x and x-1 respectively.
> > > >>> > > > > > > I.e. If the window just before checkpoint is 100, I see
> > > >>>that
> > > >>> the
> > > >>> > > > > > > checkpointed call had window id 99.
> > > >>> > > > > > >
> > > >>> > > > > > > Note: This is observed in the local mode of Apex.
> > > >>> > > > > > >
> > > >>> > > > > > > Thanks
> > > >>> > > > > > > -Bhupesh
> > > >>> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
> > > >>><tim@datatorrent.com
> > > >>> >
> > > >>> > > > wrote:
> > > >>> > > > > > >
> > > >>> > > > > > >> Hi Bhupesh,
> > > >>> > > > > > >>
> > > >>> > > > > > >> The sequencing of checkpoint called in relation to
> > > >>>beginWindow
> > > >>> > and
> > > >>> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT
> > and
> > > >>> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
> > > >>> WINDOW_COUNTs
> > > >>> > > are
> > > >>> > > > > not
> > > >>> > > > > > >> configured to be the same then it is possible that
> > > >>> checkpointed
> > > >>> > is
> > > >>> > > > > > called
> > > >>> > > > > > >> within an application window. So the sequence of
> events
> > > >>>would
> > > >>> be
> > > >>> > > > this:
> > > >>> > > > > > >>
> > > >>> > > > > > >> beginWindow -> checkpointed -> endWindow
> > > >>> > > > > > >>
> > > >>> > > > > > >> If the APPLICATION_WINDOW_COUNT and
> > > >>>CHECKPOINT_WINDOW_COUNT
> > > >>> are
> > > >>> > > the
> > > >>> > > > > same
> > > >>> > > > > > >> (default). Then the sequence of calls would be this:
> > > >>> > > > > > >>
> > > >>> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed
> > > >>>(100)
> > > >>> ->
> > > >>> > > > > > beginWindow
> > > >>> > > > > > >> (101)
> > > >>> > > > > > >>
> > > >>> > > > > > >> The numbers in the sequence represent possible
> streaming
> > > >>> window
> > > >>> > > Ids
> > > >>> > > > > that
> > > >>> > > > > > >> each call would be associated with.
> > > >>> > > > > > >>
> > > >>> > > > > > >> The StateMachine which calls these callbacks for
> > non-input
> > > >>> > > operators
> > > >>> > > > > is
> > > >>> > > > > > in
> > > >>> > > > > > >> GenericNode.java.
> > > >>> > > > > > >>
> > > >>> > > > > > >> Tim
> > > >>> > > > > > >>
> > > >>> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > > >>> > > > > > bhupesh@datatorrent.com>
> > > >>> > > > > > >> wrote:
> > > >>> > > > > > >>
> > > >>> > > > > > >> > Hi Chetan / Community,
> > > >>> > > > > > >> >
> > > >>> > > > > > >> > Can someone please elaborate on why the window id
> > > >>>supplied
> > > >>> to
> > > >>> > > > > > >> > CheckpointListener and the Operator would differ.
> > > >>> > > > > > >> > I tried looking at the window ids of checkpointed()
> > and
> > > >>>the
> > > >>> > > > > > beginWindow()
> > > >>> > > > > > >> > calls and they differ by 1. Don't know why this
> should
> > > >>>be
> > > >>> the
> > > >>> > > > case.
> > > >>> > > > > > >> >
> > > >>> > > > > > >> > Thanks.
> > > >>> > > > > > >> > -Bhupesh
> > > >>> > > > > > >> >
> > > >>> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > > >>> > > > > > chetan@datatorrent.com>
> > > >>> > > > > > >> > wrote:
> > > >>> > > > > > >> >
> > > >>> > > > > > >> > > Short answer is yes.
> > > >>> > > > > > >> > >
> > > >>> > > > > > >> > > All the control tuples are scheduled to be
> delivered
> > > >>> outside
> > > >>> > > of
> > > >>> > > > > the
> > > >>> > > > > > >> > window.
> > > >>> > > > > > >> > > As checkpointed callback is triggered because of
> > > >>> CHECKPOINT
> > > >>> > > > > control
> > > >>> > > > > > >> > tuple,
> > > >>> > > > > > >> > > it will happen after endWindow and before the next
> > > >>> > > beginWindow.
> > > >>> > > > > > >> > >
> > > >>> > > > > > >> > > The windowId supplied to CheckpointListener and
> the
> > > >>>one
> > > >>> > > provided
> > > >>> > > > > to
> > > >>> > > > > > >> > > Operator need not match even though the sequence
> is
> > > >>> defined.
> > > >>> > > So
> > > >>> > > > I
> > > >>> > > > > am
> > > >>> > > > > > >> > > curious how you intend to use this knowledge.
> > > >>> > > > > > >> > >
> > > >>> > > > > > >> > > --
> > > >>> > > > > > >> > > Chetan
> > > >>> > > > > > >> > >
> > > >>> > > > > > >> > >
> > > >>> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > > >>> > > > > > thomas@datatorrent.com>
> > > >>> > > > > > >> > > wrote:
> > > >>> > > > > > >> > >
> > > >>> > > > > > >> > > > It has not changed the operator execution model.
> > > >>>State
> > > >>> > > > > > serialization
> > > >>> > > > > > >> is
> > > >>> > > > > > >> > > > still synchronous, write to HDFS is taken out of
> > the
> > > >>> > > operator
> > > >>> > > > > > thread.
> > > >>> > > > > > >> > > >
> > > >>> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > > >>> > > > > amol@datatorrent.com
> > > >>> > > > > > >
> > > >>> > > > > > >> > > wrote:
> > > >>> > > > > > >> > > >
> > > >>> > > > > > >> > > > >
> > > >>> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
> > > >>>changed
> > > >>> > > this?
> > > >>> > > > > > >> > > > >
> > > >>> > > > > > >> > > > > Amol
> > > >>> > > > > > >> > > > >
> > > >>> > > > > > >> > > > > Sent from my iPhone
> > > >>> > > > > > >> > > > >
> > > >>> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh
> Chawda <
> > > >>> > > > > > >> > > bhupesh@datatorrent.com>
> > > >>> > > > > > >> > > > > wrote:
> > > >>> > > > > > >> > > > > >
> > > >>> > > > > > >> > > > > > Hi All,
> > > >>> > > > > > >> > > > > >
> > > >>> > > > > > >> > > > > > Is it safe to assume that the checkpointed()
> > and
> > > >>>the
> > > >>> > > > > > >> beginWindow()
> > > >>> > > > > > >> > > > calls
> > > >>> > > > > > >> > > > > > are sequenced?
> > > >>> > > > > > >> > > > > > In other words, are these calls part of the
> > same
> > > >>> > thread
> > > >>> > > > and
> > > >>> > > > > > may
> > > >>> > > > > > >> not
> > > >>> > > > > > >> > > run
> > > >>> > > > > > >> > > > > in
> > > >>> > > > > > >> > > > > > parallel?
> > > >>> > > > > > >> > > > > >
> > > >>> > > > > > >> > > > > > Thanks.
> > > >>> > > > > > >> > > > > >
> > > >>> > > > > > >> > > > > > --
> > > >>> > > > > > >> > > > > > -Bhupesh
> > > >>> > > > > > >> > > > >
> > > >>> > > > > > >> > > >
> > > >>> > > > > > >> > >
> > > >>> > > > > > >> >
> > > >>> > > > > > >>
> > > >>> > > > > >
> > > >>> > > > >
> > > >>> > > >
> > > >>> > >
> > > >>> >
> > > >>>
> > > >
> > >
> > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Bhupesh Chawda <bh...@datatorrent.com>.
I was trying to do some processing between two checkpoint windows.
So, for example, an operator deletes some files from hdfs as part of normal
processing within an application window. However, since hdfs deletes cannot
be rolled back by the platform, I am trying NOT to delete the files from
hdfs but just record them in a list. Now when a checkpoint happens, I need
to take a snapshot of the list *as of the checkpoint* and later delete them
asynchronously. This process of taking the snapshot is what I need to do
between two checkpoint windows. This can be done anytime between the
endWindow just before the checkpoint and beginWindow just after the
checkpoint.
So the question is, can this be done in the checkpointed callback?
If the callback is asynchronous, i. e. It may be called after the next
window starts, then probably checkpointed is not the right place to do it.
On 11-Nov-2015 11:20 am, "Thomas Weise" <th...@datatorrent.com> wrote:
> There does not seem to be a problem with the order of the checkpointed
> callbacks.
>
> I would however question why the callback is delayed until write to storage
> is complete. From operator perspective, it can be called as soon as
> serialization is complete, which is always synchronous.
>
> The checkpoint cannot be reported to StrAM until the file copy is done, as
> currently implemented.
>
> Bhupesh, what problem were you looking to solve through the checkpointed
> callback?
>
>
>
>
> On Tue, Nov 10, 2015 at 3:57 PM, Chetan Narsude <ch...@apache.org> wrote:
>
> > With async checkpointing the Node.reportStats is reporting it back and it
> > looks like the code is checking for the top of the queue to see if it’s
> > done (or not reporting at all). So I do not see a reason why they will be
> > reported out of order.
> >
> > Chandni, I read your subsequent responses. Valid point in the last email
> > about documentation. It amazes me how much the documentation of our API
> > can be improved (self guilty probably the most). One comment that I
> wanted
> > to make even before that is that the semantics of copyToHdfs is that it
> > either succeeds or throws exception in which case the recovery kicks in
> > and nothing is reported as checkpointed.
> >
> > Tim, we need the unit test, man!
> >
> > —
> > Chetan
> >
> >
> > On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <cn...@cisco.com>
> > wrote:
> >
> > >There are a lot of things which are different when it comes to async
> > >checkpointing. I was evaluating it in the morning and expect that
> either I
> > >am able to explain or open jira issues. With my partial observation is
> > >that with Async checkpointing, checkpointed is not issued (chandni, the
> > >last statement in the if block is ³return²). I am digging into it but
> feel
> > >free to chime in if someone else is able to find that.
> > >
> > >Also I realized that my morning email applies as it is to committed but
> > >checkpointed has deviated a little bit from that. Will post the revised
> > >response soon.
> > >
> > >‹
> > >Chetan
> > >
> > >
> > >
> > >
> > >On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
> > >
> > >>Chetan,
> > >>
> > >>Looking at the checkpoint(windowId) in Node.java, I don't think the
> steps
> > >>you mentioned are followed.
> > >>
> > >>*if (using AsyncFSStorageAgent) {*
> > >>* asyncFSStorageAgent.copyToHdfs(...)*
> > >>*}*
> > >>*operator.checkpointed(windowId);*
> > >>
> > >>This means even copyToHdfs fails the operator is notified that the
> window
> > >>is check-pointed.
> > >>
> > >>Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent
> > >>for
> > >>a window since the operator is notified that the window is
> checkpointed?
> > >>
> > >>Chandni
> > >>
> > >>On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
> > >>wrote:
> > >>
> > >>> Will do
> > >>>
> > >>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
> > >>><pr...@datatorrent.com>
> > >>> wrote:
> > >>>
> > >>> > Is there a unit test covering it? Otherwise can you write one to
> test
> > >>>the
> > >>> > hypothesis.
> > >>> >
> > >>> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
> > >>><ti...@datatorrent.com>
> > >>> > wrote:
> > >>> >
> > >>> > > That is what it is looking like to me. The task is submitted
> > >>> > > GenericNode#checkpoint line 504, then at the end of the
> > >>> > > GenericNode#checkpoint line 531 checkpointed is called. I am
> likely
> > >>> > missing
> > >>> > > something, just would like to know what :)
> > >>> > >
> > >>> > > Tim
> > >>> > >
> > >>> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> > >>> > pramod@datatorrent.com>
> > >>> > > wrote:
> > >>> > >
> > >>> > > > Tim,
> > >>> > > >
> > >>> > > > Are you suggesting that checkpointed is called before the
> > >>>checkpoint
> > >>> is
> > >>> > > > completely persisted in the storage.
> > >>> > > >
> > >>> > > > Thanks
> > >>> > > >
> > >>> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> > >>> tim@datatorrent.com>
> > >>> > > > wrote:
> > >>> > > >
> > >>> > > > > Chetan,
> > >>> > > > >
> > >>> > > > > I do not see the process of reporting the checkpoint to
> stram,
> > >>> > > receiving
> > >>> > > > > the ack, and then calling checkpointed. The logic I'm seeing
> in
> > >>> > > > GenericNode
> > >>> > > > > line 484 is that the checkpoint method is called, it spawns
> > >>>another
> > >>> > > > thread
> > >>> > > > > that writes to hdfs, and then checkpointed is immediately
> > >>>called
> > >>> > > > > afterwards. I am missing something, can you give me some
> > >>>pointers
> > >>> so
> > >>> > > > that I
> > >>> > > > > can better understand the flow?
> > >>> > > > >
> > >>> > > > > Tim
> > >>> > > > >
> > >>> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> > >>> > > ram@datatorrent.com
> > >>> > > > >
> > >>> > > > > wrote:
> > >>> > > > >
> > >>> > > > > > Chetan's answer provides a good explanation as well as
> > >>>clarifying
> > >>> > > that
> > >>> > > > > > the difference can be more than 1.
> > >>> > > > > >
> > >>> > > > > > Since checkpointing (i.e. "commit notification" as Thomas
> > >>>refers
> > >>> to
> > >>> > > > > > it) is asynchronous, I'm curious
> > >>> > > > > > about whether the window ids in the checkpointed call are
> > >>> > guaranteed
> > >>> > > > > > to be sequential or if they could
> > >>> > > > > > be out of order, i.e. can the checkpointed call see window
> id
> > >>>101
> > >>> > > > > > before it sees 100 ?
> > >>> > > > > >
> > >>> > > > > > Ram
> > >>> > > > > >
> > >>> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > >>> > > > > > <bh...@datatorrent.com> wrote:
> > >>> > > > > > > Hi Tim,
> > >>> > > > > > > Thanks for the detailed explanation.
> > >>> > > > > > > I understand that the sequence would be
> > >>> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> > >>> > > beginWindow
> > >>> > > > > > > (x+1)
> > >>> > > > > > >
> > >>> > > > > > > However when I try to print out the window ids in
> > >>>beginWindow,
> > >>> > > > > endWindow
> > >>> > > > > > > and checkpointed calls, I see x and x-1 respectively.
> > >>> > > > > > > I.e. If the window just before checkpoint is 100, I see
> > >>>that
> > >>> the
> > >>> > > > > > > checkpointed call had window id 99.
> > >>> > > > > > >
> > >>> > > > > > > Note: This is observed in the local mode of Apex.
> > >>> > > > > > >
> > >>> > > > > > > Thanks
> > >>> > > > > > > -Bhupesh
> > >>> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
> > >>><tim@datatorrent.com
> > >>> >
> > >>> > > > wrote:
> > >>> > > > > > >
> > >>> > > > > > >> Hi Bhupesh,
> > >>> > > > > > >>
> > >>> > > > > > >> The sequencing of checkpoint called in relation to
> > >>>beginWindow
> > >>> > and
> > >>> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT
> and
> > >>> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
> > >>> WINDOW_COUNTs
> > >>> > > are
> > >>> > > > > not
> > >>> > > > > > >> configured to be the same then it is possible that
> > >>> checkpointed
> > >>> > is
> > >>> > > > > > called
> > >>> > > > > > >> within an application window. So the sequence of events
> > >>>would
> > >>> be
> > >>> > > > this:
> > >>> > > > > > >>
> > >>> > > > > > >> beginWindow -> checkpointed -> endWindow
> > >>> > > > > > >>
> > >>> > > > > > >> If the APPLICATION_WINDOW_COUNT and
> > >>>CHECKPOINT_WINDOW_COUNT
> > >>> are
> > >>> > > the
> > >>> > > > > same
> > >>> > > > > > >> (default). Then the sequence of calls would be this:
> > >>> > > > > > >>
> > >>> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed
> > >>>(100)
> > >>> ->
> > >>> > > > > > beginWindow
> > >>> > > > > > >> (101)
> > >>> > > > > > >>
> > >>> > > > > > >> The numbers in the sequence represent possible streaming
> > >>> window
> > >>> > > Ids
> > >>> > > > > that
> > >>> > > > > > >> each call would be associated with.
> > >>> > > > > > >>
> > >>> > > > > > >> The StateMachine which calls these callbacks for
> non-input
> > >>> > > operators
> > >>> > > > > is
> > >>> > > > > > in
> > >>> > > > > > >> GenericNode.java.
> > >>> > > > > > >>
> > >>> > > > > > >> Tim
> > >>> > > > > > >>
> > >>> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > >>> > > > > > bhupesh@datatorrent.com>
> > >>> > > > > > >> wrote:
> > >>> > > > > > >>
> > >>> > > > > > >> > Hi Chetan / Community,
> > >>> > > > > > >> >
> > >>> > > > > > >> > Can someone please elaborate on why the window id
> > >>>supplied
> > >>> to
> > >>> > > > > > >> > CheckpointListener and the Operator would differ.
> > >>> > > > > > >> > I tried looking at the window ids of checkpointed()
> and
> > >>>the
> > >>> > > > > > beginWindow()
> > >>> > > > > > >> > calls and they differ by 1. Don't know why this should
> > >>>be
> > >>> the
> > >>> > > > case.
> > >>> > > > > > >> >
> > >>> > > > > > >> > Thanks.
> > >>> > > > > > >> > -Bhupesh
> > >>> > > > > > >> >
> > >>> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > >>> > > > > > chetan@datatorrent.com>
> > >>> > > > > > >> > wrote:
> > >>> > > > > > >> >
> > >>> > > > > > >> > > Short answer is yes.
> > >>> > > > > > >> > >
> > >>> > > > > > >> > > All the control tuples are scheduled to be delivered
> > >>> outside
> > >>> > > of
> > >>> > > > > the
> > >>> > > > > > >> > window.
> > >>> > > > > > >> > > As checkpointed callback is triggered because of
> > >>> CHECKPOINT
> > >>> > > > > control
> > >>> > > > > > >> > tuple,
> > >>> > > > > > >> > > it will happen after endWindow and before the next
> > >>> > > beginWindow.
> > >>> > > > > > >> > >
> > >>> > > > > > >> > > The windowId supplied to CheckpointListener and the
> > >>>one
> > >>> > > provided
> > >>> > > > > to
> > >>> > > > > > >> > > Operator need not match even though the sequence is
> > >>> defined.
> > >>> > > So
> > >>> > > > I
> > >>> > > > > am
> > >>> > > > > > >> > > curious how you intend to use this knowledge.
> > >>> > > > > > >> > >
> > >>> > > > > > >> > > --
> > >>> > > > > > >> > > Chetan
> > >>> > > > > > >> > >
> > >>> > > > > > >> > >
> > >>> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > >>> > > > > > thomas@datatorrent.com>
> > >>> > > > > > >> > > wrote:
> > >>> > > > > > >> > >
> > >>> > > > > > >> > > > It has not changed the operator execution model.
> > >>>State
> > >>> > > > > > serialization
> > >>> > > > > > >> is
> > >>> > > > > > >> > > > still synchronous, write to HDFS is taken out of
> the
> > >>> > > operator
> > >>> > > > > > thread.
> > >>> > > > > > >> > > >
> > >>> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > >>> > > > > amol@datatorrent.com
> > >>> > > > > > >
> > >>> > > > > > >> > > wrote:
> > >>> > > > > > >> > > >
> > >>> > > > > > >> > > > >
> > >>> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
> > >>>changed
> > >>> > > this?
> > >>> > > > > > >> > > > >
> > >>> > > > > > >> > > > > Amol
> > >>> > > > > > >> > > > >
> > >>> > > > > > >> > > > > Sent from my iPhone
> > >>> > > > > > >> > > > >
> > >>> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > >>> > > > > > >> > > bhupesh@datatorrent.com>
> > >>> > > > > > >> > > > > wrote:
> > >>> > > > > > >> > > > > >
> > >>> > > > > > >> > > > > > Hi All,
> > >>> > > > > > >> > > > > >
> > >>> > > > > > >> > > > > > Is it safe to assume that the checkpointed()
> and
> > >>>the
> > >>> > > > > > >> beginWindow()
> > >>> > > > > > >> > > > calls
> > >>> > > > > > >> > > > > > are sequenced?
> > >>> > > > > > >> > > > > > In other words, are these calls part of the
> same
> > >>> > thread
> > >>> > > > and
> > >>> > > > > > may
> > >>> > > > > > >> not
> > >>> > > > > > >> > > run
> > >>> > > > > > >> > > > > in
> > >>> > > > > > >> > > > > > parallel?
> > >>> > > > > > >> > > > > >
> > >>> > > > > > >> > > > > > Thanks.
> > >>> > > > > > >> > > > > >
> > >>> > > > > > >> > > > > > --
> > >>> > > > > > >> > > > > > -Bhupesh
> > >>> > > > > > >> > > > >
> > >>> > > > > > >> > > >
> > >>> > > > > > >> > >
> > >>> > > > > > >> >
> > >>> > > > > > >>
> > >>> > > > > >
> > >>> > > > >
> > >>> > > >
> > >>> > >
> > >>> >
> > >>>
> > >
> >
> >
> >
>
Re: Sequencing of operator calls
Posted by Thomas Weise <th...@datatorrent.com>.
There does not seem to be a problem with the order of the checkpointed
callbacks.
I would however question why the callback is delayed until write to storage
is complete. From operator perspective, it can be called as soon as
serialization is complete, which is always synchronous.
The checkpoint cannot be reported to StrAM until the file copy is done, as
currently implemented.
Bhupesh, what problem were you looking to solve through the checkpointed
callback?
On Tue, Nov 10, 2015 at 3:57 PM, Chetan Narsude <ch...@apache.org> wrote:
> With async checkpointing the Node.reportStats is reporting it back and it
> looks like the code is checking for the top of the queue to see if it’s
> done (or not reporting at all). So I do not see a reason why they will be
> reported out of order.
>
> Chandni, I read your subsequent responses. Valid point in the last email
> about documentation. It amazes me how much the documentation of our API
> can be improved (self guilty probably the most). One comment that I wanted
> to make even before that is that the semantics of copyToHdfs is that it
> either succeeds or throws exception in which case the recovery kicks in
> and nothing is reported as checkpointed.
>
> Tim, we need the unit test, man!
>
> —
> Chetan
>
>
> On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <cn...@cisco.com>
> wrote:
>
> >There are a lot of things which are different when it comes to async
> >checkpointing. I was evaluating it in the morning and expect that either I
> >am able to explain or open jira issues. With my partial observation is
> >that with Async checkpointing, checkpointed is not issued (chandni, the
> >last statement in the if block is ³return²). I am digging into it but feel
> >free to chime in if someone else is able to find that.
> >
> >Also I realized that my morning email applies as it is to committed but
> >checkpointed has deviated a little bit from that. Will post the revised
> >response soon.
> >
> >‹
> >Chetan
> >
> >
> >
> >
> >On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
> >
> >>Chetan,
> >>
> >>Looking at the checkpoint(windowId) in Node.java, I don't think the steps
> >>you mentioned are followed.
> >>
> >>*if (using AsyncFSStorageAgent) {*
> >>* asyncFSStorageAgent.copyToHdfs(...)*
> >>*}*
> >>*operator.checkpointed(windowId);*
> >>
> >>This means even copyToHdfs fails the operator is notified that the window
> >>is check-pointed.
> >>
> >>Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent
> >>for
> >>a window since the operator is notified that the window is checkpointed?
> >>
> >>Chandni
> >>
> >>On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
> >>wrote:
> >>
> >>> Will do
> >>>
> >>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
> >>><pr...@datatorrent.com>
> >>> wrote:
> >>>
> >>> > Is there a unit test covering it? Otherwise can you write one to test
> >>>the
> >>> > hypothesis.
> >>> >
> >>> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
> >>><ti...@datatorrent.com>
> >>> > wrote:
> >>> >
> >>> > > That is what it is looking like to me. The task is submitted
> >>> > > GenericNode#checkpoint line 504, then at the end of the
> >>> > > GenericNode#checkpoint line 531 checkpointed is called. I am likely
> >>> > missing
> >>> > > something, just would like to know what :)
> >>> > >
> >>> > > Tim
> >>> > >
> >>> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> >>> > pramod@datatorrent.com>
> >>> > > wrote:
> >>> > >
> >>> > > > Tim,
> >>> > > >
> >>> > > > Are you suggesting that checkpointed is called before the
> >>>checkpoint
> >>> is
> >>> > > > completely persisted in the storage.
> >>> > > >
> >>> > > > Thanks
> >>> > > >
> >>> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> >>> tim@datatorrent.com>
> >>> > > > wrote:
> >>> > > >
> >>> > > > > Chetan,
> >>> > > > >
> >>> > > > > I do not see the process of reporting the checkpoint to stram,
> >>> > > receiving
> >>> > > > > the ack, and then calling checkpointed. The logic I'm seeing in
> >>> > > > GenericNode
> >>> > > > > line 484 is that the checkpoint method is called, it spawns
> >>>another
> >>> > > > thread
> >>> > > > > that writes to hdfs, and then checkpointed is immediately
> >>>called
> >>> > > > > afterwards. I am missing something, can you give me some
> >>>pointers
> >>> so
> >>> > > > that I
> >>> > > > > can better understand the flow?
> >>> > > > >
> >>> > > > > Tim
> >>> > > > >
> >>> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> >>> > > ram@datatorrent.com
> >>> > > > >
> >>> > > > > wrote:
> >>> > > > >
> >>> > > > > > Chetan's answer provides a good explanation as well as
> >>>clarifying
> >>> > > that
> >>> > > > > > the difference can be more than 1.
> >>> > > > > >
> >>> > > > > > Since checkpointing (i.e. "commit notification" as Thomas
> >>>refers
> >>> to
> >>> > > > > > it) is asynchronous, I'm curious
> >>> > > > > > about whether the window ids in the checkpointed call are
> >>> > guaranteed
> >>> > > > > > to be sequential or if they could
> >>> > > > > > be out of order, i.e. can the checkpointed call see window id
> >>>101
> >>> > > > > > before it sees 100 ?
> >>> > > > > >
> >>> > > > > > Ram
> >>> > > > > >
> >>> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> >>> > > > > > <bh...@datatorrent.com> wrote:
> >>> > > > > > > Hi Tim,
> >>> > > > > > > Thanks for the detailed explanation.
> >>> > > > > > > I understand that the sequence would be
> >>> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> >>> > > beginWindow
> >>> > > > > > > (x+1)
> >>> > > > > > >
> >>> > > > > > > However when I try to print out the window ids in
> >>>beginWindow,
> >>> > > > > endWindow
> >>> > > > > > > and checkpointed calls, I see x and x-1 respectively.
> >>> > > > > > > I.e. If the window just before checkpoint is 100, I see
> >>>that
> >>> the
> >>> > > > > > > checkpointed call had window id 99.
> >>> > > > > > >
> >>> > > > > > > Note: This is observed in the local mode of Apex.
> >>> > > > > > >
> >>> > > > > > > Thanks
> >>> > > > > > > -Bhupesh
> >>> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
> >>><tim@datatorrent.com
> >>> >
> >>> > > > wrote:
> >>> > > > > > >
> >>> > > > > > >> Hi Bhupesh,
> >>> > > > > > >>
> >>> > > > > > >> The sequencing of checkpoint called in relation to
> >>>beginWindow
> >>> > and
> >>> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> >>> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
> >>> WINDOW_COUNTs
> >>> > > are
> >>> > > > > not
> >>> > > > > > >> configured to be the same then it is possible that
> >>> checkpointed
> >>> > is
> >>> > > > > > called
> >>> > > > > > >> within an application window. So the sequence of events
> >>>would
> >>> be
> >>> > > > this:
> >>> > > > > > >>
> >>> > > > > > >> beginWindow -> checkpointed -> endWindow
> >>> > > > > > >>
> >>> > > > > > >> If the APPLICATION_WINDOW_COUNT and
> >>>CHECKPOINT_WINDOW_COUNT
> >>> are
> >>> > > the
> >>> > > > > same
> >>> > > > > > >> (default). Then the sequence of calls would be this:
> >>> > > > > > >>
> >>> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed
> >>>(100)
> >>> ->
> >>> > > > > > beginWindow
> >>> > > > > > >> (101)
> >>> > > > > > >>
> >>> > > > > > >> The numbers in the sequence represent possible streaming
> >>> window
> >>> > > Ids
> >>> > > > > that
> >>> > > > > > >> each call would be associated with.
> >>> > > > > > >>
> >>> > > > > > >> The StateMachine which calls these callbacks for non-input
> >>> > > operators
> >>> > > > > is
> >>> > > > > > in
> >>> > > > > > >> GenericNode.java.
> >>> > > > > > >>
> >>> > > > > > >> Tim
> >>> > > > > > >>
> >>> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> >>> > > > > > bhupesh@datatorrent.com>
> >>> > > > > > >> wrote:
> >>> > > > > > >>
> >>> > > > > > >> > Hi Chetan / Community,
> >>> > > > > > >> >
> >>> > > > > > >> > Can someone please elaborate on why the window id
> >>>supplied
> >>> to
> >>> > > > > > >> > CheckpointListener and the Operator would differ.
> >>> > > > > > >> > I tried looking at the window ids of checkpointed() and
> >>>the
> >>> > > > > > beginWindow()
> >>> > > > > > >> > calls and they differ by 1. Don't know why this should
> >>>be
> >>> the
> >>> > > > case.
> >>> > > > > > >> >
> >>> > > > > > >> > Thanks.
> >>> > > > > > >> > -Bhupesh
> >>> > > > > > >> >
> >>> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> >>> > > > > > chetan@datatorrent.com>
> >>> > > > > > >> > wrote:
> >>> > > > > > >> >
> >>> > > > > > >> > > Short answer is yes.
> >>> > > > > > >> > >
> >>> > > > > > >> > > All the control tuples are scheduled to be delivered
> >>> outside
> >>> > > of
> >>> > > > > the
> >>> > > > > > >> > window.
> >>> > > > > > >> > > As checkpointed callback is triggered because of
> >>> CHECKPOINT
> >>> > > > > control
> >>> > > > > > >> > tuple,
> >>> > > > > > >> > > it will happen after endWindow and before the next
> >>> > > beginWindow.
> >>> > > > > > >> > >
> >>> > > > > > >> > > The windowId supplied to CheckpointListener and the
> >>>one
> >>> > > provided
> >>> > > > > to
> >>> > > > > > >> > > Operator need not match even though the sequence is
> >>> defined.
> >>> > > So
> >>> > > > I
> >>> > > > > am
> >>> > > > > > >> > > curious how you intend to use this knowledge.
> >>> > > > > > >> > >
> >>> > > > > > >> > > --
> >>> > > > > > >> > > Chetan
> >>> > > > > > >> > >
> >>> > > > > > >> > >
> >>> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> >>> > > > > > thomas@datatorrent.com>
> >>> > > > > > >> > > wrote:
> >>> > > > > > >> > >
> >>> > > > > > >> > > > It has not changed the operator execution model.
> >>>State
> >>> > > > > > serialization
> >>> > > > > > >> is
> >>> > > > > > >> > > > still synchronous, write to HDFS is taken out of the
> >>> > > operator
> >>> > > > > > thread.
> >>> > > > > > >> > > >
> >>> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> >>> > > > > amol@datatorrent.com
> >>> > > > > > >
> >>> > > > > > >> > > wrote:
> >>> > > > > > >> > > >
> >>> > > > > > >> > > > >
> >>> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
> >>>changed
> >>> > > this?
> >>> > > > > > >> > > > >
> >>> > > > > > >> > > > > Amol
> >>> > > > > > >> > > > >
> >>> > > > > > >> > > > > Sent from my iPhone
> >>> > > > > > >> > > > >
> >>> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> >>> > > > > > >> > > bhupesh@datatorrent.com>
> >>> > > > > > >> > > > > wrote:
> >>> > > > > > >> > > > > >
> >>> > > > > > >> > > > > > Hi All,
> >>> > > > > > >> > > > > >
> >>> > > > > > >> > > > > > Is it safe to assume that the checkpointed() and
> >>>the
> >>> > > > > > >> beginWindow()
> >>> > > > > > >> > > > calls
> >>> > > > > > >> > > > > > are sequenced?
> >>> > > > > > >> > > > > > In other words, are these calls part of the same
> >>> > thread
> >>> > > > and
> >>> > > > > > may
> >>> > > > > > >> not
> >>> > > > > > >> > > run
> >>> > > > > > >> > > > > in
> >>> > > > > > >> > > > > > parallel?
> >>> > > > > > >> > > > > >
> >>> > > > > > >> > > > > > Thanks.
> >>> > > > > > >> > > > > >
> >>> > > > > > >> > > > > > --
> >>> > > > > > >> > > > > > -Bhupesh
> >>> > > > > > >> > > > >
> >>> > > > > > >> > > >
> >>> > > > > > >> > >
> >>> > > > > > >> >
> >>> > > > > > >>
> >>> > > > > >
> >>> > > > >
> >>> > > >
> >>> > >
> >>> >
> >>>
> >
>
>
>
Re: Sequencing of operator calls
Posted by Thomas Weise <th...@datatorrent.com>.
On Wed, Nov 11, 2015 at 2:42 PM, Chetan Narsude (cnarsude) <cnarsude@cisco
.com> wrote:
> Just to give some more light for everyone’s benefit:
>
> Checkpointed semantics is to tell the operator that its state is
> preserved. The operator does not need to be told that it’s state is
> collected (serialized) as it does know when that operation starts, when
> it’s in progress and when it ends. There are good constructs to intercept
> serialization which are not apex specific.
That's a workaround, not a solution. How an operator is checkpointed is
determined by the implementation of storage agent. Therefore the operator
writer should not assume a particular serialization protocol.
Re: Sequencing of operator calls
Posted by "Chetan Narsude (cnarsude)" <cn...@cisco.com>.
Just to give some more light for everyone’s benefit:
Checkpointed semantics is to tell the operator that its state is
preserved. The operator does not need to be told that it’s state is
collected (serialized) as it does know when that operation starts, when
it’s in progress and when it ends. There are good constructs to intercept
serialization which are not apex specific.
Bhuopesh, for your use case you would need to keep mapping between the
window id and the files that need to be processed during the window id.
Also you probably need to do this operation in committed callback and not
checkpointed (of course I am guessing due to incomplete knowledge of your
use case).
dtIngest has a similar use case that we worked on more than a year ago and
probably someone from that team can give you more insight into the problem
we are trying to avoid.
—
Chetan
On 11/11/15, 11:38 AM, "Timothy Farkas" <ti...@datatorrent.com> wrote:
>Hi Gaurav,
>
>I've added some tests to check some of the conditions when checkpointed is
>called https://github.com/apache/incubator-apex-core/pull/163
>
>If there are more ideas of things to test I can add more unit tests as
>well.
>
>Also I was wondering how a failure to copy data from local disk to hdfs is
>handled? If an exception is thrown the thread doing the task will die and
>then what? Is the task retried?
>
>Thanks,
>Tim
>
>On Wed, Nov 11, 2015 at 1:29 AM, Gaurav Gupta <ga...@datatorrent.com>
>wrote:
>
>> Chetan,
>>
>> Your analysis is correct.
>>
>> I will again summarize how AsyncFSStorageAgent works. It involves two
>>steps
>>
>> 1. Serializing the Operator state to local disk.
>> 2. Moving this serialized state to HDFS.
>>
>> These are how the above two steps work in different scenario
>>
>> 1. If PROCESSING is EXACTLY ONCE for an operator, then both the above
>> steps happen in the Operator thread. Once the above two steps are
>> completed, checkpointed() call back to Operator is made.
>> 2. For other scenarios, the 1st step happens in the Operator thread and
>> 2nd step is triggered in different thread. During the reportStats call,
>>if
>> the 2nd step is complete checkpointed() call back to Operator is made.
>>
>> Thanks
>> - Gaurav
>>
>> > On Nov 10, 2015, at 3:57 PM, Chetan Narsude <ch...@apache.org> wrote:
>> >
>> > With async checkpointing the Node.reportStats is reporting it back
>>and it
>> > looks like the code is checking for the top of the queue to see if
>>it’s
>> > done (or not reporting at all). So I do not see a reason why they
>>will be
>> > reported out of order.
>> >
>> > Chandni, I read your subsequent responses. Valid point in the last
>>email
>> > about documentation. It amazes me how much the documentation of our
>>API
>> > can be improved (self guilty probably the most). One comment that I
>> wanted
>> > to make even before that is that the semantics of copyToHdfs is that
>>it
>> > either succeeds or throws exception in which case the recovery kicks
>>in
>> > and nothing is reported as checkpointed.
>> >
>> > Tim, we need the unit test, man!
>> >
>> > —
>> > Chetan
>> >
>> >
>> > On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <cn...@cisco.com>
>> > wrote:
>> >
>> >> There are a lot of things which are different when it comes to async
>> >> checkpointing. I was evaluating it in the morning and expect that
>> either I
>> >> am able to explain or open jira issues. With my partial observation
>>is
>> >> that with Async checkpointing, checkpointed is not issued (chandni,
>>the
>> >> last statement in the if block is ³return²). I am digging into it but
>> feel
>> >> free to chime in if someone else is able to find that.
>> >>
>> >> Also I realized that my morning email applies as it is to committed
>>but
>> >> checkpointed has deviated a little bit from that. Will post the
>>revised
>> >> response soon.
>> >>
>> >> ‹
>> >> Chetan
>> >>
>> >>
>> >>
>> >>
>> >> On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com>
>>wrote:
>> >>
>> >>> Chetan,
>> >>>
>> >>> Looking at the checkpoint(windowId) in Node.java, I don't think the
>> steps
>> >>> you mentioned are followed.
>> >>>
>> >>> *if (using AsyncFSStorageAgent) {*
>> >>> * asyncFSStorageAgent.copyToHdfs(...)*
>> >>> *}*
>> >>> *operator.checkpointed(windowId);*
>> >>>
>> >>> This means even copyToHdfs fails the operator is notified that the
>> window
>> >>> is check-pointed.
>> >>>
>> >>> Are we saying that copyToHdfs will never fail with
>>AsyncFSStorageAgent
>> >>> for
>> >>> a window since the operator is notified that the window is
>> checkpointed?
>> >>>
>> >>> Chandni
>> >>>
>> >>> On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas
>><ti...@datatorrent.com>
>> >>> wrote:
>> >>>
>> >>>> Will do
>> >>>>
>> >>>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
>> >>>> <pr...@datatorrent.com>
>> >>>> wrote:
>> >>>>
>> >>>>> Is there a unit test covering it? Otherwise can you write one to
>>test
>> >>>> the
>> >>>>> hypothesis.
>> >>>>>
>> >>>>> On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
>> >>>> <ti...@datatorrent.com>
>> >>>>> wrote:
>> >>>>>
>> >>>>>> That is what it is looking like to me. The task is submitted
>> >>>>>> GenericNode#checkpoint line 504, then at the end of the
>> >>>>>> GenericNode#checkpoint line 531 checkpointed is called. I am
>>likely
>> >>>>> missing
>> >>>>>> something, just would like to know what :)
>> >>>>>>
>> >>>>>> Tim
>> >>>>>>
>> >>>>>> On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
>> >>>>> pramod@datatorrent.com>
>> >>>>>> wrote:
>> >>>>>>
>> >>>>>>> Tim,
>> >>>>>>>
>> >>>>>>> Are you suggesting that checkpointed is called before the
>> >>>> checkpoint
>> >>>> is
>> >>>>>>> completely persisted in the storage.
>> >>>>>>>
>> >>>>>>> Thanks
>> >>>>>>>
>> >>>>>>> On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
>> >>>> tim@datatorrent.com>
>> >>>>>>> wrote:
>> >>>>>>>
>> >>>>>>>> Chetan,
>> >>>>>>>>
>> >>>>>>>> I do not see the process of reporting the checkpoint to stram,
>> >>>>>> receiving
>> >>>>>>>> the ack, and then calling checkpointed. The logic I'm seeing in
>> >>>>>>> GenericNode
>> >>>>>>>> line 484 is that the checkpoint method is called, it spawns
>> >>>> another
>> >>>>>>> thread
>> >>>>>>>> that writes to hdfs, and then checkpointed is immediately
>> >>>> called
>> >>>>>>>> afterwards. I am missing something, can you give me some
>> >>>> pointers
>> >>>> so
>> >>>>>>> that I
>> >>>>>>>> can better understand the flow?
>> >>>>>>>>
>> >>>>>>>> Tim
>> >>>>>>>>
>> >>>>>>>> On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
>> >>>>>> ram@datatorrent.com
>> >>>>>>>>
>> >>>>>>>> wrote:
>> >>>>>>>>
>> >>>>>>>>> Chetan's answer provides a good explanation as well as
>> >>>> clarifying
>> >>>>>> that
>> >>>>>>>>> the difference can be more than 1.
>> >>>>>>>>>
>> >>>>>>>>> Since checkpointing (i.e. "commit notification" as Thomas
>> >>>> refers
>> >>>> to
>> >>>>>>>>> it) is asynchronous, I'm curious
>> >>>>>>>>> about whether the window ids in the checkpointed call are
>> >>>>> guaranteed
>> >>>>>>>>> to be sequential or if they could
>> >>>>>>>>> be out of order, i.e. can the checkpointed call see window id
>> >>>> 101
>> >>>>>>>>> before it sees 100 ?
>> >>>>>>>>>
>> >>>>>>>>> Ram
>> >>>>>>>>>
>> >>>>>>>>> On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
>> >>>>>>>>> <bh...@datatorrent.com> wrote:
>> >>>>>>>>>> Hi Tim,
>> >>>>>>>>>> Thanks for the detailed explanation.
>> >>>>>>>>>> I understand that the sequence would be
>> >>>>>>>>>> beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
>> >>>>>> beginWindow
>> >>>>>>>>>> (x+1)
>> >>>>>>>>>>
>> >>>>>>>>>> However when I try to print out the window ids in
>> >>>> beginWindow,
>> >>>>>>>> endWindow
>> >>>>>>>>>> and checkpointed calls, I see x and x-1 respectively.
>> >>>>>>>>>> I.e. If the window just before checkpoint is 100, I see
>> >>>> that
>> >>>> the
>> >>>>>>>>>> checkpointed call had window id 99.
>> >>>>>>>>>>
>> >>>>>>>>>> Note: This is observed in the local mode of Apex.
>> >>>>>>>>>>
>> >>>>>>>>>> Thanks
>> >>>>>>>>>> -Bhupesh
>> >>>>>>>>>> On 10-Nov-2015 11:25 pm, "Timothy Farkas"
>> >>>> <tim@datatorrent.com
>> >>>>>
>> >>>>>>> wrote:
>> >>>>>>>>>>
>> >>>>>>>>>>> Hi Bhupesh,
>> >>>>>>>>>>>
>> >>>>>>>>>>> The sequencing of checkpoint called in relation to
>> >>>> beginWindow
>> >>>>> and
>> >>>>>>>>>>> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>> >>>>>>>>>>> CHECKPOINT_WINDOW_COUNT are configured. If the two
>> >>>> WINDOW_COUNTs
>> >>>>>> are
>> >>>>>>>> not
>> >>>>>>>>>>> configured to be the same then it is possible that
>> >>>> checkpointed
>> >>>>> is
>> >>>>>>>>> called
>> >>>>>>>>>>> within an application window. So the sequence of events
>> >>>> would
>> >>>> be
>> >>>>>>> this:
>> >>>>>>>>>>>
>> >>>>>>>>>>> beginWindow -> checkpointed -> endWindow
>> >>>>>>>>>>>
>> >>>>>>>>>>> If the APPLICATION_WINDOW_COUNT and
>> >>>> CHECKPOINT_WINDOW_COUNT
>> >>>> are
>> >>>>>> the
>> >>>>>>>> same
>> >>>>>>>>>>> (default). Then the sequence of calls would be this:
>> >>>>>>>>>>>
>> >>>>>>>>>>> beginWindow (100) -> endWindow (100) -> checkpointed
>> >>>> (100)
>> >>>> ->
>> >>>>>>>>> beginWindow
>> >>>>>>>>>>> (101)
>> >>>>>>>>>>>
>> >>>>>>>>>>> The numbers in the sequence represent possible streaming
>> >>>> window
>> >>>>>> Ids
>> >>>>>>>> that
>> >>>>>>>>>>> each call would be associated with.
>> >>>>>>>>>>>
>> >>>>>>>>>>> The StateMachine which calls these callbacks for non-input
>> >>>>>> operators
>> >>>>>>>> is
>> >>>>>>>>> in
>> >>>>>>>>>>> GenericNode.java.
>> >>>>>>>>>>>
>> >>>>>>>>>>> Tim
>> >>>>>>>>>>>
>> >>>>>>>>>>> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
>> >>>>>>>>> bhupesh@datatorrent.com>
>> >>>>>>>>>>> wrote:
>> >>>>>>>>>>>
>> >>>>>>>>>>>> Hi Chetan / Community,
>> >>>>>>>>>>>>
>> >>>>>>>>>>>> Can someone please elaborate on why the window id
>> >>>> supplied
>> >>>> to
>> >>>>>>>>>>>> CheckpointListener and the Operator would differ.
>> >>>>>>>>>>>> I tried looking at the window ids of checkpointed() and
>> >>>> the
>> >>>>>>>>> beginWindow()
>> >>>>>>>>>>>> calls and they differ by 1. Don't know why this should
>> >>>> be
>> >>>> the
>> >>>>>>> case.
>> >>>>>>>>>>>>
>> >>>>>>>>>>>> Thanks.
>> >>>>>>>>>>>> -Bhupesh
>> >>>>>>>>>>>>
>> >>>>>>>>>>>> On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
>> >>>>>>>>> chetan@datatorrent.com>
>> >>>>>>>>>>>> wrote:
>> >>>>>>>>>>>>
>> >>>>>>>>>>>>> Short answer is yes.
>> >>>>>>>>>>>>>
>> >>>>>>>>>>>>> All the control tuples are scheduled to be delivered
>> >>>> outside
>> >>>>>> of
>> >>>>>>>> the
>> >>>>>>>>>>>> window.
>> >>>>>>>>>>>>> As checkpointed callback is triggered because of
>> >>>> CHECKPOINT
>> >>>>>>>> control
>> >>>>>>>>>>>> tuple,
>> >>>>>>>>>>>>> it will happen after endWindow and before the next
>> >>>>>> beginWindow.
>> >>>>>>>>>>>>>
>> >>>>>>>>>>>>> The windowId supplied to CheckpointListener and the
>> >>>> one
>> >>>>>> provided
>> >>>>>>>> to
>> >>>>>>>>>>>>> Operator need not match even though the sequence is
>> >>>> defined.
>> >>>>>> So
>> >>>>>>> I
>> >>>>>>>> am
>> >>>>>>>>>>>>> curious how you intend to use this knowledge.
>> >>>>>>>>>>>>>
>> >>>>>>>>>>>>> --
>> >>>>>>>>>>>>> Chetan
>> >>>>>>>>>>>>>
>> >>>>>>>>>>>>>
>> >>>>>>>>>>>>> On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
>> >>>>>>>>> thomas@datatorrent.com>
>> >>>>>>>>>>>>> wrote:
>> >>>>>>>>>>>>>
>> >>>>>>>>>>>>>> It has not changed the operator execution model.
>> >>>> State
>> >>>>>>>>> serialization
>> >>>>>>>>>>> is
>> >>>>>>>>>>>>>> still synchronous, write to HDFS is taken out of the
>> >>>>>> operator
>> >>>>>>>>> thread.
>> >>>>>>>>>>>>>>
>> >>>>>>>>>>>>>> On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
>> >>>>>>>> amol@datatorrent.com
>> >>>>>>>>>>
>> >>>>>>>>>>>>> wrote:
>> >>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>> Sent too soon. Has asynchronous checkpointing
>> >>>> changed
>> >>>>>> this?
>> >>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>> Amol
>> >>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>> Sent from my iPhone
>> >>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>>> On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>> >>>>>>>>>>>>> bhupesh@datatorrent.com>
>> >>>>>>>>>>>>>>> wrote:
>> >>>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>>> Hi All,
>> >>>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>>> Is it safe to assume that the checkpointed() and
>> >>>> the
>> >>>>>>>>>>> beginWindow()
>> >>>>>>>>>>>>>> calls
>> >>>>>>>>>>>>>>>> are sequenced?
>> >>>>>>>>>>>>>>>> In other words, are these calls part of the same
>> >>>>> thread
>> >>>>>>> and
>> >>>>>>>>> may
>> >>>>>>>>>>> not
>> >>>>>>>>>>>>> run
>> >>>>>>>>>>>>>>> in
>> >>>>>>>>>>>>>>>> parallel?
>> >>>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>>> Thanks.
>> >>>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>>> --
>> >>>>>>>>>>>>>>>> -Bhupesh
>> >>>>>>>>>>>>>>>
>> >>>>>>>>>>>>>>
>> >>>>>>>>>>>>>
>> >>>>>>>>>>>>
>> >>>>>>>>>>>
>> >>>>>>>>>
>> >>>>>>>>
>> >>>>>>>
>> >>>>>>
>> >>>>>
>> >>>>
>> >>
>> >
>> >
>>
>>
Re: Sequencing of operator calls
Posted by Timothy Farkas <ti...@datatorrent.com>.
Hi Gaurav,
I've added some tests to check some of the conditions when checkpointed is
called https://github.com/apache/incubator-apex-core/pull/163
If there are more ideas of things to test I can add more unit tests as well.
Also I was wondering how a failure to copy data from local disk to hdfs is
handled? If an exception is thrown the thread doing the task will die and
then what? Is the task retried?
Thanks,
Tim
On Wed, Nov 11, 2015 at 1:29 AM, Gaurav Gupta <ga...@datatorrent.com>
wrote:
> Chetan,
>
> Your analysis is correct.
>
> I will again summarize how AsyncFSStorageAgent works. It involves two steps
>
> 1. Serializing the Operator state to local disk.
> 2. Moving this serialized state to HDFS.
>
> These are how the above two steps work in different scenario
>
> 1. If PROCESSING is EXACTLY ONCE for an operator, then both the above
> steps happen in the Operator thread. Once the above two steps are
> completed, checkpointed() call back to Operator is made.
> 2. For other scenarios, the 1st step happens in the Operator thread and
> 2nd step is triggered in different thread. During the reportStats call, if
> the 2nd step is complete checkpointed() call back to Operator is made.
>
> Thanks
> - Gaurav
>
> > On Nov 10, 2015, at 3:57 PM, Chetan Narsude <ch...@apache.org> wrote:
> >
> > With async checkpointing the Node.reportStats is reporting it back and it
> > looks like the code is checking for the top of the queue to see if it’s
> > done (or not reporting at all). So I do not see a reason why they will be
> > reported out of order.
> >
> > Chandni, I read your subsequent responses. Valid point in the last email
> > about documentation. It amazes me how much the documentation of our API
> > can be improved (self guilty probably the most). One comment that I
> wanted
> > to make even before that is that the semantics of copyToHdfs is that it
> > either succeeds or throws exception in which case the recovery kicks in
> > and nothing is reported as checkpointed.
> >
> > Tim, we need the unit test, man!
> >
> > —
> > Chetan
> >
> >
> > On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <cn...@cisco.com>
> > wrote:
> >
> >> There are a lot of things which are different when it comes to async
> >> checkpointing. I was evaluating it in the morning and expect that
> either I
> >> am able to explain or open jira issues. With my partial observation is
> >> that with Async checkpointing, checkpointed is not issued (chandni, the
> >> last statement in the if block is ³return²). I am digging into it but
> feel
> >> free to chime in if someone else is able to find that.
> >>
> >> Also I realized that my morning email applies as it is to committed but
> >> checkpointed has deviated a little bit from that. Will post the revised
> >> response soon.
> >>
> >> ‹
> >> Chetan
> >>
> >>
> >>
> >>
> >> On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
> >>
> >>> Chetan,
> >>>
> >>> Looking at the checkpoint(windowId) in Node.java, I don't think the
> steps
> >>> you mentioned are followed.
> >>>
> >>> *if (using AsyncFSStorageAgent) {*
> >>> * asyncFSStorageAgent.copyToHdfs(...)*
> >>> *}*
> >>> *operator.checkpointed(windowId);*
> >>>
> >>> This means even copyToHdfs fails the operator is notified that the
> window
> >>> is check-pointed.
> >>>
> >>> Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent
> >>> for
> >>> a window since the operator is notified that the window is
> checkpointed?
> >>>
> >>> Chandni
> >>>
> >>> On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
> >>> wrote:
> >>>
> >>>> Will do
> >>>>
> >>>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
> >>>> <pr...@datatorrent.com>
> >>>> wrote:
> >>>>
> >>>>> Is there a unit test covering it? Otherwise can you write one to test
> >>>> the
> >>>>> hypothesis.
> >>>>>
> >>>>> On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
> >>>> <ti...@datatorrent.com>
> >>>>> wrote:
> >>>>>
> >>>>>> That is what it is looking like to me. The task is submitted
> >>>>>> GenericNode#checkpoint line 504, then at the end of the
> >>>>>> GenericNode#checkpoint line 531 checkpointed is called. I am likely
> >>>>> missing
> >>>>>> something, just would like to know what :)
> >>>>>>
> >>>>>> Tim
> >>>>>>
> >>>>>> On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> >>>>> pramod@datatorrent.com>
> >>>>>> wrote:
> >>>>>>
> >>>>>>> Tim,
> >>>>>>>
> >>>>>>> Are you suggesting that checkpointed is called before the
> >>>> checkpoint
> >>>> is
> >>>>>>> completely persisted in the storage.
> >>>>>>>
> >>>>>>> Thanks
> >>>>>>>
> >>>>>>> On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> >>>> tim@datatorrent.com>
> >>>>>>> wrote:
> >>>>>>>
> >>>>>>>> Chetan,
> >>>>>>>>
> >>>>>>>> I do not see the process of reporting the checkpoint to stram,
> >>>>>> receiving
> >>>>>>>> the ack, and then calling checkpointed. The logic I'm seeing in
> >>>>>>> GenericNode
> >>>>>>>> line 484 is that the checkpoint method is called, it spawns
> >>>> another
> >>>>>>> thread
> >>>>>>>> that writes to hdfs, and then checkpointed is immediately
> >>>> called
> >>>>>>>> afterwards. I am missing something, can you give me some
> >>>> pointers
> >>>> so
> >>>>>>> that I
> >>>>>>>> can better understand the flow?
> >>>>>>>>
> >>>>>>>> Tim
> >>>>>>>>
> >>>>>>>> On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> >>>>>> ram@datatorrent.com
> >>>>>>>>
> >>>>>>>> wrote:
> >>>>>>>>
> >>>>>>>>> Chetan's answer provides a good explanation as well as
> >>>> clarifying
> >>>>>> that
> >>>>>>>>> the difference can be more than 1.
> >>>>>>>>>
> >>>>>>>>> Since checkpointing (i.e. "commit notification" as Thomas
> >>>> refers
> >>>> to
> >>>>>>>>> it) is asynchronous, I'm curious
> >>>>>>>>> about whether the window ids in the checkpointed call are
> >>>>> guaranteed
> >>>>>>>>> to be sequential or if they could
> >>>>>>>>> be out of order, i.e. can the checkpointed call see window id
> >>>> 101
> >>>>>>>>> before it sees 100 ?
> >>>>>>>>>
> >>>>>>>>> Ram
> >>>>>>>>>
> >>>>>>>>> On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> >>>>>>>>> <bh...@datatorrent.com> wrote:
> >>>>>>>>>> Hi Tim,
> >>>>>>>>>> Thanks for the detailed explanation.
> >>>>>>>>>> I understand that the sequence would be
> >>>>>>>>>> beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> >>>>>> beginWindow
> >>>>>>>>>> (x+1)
> >>>>>>>>>>
> >>>>>>>>>> However when I try to print out the window ids in
> >>>> beginWindow,
> >>>>>>>> endWindow
> >>>>>>>>>> and checkpointed calls, I see x and x-1 respectively.
> >>>>>>>>>> I.e. If the window just before checkpoint is 100, I see
> >>>> that
> >>>> the
> >>>>>>>>>> checkpointed call had window id 99.
> >>>>>>>>>>
> >>>>>>>>>> Note: This is observed in the local mode of Apex.
> >>>>>>>>>>
> >>>>>>>>>> Thanks
> >>>>>>>>>> -Bhupesh
> >>>>>>>>>> On 10-Nov-2015 11:25 pm, "Timothy Farkas"
> >>>> <tim@datatorrent.com
> >>>>>
> >>>>>>> wrote:
> >>>>>>>>>>
> >>>>>>>>>>> Hi Bhupesh,
> >>>>>>>>>>>
> >>>>>>>>>>> The sequencing of checkpoint called in relation to
> >>>> beginWindow
> >>>>> and
> >>>>>>>>>>> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> >>>>>>>>>>> CHECKPOINT_WINDOW_COUNT are configured. If the two
> >>>> WINDOW_COUNTs
> >>>>>> are
> >>>>>>>> not
> >>>>>>>>>>> configured to be the same then it is possible that
> >>>> checkpointed
> >>>>> is
> >>>>>>>>> called
> >>>>>>>>>>> within an application window. So the sequence of events
> >>>> would
> >>>> be
> >>>>>>> this:
> >>>>>>>>>>>
> >>>>>>>>>>> beginWindow -> checkpointed -> endWindow
> >>>>>>>>>>>
> >>>>>>>>>>> If the APPLICATION_WINDOW_COUNT and
> >>>> CHECKPOINT_WINDOW_COUNT
> >>>> are
> >>>>>> the
> >>>>>>>> same
> >>>>>>>>>>> (default). Then the sequence of calls would be this:
> >>>>>>>>>>>
> >>>>>>>>>>> beginWindow (100) -> endWindow (100) -> checkpointed
> >>>> (100)
> >>>> ->
> >>>>>>>>> beginWindow
> >>>>>>>>>>> (101)
> >>>>>>>>>>>
> >>>>>>>>>>> The numbers in the sequence represent possible streaming
> >>>> window
> >>>>>> Ids
> >>>>>>>> that
> >>>>>>>>>>> each call would be associated with.
> >>>>>>>>>>>
> >>>>>>>>>>> The StateMachine which calls these callbacks for non-input
> >>>>>> operators
> >>>>>>>> is
> >>>>>>>>> in
> >>>>>>>>>>> GenericNode.java.
> >>>>>>>>>>>
> >>>>>>>>>>> Tim
> >>>>>>>>>>>
> >>>>>>>>>>> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> >>>>>>>>> bhupesh@datatorrent.com>
> >>>>>>>>>>> wrote:
> >>>>>>>>>>>
> >>>>>>>>>>>> Hi Chetan / Community,
> >>>>>>>>>>>>
> >>>>>>>>>>>> Can someone please elaborate on why the window id
> >>>> supplied
> >>>> to
> >>>>>>>>>>>> CheckpointListener and the Operator would differ.
> >>>>>>>>>>>> I tried looking at the window ids of checkpointed() and
> >>>> the
> >>>>>>>>> beginWindow()
> >>>>>>>>>>>> calls and they differ by 1. Don't know why this should
> >>>> be
> >>>> the
> >>>>>>> case.
> >>>>>>>>>>>>
> >>>>>>>>>>>> Thanks.
> >>>>>>>>>>>> -Bhupesh
> >>>>>>>>>>>>
> >>>>>>>>>>>> On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> >>>>>>>>> chetan@datatorrent.com>
> >>>>>>>>>>>> wrote:
> >>>>>>>>>>>>
> >>>>>>>>>>>>> Short answer is yes.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> All the control tuples are scheduled to be delivered
> >>>> outside
> >>>>>> of
> >>>>>>>> the
> >>>>>>>>>>>> window.
> >>>>>>>>>>>>> As checkpointed callback is triggered because of
> >>>> CHECKPOINT
> >>>>>>>> control
> >>>>>>>>>>>> tuple,
> >>>>>>>>>>>>> it will happen after endWindow and before the next
> >>>>>> beginWindow.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> The windowId supplied to CheckpointListener and the
> >>>> one
> >>>>>> provided
> >>>>>>>> to
> >>>>>>>>>>>>> Operator need not match even though the sequence is
> >>>> defined.
> >>>>>> So
> >>>>>>> I
> >>>>>>>> am
> >>>>>>>>>>>>> curious how you intend to use this knowledge.
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> --
> >>>>>>>>>>>>> Chetan
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>> On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> >>>>>>>>> thomas@datatorrent.com>
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>
> >>>>>>>>>>>>>> It has not changed the operator execution model.
> >>>> State
> >>>>>>>>> serialization
> >>>>>>>>>>> is
> >>>>>>>>>>>>>> still synchronous, write to HDFS is taken out of the
> >>>>>> operator
> >>>>>>>>> thread.
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>> On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> >>>>>>>> amol@datatorrent.com
> >>>>>>>>>>
> >>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Sent too soon. Has asynchronous checkpointing
> >>>> changed
> >>>>>> this?
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Amol
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>> Sent from my iPhone
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> >>>>>>>>>>>>> bhupesh@datatorrent.com>
> >>>>>>>>>>>>>>> wrote:
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Hi All,
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Is it safe to assume that the checkpointed() and
> >>>> the
> >>>>>>>>>>> beginWindow()
> >>>>>>>>>>>>>> calls
> >>>>>>>>>>>>>>>> are sequenced?
> >>>>>>>>>>>>>>>> In other words, are these calls part of the same
> >>>>> thread
> >>>>>>> and
> >>>>>>>>> may
> >>>>>>>>>>> not
> >>>>>>>>>>>>> run
> >>>>>>>>>>>>>>> in
> >>>>>>>>>>>>>>>> parallel?
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> Thanks.
> >>>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>>> --
> >>>>>>>>>>>>>>>> -Bhupesh
> >>>>>>>>>>>>>>>
> >>>>>>>>>>>>>>
> >>>>>>>>>>>>>
> >>>>>>>>>>>>
> >>>>>>>>>>>
> >>>>>>>>>
> >>>>>>>>
> >>>>>>>
> >>>>>>
> >>>>>
> >>>>
> >>
> >
> >
>
>
Re: Sequencing of operator calls
Posted by Gaurav Gupta <ga...@datatorrent.com>.
Chetan,
Your analysis is correct.
I will again summarize how AsyncFSStorageAgent works. It involves two steps
1. Serializing the Operator state to local disk.
2. Moving this serialized state to HDFS.
These are how the above two steps work in different scenario
1. If PROCESSING is EXACTLY ONCE for an operator, then both the above steps happen in the Operator thread. Once the above two steps are completed, checkpointed() call back to Operator is made.
2. For other scenarios, the 1st step happens in the Operator thread and 2nd step is triggered in different thread. During the reportStats call, if the 2nd step is complete checkpointed() call back to Operator is made.
Thanks
- Gaurav
> On Nov 10, 2015, at 3:57 PM, Chetan Narsude <ch...@apache.org> wrote:
>
> With async checkpointing the Node.reportStats is reporting it back and it
> looks like the code is checking for the top of the queue to see if it’s
> done (or not reporting at all). So I do not see a reason why they will be
> reported out of order.
>
> Chandni, I read your subsequent responses. Valid point in the last email
> about documentation. It amazes me how much the documentation of our API
> can be improved (self guilty probably the most). One comment that I wanted
> to make even before that is that the semantics of copyToHdfs is that it
> either succeeds or throws exception in which case the recovery kicks in
> and nothing is reported as checkpointed.
>
> Tim, we need the unit test, man!
>
> —
> Chetan
>
>
> On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <cn...@cisco.com>
> wrote:
>
>> There are a lot of things which are different when it comes to async
>> checkpointing. I was evaluating it in the morning and expect that either I
>> am able to explain or open jira issues. With my partial observation is
>> that with Async checkpointing, checkpointed is not issued (chandni, the
>> last statement in the if block is ³return²). I am digging into it but feel
>> free to chime in if someone else is able to find that.
>>
>> Also I realized that my morning email applies as it is to committed but
>> checkpointed has deviated a little bit from that. Will post the revised
>> response soon.
>>
>> ‹
>> Chetan
>>
>>
>>
>>
>> On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
>>
>>> Chetan,
>>>
>>> Looking at the checkpoint(windowId) in Node.java, I don't think the steps
>>> you mentioned are followed.
>>>
>>> *if (using AsyncFSStorageAgent) {*
>>> * asyncFSStorageAgent.copyToHdfs(...)*
>>> *}*
>>> *operator.checkpointed(windowId);*
>>>
>>> This means even copyToHdfs fails the operator is notified that the window
>>> is check-pointed.
>>>
>>> Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent
>>> for
>>> a window since the operator is notified that the window is checkpointed?
>>>
>>> Chandni
>>>
>>> On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
>>> wrote:
>>>
>>>> Will do
>>>>
>>>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
>>>> <pr...@datatorrent.com>
>>>> wrote:
>>>>
>>>>> Is there a unit test covering it? Otherwise can you write one to test
>>>> the
>>>>> hypothesis.
>>>>>
>>>>> On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
>>>> <ti...@datatorrent.com>
>>>>> wrote:
>>>>>
>>>>>> That is what it is looking like to me. The task is submitted
>>>>>> GenericNode#checkpoint line 504, then at the end of the
>>>>>> GenericNode#checkpoint line 531 checkpointed is called. I am likely
>>>>> missing
>>>>>> something, just would like to know what :)
>>>>>>
>>>>>> Tim
>>>>>>
>>>>>> On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
>>>>> pramod@datatorrent.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Tim,
>>>>>>>
>>>>>>> Are you suggesting that checkpointed is called before the
>>>> checkpoint
>>>> is
>>>>>>> completely persisted in the storage.
>>>>>>>
>>>>>>> Thanks
>>>>>>>
>>>>>>> On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
>>>> tim@datatorrent.com>
>>>>>>> wrote:
>>>>>>>
>>>>>>>> Chetan,
>>>>>>>>
>>>>>>>> I do not see the process of reporting the checkpoint to stram,
>>>>>> receiving
>>>>>>>> the ack, and then calling checkpointed. The logic I'm seeing in
>>>>>>> GenericNode
>>>>>>>> line 484 is that the checkpoint method is called, it spawns
>>>> another
>>>>>>> thread
>>>>>>>> that writes to hdfs, and then checkpointed is immediately
>>>> called
>>>>>>>> afterwards. I am missing something, can you give me some
>>>> pointers
>>>> so
>>>>>>> that I
>>>>>>>> can better understand the flow?
>>>>>>>>
>>>>>>>> Tim
>>>>>>>>
>>>>>>>> On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
>>>>>> ram@datatorrent.com
>>>>>>>>
>>>>>>>> wrote:
>>>>>>>>
>>>>>>>>> Chetan's answer provides a good explanation as well as
>>>> clarifying
>>>>>> that
>>>>>>>>> the difference can be more than 1.
>>>>>>>>>
>>>>>>>>> Since checkpointing (i.e. "commit notification" as Thomas
>>>> refers
>>>> to
>>>>>>>>> it) is asynchronous, I'm curious
>>>>>>>>> about whether the window ids in the checkpointed call are
>>>>> guaranteed
>>>>>>>>> to be sequential or if they could
>>>>>>>>> be out of order, i.e. can the checkpointed call see window id
>>>> 101
>>>>>>>>> before it sees 100 ?
>>>>>>>>>
>>>>>>>>> Ram
>>>>>>>>>
>>>>>>>>> On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
>>>>>>>>> <bh...@datatorrent.com> wrote:
>>>>>>>>>> Hi Tim,
>>>>>>>>>> Thanks for the detailed explanation.
>>>>>>>>>> I understand that the sequence would be
>>>>>>>>>> beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
>>>>>> beginWindow
>>>>>>>>>> (x+1)
>>>>>>>>>>
>>>>>>>>>> However when I try to print out the window ids in
>>>> beginWindow,
>>>>>>>> endWindow
>>>>>>>>>> and checkpointed calls, I see x and x-1 respectively.
>>>>>>>>>> I.e. If the window just before checkpoint is 100, I see
>>>> that
>>>> the
>>>>>>>>>> checkpointed call had window id 99.
>>>>>>>>>>
>>>>>>>>>> Note: This is observed in the local mode of Apex.
>>>>>>>>>>
>>>>>>>>>> Thanks
>>>>>>>>>> -Bhupesh
>>>>>>>>>> On 10-Nov-2015 11:25 pm, "Timothy Farkas"
>>>> <tim@datatorrent.com
>>>>>
>>>>>>> wrote:
>>>>>>>>>>
>>>>>>>>>>> Hi Bhupesh,
>>>>>>>>>>>
>>>>>>>>>>> The sequencing of checkpoint called in relation to
>>>> beginWindow
>>>>> and
>>>>>>>>>>> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>>>>>>>>>>> CHECKPOINT_WINDOW_COUNT are configured. If the two
>>>> WINDOW_COUNTs
>>>>>> are
>>>>>>>> not
>>>>>>>>>>> configured to be the same then it is possible that
>>>> checkpointed
>>>>> is
>>>>>>>>> called
>>>>>>>>>>> within an application window. So the sequence of events
>>>> would
>>>> be
>>>>>>> this:
>>>>>>>>>>>
>>>>>>>>>>> beginWindow -> checkpointed -> endWindow
>>>>>>>>>>>
>>>>>>>>>>> If the APPLICATION_WINDOW_COUNT and
>>>> CHECKPOINT_WINDOW_COUNT
>>>> are
>>>>>> the
>>>>>>>> same
>>>>>>>>>>> (default). Then the sequence of calls would be this:
>>>>>>>>>>>
>>>>>>>>>>> beginWindow (100) -> endWindow (100) -> checkpointed
>>>> (100)
>>>> ->
>>>>>>>>> beginWindow
>>>>>>>>>>> (101)
>>>>>>>>>>>
>>>>>>>>>>> The numbers in the sequence represent possible streaming
>>>> window
>>>>>> Ids
>>>>>>>> that
>>>>>>>>>>> each call would be associated with.
>>>>>>>>>>>
>>>>>>>>>>> The StateMachine which calls these callbacks for non-input
>>>>>> operators
>>>>>>>> is
>>>>>>>>> in
>>>>>>>>>>> GenericNode.java.
>>>>>>>>>>>
>>>>>>>>>>> Tim
>>>>>>>>>>>
>>>>>>>>>>> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
>>>>>>>>> bhupesh@datatorrent.com>
>>>>>>>>>>> wrote:
>>>>>>>>>>>
>>>>>>>>>>>> Hi Chetan / Community,
>>>>>>>>>>>>
>>>>>>>>>>>> Can someone please elaborate on why the window id
>>>> supplied
>>>> to
>>>>>>>>>>>> CheckpointListener and the Operator would differ.
>>>>>>>>>>>> I tried looking at the window ids of checkpointed() and
>>>> the
>>>>>>>>> beginWindow()
>>>>>>>>>>>> calls and they differ by 1. Don't know why this should
>>>> be
>>>> the
>>>>>>> case.
>>>>>>>>>>>>
>>>>>>>>>>>> Thanks.
>>>>>>>>>>>> -Bhupesh
>>>>>>>>>>>>
>>>>>>>>>>>> On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
>>>>>>>>> chetan@datatorrent.com>
>>>>>>>>>>>> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>> Short answer is yes.
>>>>>>>>>>>>>
>>>>>>>>>>>>> All the control tuples are scheduled to be delivered
>>>> outside
>>>>>> of
>>>>>>>> the
>>>>>>>>>>>> window.
>>>>>>>>>>>>> As checkpointed callback is triggered because of
>>>> CHECKPOINT
>>>>>>>> control
>>>>>>>>>>>> tuple,
>>>>>>>>>>>>> it will happen after endWindow and before the next
>>>>>> beginWindow.
>>>>>>>>>>>>>
>>>>>>>>>>>>> The windowId supplied to CheckpointListener and the
>>>> one
>>>>>> provided
>>>>>>>> to
>>>>>>>>>>>>> Operator need not match even though the sequence is
>>>> defined.
>>>>>> So
>>>>>>> I
>>>>>>>> am
>>>>>>>>>>>>> curious how you intend to use this knowledge.
>>>>>>>>>>>>>
>>>>>>>>>>>>> --
>>>>>>>>>>>>> Chetan
>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>> On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
>>>>>>>>> thomas@datatorrent.com>
>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> It has not changed the operator execution model.
>>>> State
>>>>>>>>> serialization
>>>>>>>>>>> is
>>>>>>>>>>>>>> still synchronous, write to HDFS is taken out of the
>>>>>> operator
>>>>>>>>> thread.
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
>>>>>>>> amol@datatorrent.com
>>>>>>>>>>
>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Sent too soon. Has asynchronous checkpointing
>>>> changed
>>>>>> this?
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Amol
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Sent from my iPhone
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>>>>>>>>>>>>> bhupesh@datatorrent.com>
>>>>>>>>>>>>>>> wrote:
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Hi All,
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Is it safe to assume that the checkpointed() and
>>>> the
>>>>>>>>>>> beginWindow()
>>>>>>>>>>>>>> calls
>>>>>>>>>>>>>>>> are sequenced?
>>>>>>>>>>>>>>>> In other words, are these calls part of the same
>>>>> thread
>>>>>>> and
>>>>>>>>> may
>>>>>>>>>>> not
>>>>>>>>>>>>> run
>>>>>>>>>>>>>>> in
>>>>>>>>>>>>>>>> parallel?
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> Thanks.
>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>> --
>>>>>>>>>>>>>>>> -Bhupesh
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>
>
>
Re: Sequencing of operator calls
Posted by Chetan Narsude <ch...@apache.org>.
With async checkpointing the Node.reportStats is reporting it back and it
looks like the code is checking for the top of the queue to see if it’s
done (or not reporting at all). So I do not see a reason why they will be
reported out of order.
Chandni, I read your subsequent responses. Valid point in the last email
about documentation. It amazes me how much the documentation of our API
can be improved (self guilty probably the most). One comment that I wanted
to make even before that is that the semantics of copyToHdfs is that it
either succeeds or throws exception in which case the recovery kicks in
and nothing is reported as checkpointed.
Tim, we need the unit test, man!
—
Chetan
On 11/10/15, 3:05 PM, "Chetan Narsude (cnarsude)" <cn...@cisco.com>
wrote:
>There are a lot of things which are different when it comes to async
>checkpointing. I was evaluating it in the morning and expect that either I
>am able to explain or open jira issues. With my partial observation is
>that with Async checkpointing, checkpointed is not issued (chandni, the
>last statement in the if block is ³return²). I am digging into it but feel
>free to chime in if someone else is able to find that.
>
>Also I realized that my morning email applies as it is to committed but
>checkpointed has deviated a little bit from that. Will post the revised
>response soon.
>
>‹
>Chetan
>
>
>
>
>On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
>
>>Chetan,
>>
>>Looking at the checkpoint(windowId) in Node.java, I don't think the steps
>>you mentioned are followed.
>>
>>*if (using AsyncFSStorageAgent) {*
>>* asyncFSStorageAgent.copyToHdfs(...)*
>>*}*
>>*operator.checkpointed(windowId);*
>>
>>This means even copyToHdfs fails the operator is notified that the window
>>is check-pointed.
>>
>>Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent
>>for
>>a window since the operator is notified that the window is checkpointed?
>>
>>Chandni
>>
>>On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
>>wrote:
>>
>>> Will do
>>>
>>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
>>><pr...@datatorrent.com>
>>> wrote:
>>>
>>> > Is there a unit test covering it? Otherwise can you write one to test
>>>the
>>> > hypothesis.
>>> >
>>> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas
>>><ti...@datatorrent.com>
>>> > wrote:
>>> >
>>> > > That is what it is looking like to me. The task is submitted
>>> > > GenericNode#checkpoint line 504, then at the end of the
>>> > > GenericNode#checkpoint line 531 checkpointed is called. I am likely
>>> > missing
>>> > > something, just would like to know what :)
>>> > >
>>> > > Tim
>>> > >
>>> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
>>> > pramod@datatorrent.com>
>>> > > wrote:
>>> > >
>>> > > > Tim,
>>> > > >
>>> > > > Are you suggesting that checkpointed is called before the
>>>checkpoint
>>> is
>>> > > > completely persisted in the storage.
>>> > > >
>>> > > > Thanks
>>> > > >
>>> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
>>> tim@datatorrent.com>
>>> > > > wrote:
>>> > > >
>>> > > > > Chetan,
>>> > > > >
>>> > > > > I do not see the process of reporting the checkpoint to stram,
>>> > > receiving
>>> > > > > the ack, and then calling checkpointed. The logic I'm seeing in
>>> > > > GenericNode
>>> > > > > line 484 is that the checkpoint method is called, it spawns
>>>another
>>> > > > thread
>>> > > > > that writes to hdfs, and then checkpointed is immediately
>>>called
>>> > > > > afterwards. I am missing something, can you give me some
>>>pointers
>>> so
>>> > > > that I
>>> > > > > can better understand the flow?
>>> > > > >
>>> > > > > Tim
>>> > > > >
>>> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
>>> > > ram@datatorrent.com
>>> > > > >
>>> > > > > wrote:
>>> > > > >
>>> > > > > > Chetan's answer provides a good explanation as well as
>>>clarifying
>>> > > that
>>> > > > > > the difference can be more than 1.
>>> > > > > >
>>> > > > > > Since checkpointing (i.e. "commit notification" as Thomas
>>>refers
>>> to
>>> > > > > > it) is asynchronous, I'm curious
>>> > > > > > about whether the window ids in the checkpointed call are
>>> > guaranteed
>>> > > > > > to be sequential or if they could
>>> > > > > > be out of order, i.e. can the checkpointed call see window id
>>>101
>>> > > > > > before it sees 100 ?
>>> > > > > >
>>> > > > > > Ram
>>> > > > > >
>>> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
>>> > > > > > <bh...@datatorrent.com> wrote:
>>> > > > > > > Hi Tim,
>>> > > > > > > Thanks for the detailed explanation.
>>> > > > > > > I understand that the sequence would be
>>> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
>>> > > beginWindow
>>> > > > > > > (x+1)
>>> > > > > > >
>>> > > > > > > However when I try to print out the window ids in
>>>beginWindow,
>>> > > > > endWindow
>>> > > > > > > and checkpointed calls, I see x and x-1 respectively.
>>> > > > > > > I.e. If the window just before checkpoint is 100, I see
>>>that
>>> the
>>> > > > > > > checkpointed call had window id 99.
>>> > > > > > >
>>> > > > > > > Note: This is observed in the local mode of Apex.
>>> > > > > > >
>>> > > > > > > Thanks
>>> > > > > > > -Bhupesh
>>> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
>>><tim@datatorrent.com
>>> >
>>> > > > wrote:
>>> > > > > > >
>>> > > > > > >> Hi Bhupesh,
>>> > > > > > >>
>>> > > > > > >> The sequencing of checkpoint called in relation to
>>>beginWindow
>>> > and
>>> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>>> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
>>> WINDOW_COUNTs
>>> > > are
>>> > > > > not
>>> > > > > > >> configured to be the same then it is possible that
>>> checkpointed
>>> > is
>>> > > > > > called
>>> > > > > > >> within an application window. So the sequence of events
>>>would
>>> be
>>> > > > this:
>>> > > > > > >>
>>> > > > > > >> beginWindow -> checkpointed -> endWindow
>>> > > > > > >>
>>> > > > > > >> If the APPLICATION_WINDOW_COUNT and
>>>CHECKPOINT_WINDOW_COUNT
>>> are
>>> > > the
>>> > > > > same
>>> > > > > > >> (default). Then the sequence of calls would be this:
>>> > > > > > >>
>>> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed
>>>(100)
>>> ->
>>> > > > > > beginWindow
>>> > > > > > >> (101)
>>> > > > > > >>
>>> > > > > > >> The numbers in the sequence represent possible streaming
>>> window
>>> > > Ids
>>> > > > > that
>>> > > > > > >> each call would be associated with.
>>> > > > > > >>
>>> > > > > > >> The StateMachine which calls these callbacks for non-input
>>> > > operators
>>> > > > > is
>>> > > > > > in
>>> > > > > > >> GenericNode.java.
>>> > > > > > >>
>>> > > > > > >> Tim
>>> > > > > > >>
>>> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
>>> > > > > > bhupesh@datatorrent.com>
>>> > > > > > >> wrote:
>>> > > > > > >>
>>> > > > > > >> > Hi Chetan / Community,
>>> > > > > > >> >
>>> > > > > > >> > Can someone please elaborate on why the window id
>>>supplied
>>> to
>>> > > > > > >> > CheckpointListener and the Operator would differ.
>>> > > > > > >> > I tried looking at the window ids of checkpointed() and
>>>the
>>> > > > > > beginWindow()
>>> > > > > > >> > calls and they differ by 1. Don't know why this should
>>>be
>>> the
>>> > > > case.
>>> > > > > > >> >
>>> > > > > > >> > Thanks.
>>> > > > > > >> > -Bhupesh
>>> > > > > > >> >
>>> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
>>> > > > > > chetan@datatorrent.com>
>>> > > > > > >> > wrote:
>>> > > > > > >> >
>>> > > > > > >> > > Short answer is yes.
>>> > > > > > >> > >
>>> > > > > > >> > > All the control tuples are scheduled to be delivered
>>> outside
>>> > > of
>>> > > > > the
>>> > > > > > >> > window.
>>> > > > > > >> > > As checkpointed callback is triggered because of
>>> CHECKPOINT
>>> > > > > control
>>> > > > > > >> > tuple,
>>> > > > > > >> > > it will happen after endWindow and before the next
>>> > > beginWindow.
>>> > > > > > >> > >
>>> > > > > > >> > > The windowId supplied to CheckpointListener and the
>>>one
>>> > > provided
>>> > > > > to
>>> > > > > > >> > > Operator need not match even though the sequence is
>>> defined.
>>> > > So
>>> > > > I
>>> > > > > am
>>> > > > > > >> > > curious how you intend to use this knowledge.
>>> > > > > > >> > >
>>> > > > > > >> > > --
>>> > > > > > >> > > Chetan
>>> > > > > > >> > >
>>> > > > > > >> > >
>>> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
>>> > > > > > thomas@datatorrent.com>
>>> > > > > > >> > > wrote:
>>> > > > > > >> > >
>>> > > > > > >> > > > It has not changed the operator execution model.
>>>State
>>> > > > > > serialization
>>> > > > > > >> is
>>> > > > > > >> > > > still synchronous, write to HDFS is taken out of the
>>> > > operator
>>> > > > > > thread.
>>> > > > > > >> > > >
>>> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
>>> > > > > amol@datatorrent.com
>>> > > > > > >
>>> > > > > > >> > > wrote:
>>> > > > > > >> > > >
>>> > > > > > >> > > > >
>>> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
>>>changed
>>> > > this?
>>> > > > > > >> > > > >
>>> > > > > > >> > > > > Amol
>>> > > > > > >> > > > >
>>> > > > > > >> > > > > Sent from my iPhone
>>> > > > > > >> > > > >
>>> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>>> > > > > > >> > > bhupesh@datatorrent.com>
>>> > > > > > >> > > > > wrote:
>>> > > > > > >> > > > > >
>>> > > > > > >> > > > > > Hi All,
>>> > > > > > >> > > > > >
>>> > > > > > >> > > > > > Is it safe to assume that the checkpointed() and
>>>the
>>> > > > > > >> beginWindow()
>>> > > > > > >> > > > calls
>>> > > > > > >> > > > > > are sequenced?
>>> > > > > > >> > > > > > In other words, are these calls part of the same
>>> > thread
>>> > > > and
>>> > > > > > may
>>> > > > > > >> not
>>> > > > > > >> > > run
>>> > > > > > >> > > > > in
>>> > > > > > >> > > > > > parallel?
>>> > > > > > >> > > > > >
>>> > > > > > >> > > > > > Thanks.
>>> > > > > > >> > > > > >
>>> > > > > > >> > > > > > --
>>> > > > > > >> > > > > > -Bhupesh
>>> > > > > > >> > > > >
>>> > > > > > >> > > >
>>> > > > > > >> > >
>>> > > > > > >> >
>>> > > > > > >>
>>> > > > > >
>>> > > > >
>>> > > >
>>> > >
>>> >
>>>
>
Re: Sequencing of operator calls
Posted by Chandni Singh <ch...@datatorrent.com>.
So I guess AsyncFSStorageAgent.copyToHdfs is a synchronous write.
My mistake. However I think AsyncFSStorageAgent needs to be documented
properly. Even the public methods have no documentation also this piece of
code in GenericNode.
For Exactly once, we are enforcing synchronous checkpointing which is ok.
Chandni
On Tue, Nov 10, 2015 at 3:20 PM, Chandni Singh <ch...@datatorrent.com>
wrote:
> Chetan,
>
> I am pasting the code here and highlighting the lines. The last statement
> is NOT a 'return' if Processing_Mode == EXACTLY_ONCE
>
>
> void checkpoint(long windowId)
> {
> if (!context.stateless) {
> StorageAgent ba = context.getValue(OperatorContext.STORAGE_AGENT);
> if (ba != null) {
> try {
> checkpointStats = new Stats.CheckpointStats();
> checkpointStats.checkpointStartTime = System.currentTimeMillis();
> ba.save(operator, id, windowId);
> if (ba instanceof AsyncFSStorageAgent) {
> AsyncFSStorageAgent asyncFSStorageAgent = (AsyncFSStorageAgent) ba;
> if (!asyncFSStorageAgent.isSyncCheckpoint()) {
> if(PROCESSING_MODE != ProcessingMode.EXACTLY_ONCE) {
> CheckpointHandler checkpointHandler = new CheckpointHandler();
> checkpointHandler.agent = asyncFSStorageAgent;
> checkpointHandler.operatorId = id;
> checkpointHandler.windowId = windowId;
> checkpointHandler.stats = checkpointStats;
> FutureTask<Stats.CheckpointStats> futureTask = new FutureTask<Stats.CheckpointStats>(checkpointHandler);
> taskQueue.add(new Pair<FutureTask<Stats.CheckpointStats>, Long>(futureTask, windowId));
> executorService.submit(futureTask);
> checkpoint = null;
> checkpointStats = null;
> return;
> }else{
>
> * asyncFSStorageAgent.copyToHDFS(id, windowId);* }
> }
> }
> checkpointStats.checkpointTime = System.currentTimeMillis() - checkpointStats.checkpointStartTime;
> }
> catch (IOException ie) {
> try {
> logger.warn("Rolling back checkpoint {} for Operator {} due to the exception {}",
> Codec.getStringWindowId(windowId), operator, ie);
> ba.delete(id, windowId);
> }
> catch (IOException ex) {
> logger.warn("Error while rolling back checkpoint", ex);
> }
> throw new RuntimeException(ie);
> }
> }
> }
>
> checkpoint = new Checkpoint(windowId, applicationWindowCount, checkpointWindowCount);
> if (operator instanceof Operator.CheckpointListener) {
>
> * ((Operator.CheckpointListener) operator).checkpointed(windowId);* }
> }
>
>
> Chandni
>
>
> On Tue, Nov 10, 2015 at 3:05 PM, Chetan Narsude (cnarsude) <
> cnarsude@cisco.com> wrote:
>
>> There are a lot of things which are different when it comes to async
>> checkpointing. I was evaluating it in the morning and expect that either I
>> am able to explain or open jira issues. With my partial observation is
>> that with Async checkpointing, checkpointed is not issued (chandni, the
>> last statement in the if block is ³return²). I am digging into it but feel
>> free to chime in if someone else is able to find that.
>>
>> Also I realized that my morning email applies as it is to committed but
>> checkpointed has deviated a little bit from that. Will post the revised
>> response soon.
>>
>> ‹
>> Chetan
>>
>>
>>
>>
>> On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
>>
>> >Chetan,
>> >
>> >Looking at the checkpoint(windowId) in Node.java, I don't think the steps
>> >you mentioned are followed.
>> >
>> >*if (using AsyncFSStorageAgent) {*
>> >* asyncFSStorageAgent.copyToHdfs(...)*
>> >*}*
>> >*operator.checkpointed(windowId);*
>> >
>> >This means even copyToHdfs fails the operator is notified that the window
>> >is check-pointed.
>> >
>> >Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent
>> for
>> >a window since the operator is notified that the window is checkpointed?
>> >
>> >Chandni
>> >
>> >On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
>> >wrote:
>> >
>> >> Will do
>> >>
>> >> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
>> >><pr...@datatorrent.com>
>> >> wrote:
>> >>
>> >> > Is there a unit test covering it? Otherwise can you write one to test
>> >>the
>> >> > hypothesis.
>> >> >
>> >> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas <
>> tim@datatorrent.com>
>> >> > wrote:
>> >> >
>> >> > > That is what it is looking like to me. The task is submitted
>> >> > > GenericNode#checkpoint line 504, then at the end of the
>> >> > > GenericNode#checkpoint line 531 checkpointed is called. I am likely
>> >> > missing
>> >> > > something, just would like to know what :)
>> >> > >
>> >> > > Tim
>> >> > >
>> >> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
>> >> > pramod@datatorrent.com>
>> >> > > wrote:
>> >> > >
>> >> > > > Tim,
>> >> > > >
>> >> > > > Are you suggesting that checkpointed is called before the
>> >>checkpoint
>> >> is
>> >> > > > completely persisted in the storage.
>> >> > > >
>> >> > > > Thanks
>> >> > > >
>> >> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
>> >> tim@datatorrent.com>
>> >> > > > wrote:
>> >> > > >
>> >> > > > > Chetan,
>> >> > > > >
>> >> > > > > I do not see the process of reporting the checkpoint to stram,
>> >> > > receiving
>> >> > > > > the ack, and then calling checkpointed. The logic I'm seeing in
>> >> > > > GenericNode
>> >> > > > > line 484 is that the checkpoint method is called, it spawns
>> >>another
>> >> > > > thread
>> >> > > > > that writes to hdfs, and then checkpointed is immediately
>> called
>> >> > > > > afterwards. I am missing something, can you give me some
>> >>pointers
>> >> so
>> >> > > > that I
>> >> > > > > can better understand the flow?
>> >> > > > >
>> >> > > > > Tim
>> >> > > > >
>> >> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
>> >> > > ram@datatorrent.com
>> >> > > > >
>> >> > > > > wrote:
>> >> > > > >
>> >> > > > > > Chetan's answer provides a good explanation as well as
>> >>clarifying
>> >> > > that
>> >> > > > > > the difference can be more than 1.
>> >> > > > > >
>> >> > > > > > Since checkpointing (i.e. "commit notification" as Thomas
>> >>refers
>> >> to
>> >> > > > > > it) is asynchronous, I'm curious
>> >> > > > > > about whether the window ids in the checkpointed call are
>> >> > guaranteed
>> >> > > > > > to be sequential or if they could
>> >> > > > > > be out of order, i.e. can the checkpointed call see window id
>> >>101
>> >> > > > > > before it sees 100 ?
>> >> > > > > >
>> >> > > > > > Ram
>> >> > > > > >
>> >> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
>> >> > > > > > <bh...@datatorrent.com> wrote:
>> >> > > > > > > Hi Tim,
>> >> > > > > > > Thanks for the detailed explanation.
>> >> > > > > > > I understand that the sequence would be
>> >> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
>> >> > > beginWindow
>> >> > > > > > > (x+1)
>> >> > > > > > >
>> >> > > > > > > However when I try to print out the window ids in
>> >>beginWindow,
>> >> > > > > endWindow
>> >> > > > > > > and checkpointed calls, I see x and x-1 respectively.
>> >> > > > > > > I.e. If the window just before checkpoint is 100, I see
>> that
>> >> the
>> >> > > > > > > checkpointed call had window id 99.
>> >> > > > > > >
>> >> > > > > > > Note: This is observed in the local mode of Apex.
>> >> > > > > > >
>> >> > > > > > > Thanks
>> >> > > > > > > -Bhupesh
>> >> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
>> >><tim@datatorrent.com
>> >> >
>> >> > > > wrote:
>> >> > > > > > >
>> >> > > > > > >> Hi Bhupesh,
>> >> > > > > > >>
>> >> > > > > > >> The sequencing of checkpoint called in relation to
>> >>beginWindow
>> >> > and
>> >> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>> >> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
>> >> WINDOW_COUNTs
>> >> > > are
>> >> > > > > not
>> >> > > > > > >> configured to be the same then it is possible that
>> >> checkpointed
>> >> > is
>> >> > > > > > called
>> >> > > > > > >> within an application window. So the sequence of events
>> >>would
>> >> be
>> >> > > > this:
>> >> > > > > > >>
>> >> > > > > > >> beginWindow -> checkpointed -> endWindow
>> >> > > > > > >>
>> >> > > > > > >> If the APPLICATION_WINDOW_COUNT and
>> CHECKPOINT_WINDOW_COUNT
>> >> are
>> >> > > the
>> >> > > > > same
>> >> > > > > > >> (default). Then the sequence of calls would be this:
>> >> > > > > > >>
>> >> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed
>> (100)
>> >> ->
>> >> > > > > > beginWindow
>> >> > > > > > >> (101)
>> >> > > > > > >>
>> >> > > > > > >> The numbers in the sequence represent possible streaming
>> >> window
>> >> > > Ids
>> >> > > > > that
>> >> > > > > > >> each call would be associated with.
>> >> > > > > > >>
>> >> > > > > > >> The StateMachine which calls these callbacks for non-input
>> >> > > operators
>> >> > > > > is
>> >> > > > > > in
>> >> > > > > > >> GenericNode.java.
>> >> > > > > > >>
>> >> > > > > > >> Tim
>> >> > > > > > >>
>> >> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
>> >> > > > > > bhupesh@datatorrent.com>
>> >> > > > > > >> wrote:
>> >> > > > > > >>
>> >> > > > > > >> > Hi Chetan / Community,
>> >> > > > > > >> >
>> >> > > > > > >> > Can someone please elaborate on why the window id
>> >>supplied
>> >> to
>> >> > > > > > >> > CheckpointListener and the Operator would differ.
>> >> > > > > > >> > I tried looking at the window ids of checkpointed() and
>> >>the
>> >> > > > > > beginWindow()
>> >> > > > > > >> > calls and they differ by 1. Don't know why this should
>> be
>> >> the
>> >> > > > case.
>> >> > > > > > >> >
>> >> > > > > > >> > Thanks.
>> >> > > > > > >> > -Bhupesh
>> >> > > > > > >> >
>> >> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
>> >> > > > > > chetan@datatorrent.com>
>> >> > > > > > >> > wrote:
>> >> > > > > > >> >
>> >> > > > > > >> > > Short answer is yes.
>> >> > > > > > >> > >
>> >> > > > > > >> > > All the control tuples are scheduled to be delivered
>> >> outside
>> >> > > of
>> >> > > > > the
>> >> > > > > > >> > window.
>> >> > > > > > >> > > As checkpointed callback is triggered because of
>> >> CHECKPOINT
>> >> > > > > control
>> >> > > > > > >> > tuple,
>> >> > > > > > >> > > it will happen after endWindow and before the next
>> >> > > beginWindow.
>> >> > > > > > >> > >
>> >> > > > > > >> > > The windowId supplied to CheckpointListener and the
>> one
>> >> > > provided
>> >> > > > > to
>> >> > > > > > >> > > Operator need not match even though the sequence is
>> >> defined.
>> >> > > So
>> >> > > > I
>> >> > > > > am
>> >> > > > > > >> > > curious how you intend to use this knowledge.
>> >> > > > > > >> > >
>> >> > > > > > >> > > --
>> >> > > > > > >> > > Chetan
>> >> > > > > > >> > >
>> >> > > > > > >> > >
>> >> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
>> >> > > > > > thomas@datatorrent.com>
>> >> > > > > > >> > > wrote:
>> >> > > > > > >> > >
>> >> > > > > > >> > > > It has not changed the operator execution model.
>> >>State
>> >> > > > > > serialization
>> >> > > > > > >> is
>> >> > > > > > >> > > > still synchronous, write to HDFS is taken out of the
>> >> > > operator
>> >> > > > > > thread.
>> >> > > > > > >> > > >
>> >> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
>> >> > > > > amol@datatorrent.com
>> >> > > > > > >
>> >> > > > > > >> > > wrote:
>> >> > > > > > >> > > >
>> >> > > > > > >> > > > >
>> >> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
>> >>changed
>> >> > > this?
>> >> > > > > > >> > > > >
>> >> > > > > > >> > > > > Amol
>> >> > > > > > >> > > > >
>> >> > > > > > >> > > > > Sent from my iPhone
>> >> > > > > > >> > > > >
>> >> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>> >> > > > > > >> > > bhupesh@datatorrent.com>
>> >> > > > > > >> > > > > wrote:
>> >> > > > > > >> > > > > >
>> >> > > > > > >> > > > > > Hi All,
>> >> > > > > > >> > > > > >
>> >> > > > > > >> > > > > > Is it safe to assume that the checkpointed() and
>> >>the
>> >> > > > > > >> beginWindow()
>> >> > > > > > >> > > > calls
>> >> > > > > > >> > > > > > are sequenced?
>> >> > > > > > >> > > > > > In other words, are these calls part of the same
>> >> > thread
>> >> > > > and
>> >> > > > > > may
>> >> > > > > > >> not
>> >> > > > > > >> > > run
>> >> > > > > > >> > > > > in
>> >> > > > > > >> > > > > > parallel?
>> >> > > > > > >> > > > > >
>> >> > > > > > >> > > > > > Thanks.
>> >> > > > > > >> > > > > >
>> >> > > > > > >> > > > > > --
>> >> > > > > > >> > > > > > -Bhupesh
>> >> > > > > > >> > > > >
>> >> > > > > > >> > > >
>> >> > > > > > >> > >
>> >> > > > > > >> >
>> >> > > > > > >>
>> >> > > > > >
>> >> > > > >
>> >> > > >
>> >> > >
>> >> >
>> >>
>>
>>
>
Re: Sequencing of operator calls
Posted by Chandni Singh <ch...@datatorrent.com>.
Chetan,
I am pasting the code here and highlighting the lines. The last statement
is NOT a 'return' if Processing_Mode == EXACTLY_ONCE
void checkpoint(long windowId)
{
if (!context.stateless) {
StorageAgent ba = context.getValue(OperatorContext.STORAGE_AGENT);
if (ba != null) {
try {
checkpointStats = new Stats.CheckpointStats();
checkpointStats.checkpointStartTime = System.currentTimeMillis();
ba.save(operator, id, windowId);
if (ba instanceof AsyncFSStorageAgent) {
AsyncFSStorageAgent asyncFSStorageAgent = (AsyncFSStorageAgent) ba;
if (!asyncFSStorageAgent.isSyncCheckpoint()) {
if(PROCESSING_MODE != ProcessingMode.EXACTLY_ONCE) {
CheckpointHandler checkpointHandler = new CheckpointHandler();
checkpointHandler.agent = asyncFSStorageAgent;
checkpointHandler.operatorId = id;
checkpointHandler.windowId = windowId;
checkpointHandler.stats = checkpointStats;
FutureTask<Stats.CheckpointStats> futureTask = new
FutureTask<Stats.CheckpointStats>(checkpointHandler);
taskQueue.add(new
Pair<FutureTask<Stats.CheckpointStats>, Long>(futureTask, windowId));
executorService.submit(futureTask);
checkpoint = null;
checkpointStats = null;
return;
}else{
* asyncFSStorageAgent.copyToHDFS(id, windowId);* }
}
}
checkpointStats.checkpointTime = System.currentTimeMillis() -
checkpointStats.checkpointStartTime;
}
catch (IOException ie) {
try {
logger.warn("Rolling back checkpoint {} for Operator {} due
to the exception {}",
Codec.getStringWindowId(windowId), operator, ie);
ba.delete(id, windowId);
}
catch (IOException ex) {
logger.warn("Error while rolling back checkpoint", ex);
}
throw new RuntimeException(ie);
}
}
}
checkpoint = new Checkpoint(windowId, applicationWindowCount,
checkpointWindowCount);
if (operator instanceof Operator.CheckpointListener) {
* ((Operator.CheckpointListener) operator).checkpointed(windowId);* }
}
Chandni
On Tue, Nov 10, 2015 at 3:05 PM, Chetan Narsude (cnarsude) <
cnarsude@cisco.com> wrote:
> There are a lot of things which are different when it comes to async
> checkpointing. I was evaluating it in the morning and expect that either I
> am able to explain or open jira issues. With my partial observation is
> that with Async checkpointing, checkpointed is not issued (chandni, the
> last statement in the if block is ³return²). I am digging into it but feel
> free to chime in if someone else is able to find that.
>
> Also I realized that my morning email applies as it is to committed but
> checkpointed has deviated a little bit from that. Will post the revised
> response soon.
>
> ‹
> Chetan
>
>
>
>
> On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
>
> >Chetan,
> >
> >Looking at the checkpoint(windowId) in Node.java, I don't think the steps
> >you mentioned are followed.
> >
> >*if (using AsyncFSStorageAgent) {*
> >* asyncFSStorageAgent.copyToHdfs(...)*
> >*}*
> >*operator.checkpointed(windowId);*
> >
> >This means even copyToHdfs fails the operator is notified that the window
> >is check-pointed.
> >
> >Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent for
> >a window since the operator is notified that the window is checkpointed?
> >
> >Chandni
> >
> >On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
> >wrote:
> >
> >> Will do
> >>
> >> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
> >><pr...@datatorrent.com>
> >> wrote:
> >>
> >> > Is there a unit test covering it? Otherwise can you write one to test
> >>the
> >> > hypothesis.
> >> >
> >> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas <tim@datatorrent.com
> >
> >> > wrote:
> >> >
> >> > > That is what it is looking like to me. The task is submitted
> >> > > GenericNode#checkpoint line 504, then at the end of the
> >> > > GenericNode#checkpoint line 531 checkpointed is called. I am likely
> >> > missing
> >> > > something, just would like to know what :)
> >> > >
> >> > > Tim
> >> > >
> >> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> >> > pramod@datatorrent.com>
> >> > > wrote:
> >> > >
> >> > > > Tim,
> >> > > >
> >> > > > Are you suggesting that checkpointed is called before the
> >>checkpoint
> >> is
> >> > > > completely persisted in the storage.
> >> > > >
> >> > > > Thanks
> >> > > >
> >> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> >> tim@datatorrent.com>
> >> > > > wrote:
> >> > > >
> >> > > > > Chetan,
> >> > > > >
> >> > > > > I do not see the process of reporting the checkpoint to stram,
> >> > > receiving
> >> > > > > the ack, and then calling checkpointed. The logic I'm seeing in
> >> > > > GenericNode
> >> > > > > line 484 is that the checkpoint method is called, it spawns
> >>another
> >> > > > thread
> >> > > > > that writes to hdfs, and then checkpointed is immediately called
> >> > > > > afterwards. I am missing something, can you give me some
> >>pointers
> >> so
> >> > > > that I
> >> > > > > can better understand the flow?
> >> > > > >
> >> > > > > Tim
> >> > > > >
> >> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> >> > > ram@datatorrent.com
> >> > > > >
> >> > > > > wrote:
> >> > > > >
> >> > > > > > Chetan's answer provides a good explanation as well as
> >>clarifying
> >> > > that
> >> > > > > > the difference can be more than 1.
> >> > > > > >
> >> > > > > > Since checkpointing (i.e. "commit notification" as Thomas
> >>refers
> >> to
> >> > > > > > it) is asynchronous, I'm curious
> >> > > > > > about whether the window ids in the checkpointed call are
> >> > guaranteed
> >> > > > > > to be sequential or if they could
> >> > > > > > be out of order, i.e. can the checkpointed call see window id
> >>101
> >> > > > > > before it sees 100 ?
> >> > > > > >
> >> > > > > > Ram
> >> > > > > >
> >> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> >> > > > > > <bh...@datatorrent.com> wrote:
> >> > > > > > > Hi Tim,
> >> > > > > > > Thanks for the detailed explanation.
> >> > > > > > > I understand that the sequence would be
> >> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> >> > > beginWindow
> >> > > > > > > (x+1)
> >> > > > > > >
> >> > > > > > > However when I try to print out the window ids in
> >>beginWindow,
> >> > > > > endWindow
> >> > > > > > > and checkpointed calls, I see x and x-1 respectively.
> >> > > > > > > I.e. If the window just before checkpoint is 100, I see that
> >> the
> >> > > > > > > checkpointed call had window id 99.
> >> > > > > > >
> >> > > > > > > Note: This is observed in the local mode of Apex.
> >> > > > > > >
> >> > > > > > > Thanks
> >> > > > > > > -Bhupesh
> >> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
> >><tim@datatorrent.com
> >> >
> >> > > > wrote:
> >> > > > > > >
> >> > > > > > >> Hi Bhupesh,
> >> > > > > > >>
> >> > > > > > >> The sequencing of checkpoint called in relation to
> >>beginWindow
> >> > and
> >> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> >> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
> >> WINDOW_COUNTs
> >> > > are
> >> > > > > not
> >> > > > > > >> configured to be the same then it is possible that
> >> checkpointed
> >> > is
> >> > > > > > called
> >> > > > > > >> within an application window. So the sequence of events
> >>would
> >> be
> >> > > > this:
> >> > > > > > >>
> >> > > > > > >> beginWindow -> checkpointed -> endWindow
> >> > > > > > >>
> >> > > > > > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT
> >> are
> >> > > the
> >> > > > > same
> >> > > > > > >> (default). Then the sequence of calls would be this:
> >> > > > > > >>
> >> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed (100)
> >> ->
> >> > > > > > beginWindow
> >> > > > > > >> (101)
> >> > > > > > >>
> >> > > > > > >> The numbers in the sequence represent possible streaming
> >> window
> >> > > Ids
> >> > > > > that
> >> > > > > > >> each call would be associated with.
> >> > > > > > >>
> >> > > > > > >> The StateMachine which calls these callbacks for non-input
> >> > > operators
> >> > > > > is
> >> > > > > > in
> >> > > > > > >> GenericNode.java.
> >> > > > > > >>
> >> > > > > > >> Tim
> >> > > > > > >>
> >> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> >> > > > > > bhupesh@datatorrent.com>
> >> > > > > > >> wrote:
> >> > > > > > >>
> >> > > > > > >> > Hi Chetan / Community,
> >> > > > > > >> >
> >> > > > > > >> > Can someone please elaborate on why the window id
> >>supplied
> >> to
> >> > > > > > >> > CheckpointListener and the Operator would differ.
> >> > > > > > >> > I tried looking at the window ids of checkpointed() and
> >>the
> >> > > > > > beginWindow()
> >> > > > > > >> > calls and they differ by 1. Don't know why this should be
> >> the
> >> > > > case.
> >> > > > > > >> >
> >> > > > > > >> > Thanks.
> >> > > > > > >> > -Bhupesh
> >> > > > > > >> >
> >> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> >> > > > > > chetan@datatorrent.com>
> >> > > > > > >> > wrote:
> >> > > > > > >> >
> >> > > > > > >> > > Short answer is yes.
> >> > > > > > >> > >
> >> > > > > > >> > > All the control tuples are scheduled to be delivered
> >> outside
> >> > > of
> >> > > > > the
> >> > > > > > >> > window.
> >> > > > > > >> > > As checkpointed callback is triggered because of
> >> CHECKPOINT
> >> > > > > control
> >> > > > > > >> > tuple,
> >> > > > > > >> > > it will happen after endWindow and before the next
> >> > > beginWindow.
> >> > > > > > >> > >
> >> > > > > > >> > > The windowId supplied to CheckpointListener and the one
> >> > > provided
> >> > > > > to
> >> > > > > > >> > > Operator need not match even though the sequence is
> >> defined.
> >> > > So
> >> > > > I
> >> > > > > am
> >> > > > > > >> > > curious how you intend to use this knowledge.
> >> > > > > > >> > >
> >> > > > > > >> > > --
> >> > > > > > >> > > Chetan
> >> > > > > > >> > >
> >> > > > > > >> > >
> >> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> >> > > > > > thomas@datatorrent.com>
> >> > > > > > >> > > wrote:
> >> > > > > > >> > >
> >> > > > > > >> > > > It has not changed the operator execution model.
> >>State
> >> > > > > > serialization
> >> > > > > > >> is
> >> > > > > > >> > > > still synchronous, write to HDFS is taken out of the
> >> > > operator
> >> > > > > > thread.
> >> > > > > > >> > > >
> >> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> >> > > > > amol@datatorrent.com
> >> > > > > > >
> >> > > > > > >> > > wrote:
> >> > > > > > >> > > >
> >> > > > > > >> > > > >
> >> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
> >>changed
> >> > > this?
> >> > > > > > >> > > > >
> >> > > > > > >> > > > > Amol
> >> > > > > > >> > > > >
> >> > > > > > >> > > > > Sent from my iPhone
> >> > > > > > >> > > > >
> >> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> >> > > > > > >> > > bhupesh@datatorrent.com>
> >> > > > > > >> > > > > wrote:
> >> > > > > > >> > > > > >
> >> > > > > > >> > > > > > Hi All,
> >> > > > > > >> > > > > >
> >> > > > > > >> > > > > > Is it safe to assume that the checkpointed() and
> >>the
> >> > > > > > >> beginWindow()
> >> > > > > > >> > > > calls
> >> > > > > > >> > > > > > are sequenced?
> >> > > > > > >> > > > > > In other words, are these calls part of the same
> >> > thread
> >> > > > and
> >> > > > > > may
> >> > > > > > >> not
> >> > > > > > >> > > run
> >> > > > > > >> > > > > in
> >> > > > > > >> > > > > > parallel?
> >> > > > > > >> > > > > >
> >> > > > > > >> > > > > > Thanks.
> >> > > > > > >> > > > > >
> >> > > > > > >> > > > > > --
> >> > > > > > >> > > > > > -Bhupesh
> >> > > > > > >> > > > >
> >> > > > > > >> > > >
> >> > > > > > >> > >
> >> > > > > > >> >
> >> > > > > > >>
> >> > > > > >
> >> > > > >
> >> > > >
> >> > >
> >> >
> >>
>
>
Re: Sequencing of operator calls
Posted by "Chetan Narsude (cnarsude)" <cn...@cisco.com>.
There are a lot of things which are different when it comes to async
checkpointing. I was evaluating it in the morning and expect that either I
am able to explain or open jira issues. With my partial observation is
that with Async checkpointing, checkpointed is not issued (chandni, the
last statement in the if block is ³return²). I am digging into it but feel
free to chime in if someone else is able to find that.
Also I realized that my morning email applies as it is to committed but
checkpointed has deviated a little bit from that. Will post the revised
response soon.
‹
Chetan
On 11/10/15, 2:04 PM, "Chandni Singh" <ch...@datatorrent.com> wrote:
>Chetan,
>
>Looking at the checkpoint(windowId) in Node.java, I don't think the steps
>you mentioned are followed.
>
>*if (using AsyncFSStorageAgent) {*
>* asyncFSStorageAgent.copyToHdfs(...)*
>*}*
>*operator.checkpointed(windowId);*
>
>This means even copyToHdfs fails the operator is notified that the window
>is check-pointed.
>
>Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent for
>a window since the operator is notified that the window is checkpointed?
>
>Chandni
>
>On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
>wrote:
>
>> Will do
>>
>> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni
>><pr...@datatorrent.com>
>> wrote:
>>
>> > Is there a unit test covering it? Otherwise can you write one to test
>>the
>> > hypothesis.
>> >
>> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas <ti...@datatorrent.com>
>> > wrote:
>> >
>> > > That is what it is looking like to me. The task is submitted
>> > > GenericNode#checkpoint line 504, then at the end of the
>> > > GenericNode#checkpoint line 531 checkpointed is called. I am likely
>> > missing
>> > > something, just would like to know what :)
>> > >
>> > > Tim
>> > >
>> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
>> > pramod@datatorrent.com>
>> > > wrote:
>> > >
>> > > > Tim,
>> > > >
>> > > > Are you suggesting that checkpointed is called before the
>>checkpoint
>> is
>> > > > completely persisted in the storage.
>> > > >
>> > > > Thanks
>> > > >
>> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
>> tim@datatorrent.com>
>> > > > wrote:
>> > > >
>> > > > > Chetan,
>> > > > >
>> > > > > I do not see the process of reporting the checkpoint to stram,
>> > > receiving
>> > > > > the ack, and then calling checkpointed. The logic I'm seeing in
>> > > > GenericNode
>> > > > > line 484 is that the checkpoint method is called, it spawns
>>another
>> > > > thread
>> > > > > that writes to hdfs, and then checkpointed is immediately called
>> > > > > afterwards. I am missing something, can you give me some
>>pointers
>> so
>> > > > that I
>> > > > > can better understand the flow?
>> > > > >
>> > > > > Tim
>> > > > >
>> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
>> > > ram@datatorrent.com
>> > > > >
>> > > > > wrote:
>> > > > >
>> > > > > > Chetan's answer provides a good explanation as well as
>>clarifying
>> > > that
>> > > > > > the difference can be more than 1.
>> > > > > >
>> > > > > > Since checkpointing (i.e. "commit notification" as Thomas
>>refers
>> to
>> > > > > > it) is asynchronous, I'm curious
>> > > > > > about whether the window ids in the checkpointed call are
>> > guaranteed
>> > > > > > to be sequential or if they could
>> > > > > > be out of order, i.e. can the checkpointed call see window id
>>101
>> > > > > > before it sees 100 ?
>> > > > > >
>> > > > > > Ram
>> > > > > >
>> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
>> > > > > > <bh...@datatorrent.com> wrote:
>> > > > > > > Hi Tim,
>> > > > > > > Thanks for the detailed explanation.
>> > > > > > > I understand that the sequence would be
>> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
>> > > beginWindow
>> > > > > > > (x+1)
>> > > > > > >
>> > > > > > > However when I try to print out the window ids in
>>beginWindow,
>> > > > > endWindow
>> > > > > > > and checkpointed calls, I see x and x-1 respectively.
>> > > > > > > I.e. If the window just before checkpoint is 100, I see that
>> the
>> > > > > > > checkpointed call had window id 99.
>> > > > > > >
>> > > > > > > Note: This is observed in the local mode of Apex.
>> > > > > > >
>> > > > > > > Thanks
>> > > > > > > -Bhupesh
>> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas"
>><tim@datatorrent.com
>> >
>> > > > wrote:
>> > > > > > >
>> > > > > > >> Hi Bhupesh,
>> > > > > > >>
>> > > > > > >> The sequencing of checkpoint called in relation to
>>beginWindow
>> > and
>> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
>> WINDOW_COUNTs
>> > > are
>> > > > > not
>> > > > > > >> configured to be the same then it is possible that
>> checkpointed
>> > is
>> > > > > > called
>> > > > > > >> within an application window. So the sequence of events
>>would
>> be
>> > > > this:
>> > > > > > >>
>> > > > > > >> beginWindow -> checkpointed -> endWindow
>> > > > > > >>
>> > > > > > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT
>> are
>> > > the
>> > > > > same
>> > > > > > >> (default). Then the sequence of calls would be this:
>> > > > > > >>
>> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed (100)
>> ->
>> > > > > > beginWindow
>> > > > > > >> (101)
>> > > > > > >>
>> > > > > > >> The numbers in the sequence represent possible streaming
>> window
>> > > Ids
>> > > > > that
>> > > > > > >> each call would be associated with.
>> > > > > > >>
>> > > > > > >> The StateMachine which calls these callbacks for non-input
>> > > operators
>> > > > > is
>> > > > > > in
>> > > > > > >> GenericNode.java.
>> > > > > > >>
>> > > > > > >> Tim
>> > > > > > >>
>> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
>> > > > > > bhupesh@datatorrent.com>
>> > > > > > >> wrote:
>> > > > > > >>
>> > > > > > >> > Hi Chetan / Community,
>> > > > > > >> >
>> > > > > > >> > Can someone please elaborate on why the window id
>>supplied
>> to
>> > > > > > >> > CheckpointListener and the Operator would differ.
>> > > > > > >> > I tried looking at the window ids of checkpointed() and
>>the
>> > > > > > beginWindow()
>> > > > > > >> > calls and they differ by 1. Don't know why this should be
>> the
>> > > > case.
>> > > > > > >> >
>> > > > > > >> > Thanks.
>> > > > > > >> > -Bhupesh
>> > > > > > >> >
>> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
>> > > > > > chetan@datatorrent.com>
>> > > > > > >> > wrote:
>> > > > > > >> >
>> > > > > > >> > > Short answer is yes.
>> > > > > > >> > >
>> > > > > > >> > > All the control tuples are scheduled to be delivered
>> outside
>> > > of
>> > > > > the
>> > > > > > >> > window.
>> > > > > > >> > > As checkpointed callback is triggered because of
>> CHECKPOINT
>> > > > > control
>> > > > > > >> > tuple,
>> > > > > > >> > > it will happen after endWindow and before the next
>> > > beginWindow.
>> > > > > > >> > >
>> > > > > > >> > > The windowId supplied to CheckpointListener and the one
>> > > provided
>> > > > > to
>> > > > > > >> > > Operator need not match even though the sequence is
>> defined.
>> > > So
>> > > > I
>> > > > > am
>> > > > > > >> > > curious how you intend to use this knowledge.
>> > > > > > >> > >
>> > > > > > >> > > --
>> > > > > > >> > > Chetan
>> > > > > > >> > >
>> > > > > > >> > >
>> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
>> > > > > > thomas@datatorrent.com>
>> > > > > > >> > > wrote:
>> > > > > > >> > >
>> > > > > > >> > > > It has not changed the operator execution model.
>>State
>> > > > > > serialization
>> > > > > > >> is
>> > > > > > >> > > > still synchronous, write to HDFS is taken out of the
>> > > operator
>> > > > > > thread.
>> > > > > > >> > > >
>> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
>> > > > > amol@datatorrent.com
>> > > > > > >
>> > > > > > >> > > wrote:
>> > > > > > >> > > >
>> > > > > > >> > > > >
>> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
>>changed
>> > > this?
>> > > > > > >> > > > >
>> > > > > > >> > > > > Amol
>> > > > > > >> > > > >
>> > > > > > >> > > > > Sent from my iPhone
>> > > > > > >> > > > >
>> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>> > > > > > >> > > bhupesh@datatorrent.com>
>> > > > > > >> > > > > wrote:
>> > > > > > >> > > > > >
>> > > > > > >> > > > > > Hi All,
>> > > > > > >> > > > > >
>> > > > > > >> > > > > > Is it safe to assume that the checkpointed() and
>>the
>> > > > > > >> beginWindow()
>> > > > > > >> > > > calls
>> > > > > > >> > > > > > are sequenced?
>> > > > > > >> > > > > > In other words, are these calls part of the same
>> > thread
>> > > > and
>> > > > > > may
>> > > > > > >> not
>> > > > > > >> > > run
>> > > > > > >> > > > > in
>> > > > > > >> > > > > > parallel?
>> > > > > > >> > > > > >
>> > > > > > >> > > > > > Thanks.
>> > > > > > >> > > > > >
>> > > > > > >> > > > > > --
>> > > > > > >> > > > > > -Bhupesh
>> > > > > > >> > > > >
>> > > > > > >> > > >
>> > > > > > >> > >
>> > > > > > >> >
>> > > > > > >>
>> > > > > >
>> > > > >
>> > > >
>> > >
>> >
>>
Re: Sequencing of operator calls
Posted by Pramod Immaneni <pr...@datatorrent.com>.
This doesn't look right. If async copy is happening in a separate thread
then the checkpointed should be triggered (not called) from there on
success.
On Tue, Nov 10, 2015 at 2:04 PM, Chandni Singh <ch...@datatorrent.com>
wrote:
> Chetan,
>
> Looking at the checkpoint(windowId) in Node.java, I don't think the steps
> you mentioned are followed.
>
> *if (using AsyncFSStorageAgent) {*
> * asyncFSStorageAgent.copyToHdfs(...)*
> *}*
> *operator.checkpointed(windowId);*
>
> This means even copyToHdfs fails the operator is notified that the window
> is check-pointed.
>
> Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent for
> a window since the operator is notified that the window is checkpointed?
>
> Chandni
>
> On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
> wrote:
>
> > Will do
> >
> > On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni <
> pramod@datatorrent.com>
> > wrote:
> >
> > > Is there a unit test covering it? Otherwise can you write one to test
> the
> > > hypothesis.
> > >
> > > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas <ti...@datatorrent.com>
> > > wrote:
> > >
> > > > That is what it is looking like to me. The task is submitted
> > > > GenericNode#checkpoint line 504, then at the end of the
> > > > GenericNode#checkpoint line 531 checkpointed is called. I am likely
> > > missing
> > > > something, just would like to know what :)
> > > >
> > > > Tim
> > > >
> > > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> > > pramod@datatorrent.com>
> > > > wrote:
> > > >
> > > > > Tim,
> > > > >
> > > > > Are you suggesting that checkpointed is called before the
> checkpoint
> > is
> > > > > completely persisted in the storage.
> > > > >
> > > > > Thanks
> > > > >
> > > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> > tim@datatorrent.com>
> > > > > wrote:
> > > > >
> > > > > > Chetan,
> > > > > >
> > > > > > I do not see the process of reporting the checkpoint to stram,
> > > > receiving
> > > > > > the ack, and then calling checkpointed. The logic I'm seeing in
> > > > > GenericNode
> > > > > > line 484 is that the checkpoint method is called, it spawns
> another
> > > > > thread
> > > > > > that writes to hdfs, and then checkpointed is immediately called
> > > > > > afterwards. I am missing something, can you give me some pointers
> > so
> > > > > that I
> > > > > > can better understand the flow?
> > > > > >
> > > > > > Tim
> > > > > >
> > > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> > > > ram@datatorrent.com
> > > > > >
> > > > > > wrote:
> > > > > >
> > > > > > > Chetan's answer provides a good explanation as well as
> clarifying
> > > > that
> > > > > > > the difference can be more than 1.
> > > > > > >
> > > > > > > Since checkpointing (i.e. "commit notification" as Thomas
> refers
> > to
> > > > > > > it) is asynchronous, I'm curious
> > > > > > > about whether the window ids in the checkpointed call are
> > > guaranteed
> > > > > > > to be sequential or if they could
> > > > > > > be out of order, i.e. can the checkpointed call see window id
> 101
> > > > > > > before it sees 100 ?
> > > > > > >
> > > > > > > Ram
> > > > > > >
> > > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > > > > > > <bh...@datatorrent.com> wrote:
> > > > > > > > Hi Tim,
> > > > > > > > Thanks for the detailed explanation.
> > > > > > > > I understand that the sequence would be
> > > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> > > > beginWindow
> > > > > > > > (x+1)
> > > > > > > >
> > > > > > > > However when I try to print out the window ids in
> beginWindow,
> > > > > > endWindow
> > > > > > > > and checkpointed calls, I see x and x-1 respectively.
> > > > > > > > I.e. If the window just before checkpoint is 100, I see that
> > the
> > > > > > > > checkpointed call had window id 99.
> > > > > > > >
> > > > > > > > Note: This is observed in the local mode of Apex.
> > > > > > > >
> > > > > > > > Thanks
> > > > > > > > -Bhupesh
> > > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas" <
> tim@datatorrent.com
> > >
> > > > > wrote:
> > > > > > > >
> > > > > > > >> Hi Bhupesh,
> > > > > > > >>
> > > > > > > >> The sequencing of checkpoint called in relation to
> beginWindow
> > > and
> > > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> > > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
> > WINDOW_COUNTs
> > > > are
> > > > > > not
> > > > > > > >> configured to be the same then it is possible that
> > checkpointed
> > > is
> > > > > > > called
> > > > > > > >> within an application window. So the sequence of events
> would
> > be
> > > > > this:
> > > > > > > >>
> > > > > > > >> beginWindow -> checkpointed -> endWindow
> > > > > > > >>
> > > > > > > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT
> > are
> > > > the
> > > > > > same
> > > > > > > >> (default). Then the sequence of calls would be this:
> > > > > > > >>
> > > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed (100)
> > ->
> > > > > > > beginWindow
> > > > > > > >> (101)
> > > > > > > >>
> > > > > > > >> The numbers in the sequence represent possible streaming
> > window
> > > > Ids
> > > > > > that
> > > > > > > >> each call would be associated with.
> > > > > > > >>
> > > > > > > >> The StateMachine which calls these callbacks for non-input
> > > > operators
> > > > > > is
> > > > > > > in
> > > > > > > >> GenericNode.java.
> > > > > > > >>
> > > > > > > >> Tim
> > > > > > > >>
> > > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > > > > > > bhupesh@datatorrent.com>
> > > > > > > >> wrote:
> > > > > > > >>
> > > > > > > >> > Hi Chetan / Community,
> > > > > > > >> >
> > > > > > > >> > Can someone please elaborate on why the window id supplied
> > to
> > > > > > > >> > CheckpointListener and the Operator would differ.
> > > > > > > >> > I tried looking at the window ids of checkpointed() and
> the
> > > > > > > beginWindow()
> > > > > > > >> > calls and they differ by 1. Don't know why this should be
> > the
> > > > > case.
> > > > > > > >> >
> > > > > > > >> > Thanks.
> > > > > > > >> > -Bhupesh
> > > > > > > >> >
> > > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > > > > > > chetan@datatorrent.com>
> > > > > > > >> > wrote:
> > > > > > > >> >
> > > > > > > >> > > Short answer is yes.
> > > > > > > >> > >
> > > > > > > >> > > All the control tuples are scheduled to be delivered
> > outside
> > > > of
> > > > > > the
> > > > > > > >> > window.
> > > > > > > >> > > As checkpointed callback is triggered because of
> > CHECKPOINT
> > > > > > control
> > > > > > > >> > tuple,
> > > > > > > >> > > it will happen after endWindow and before the next
> > > > beginWindow.
> > > > > > > >> > >
> > > > > > > >> > > The windowId supplied to CheckpointListener and the one
> > > > provided
> > > > > > to
> > > > > > > >> > > Operator need not match even though the sequence is
> > defined.
> > > > So
> > > > > I
> > > > > > am
> > > > > > > >> > > curious how you intend to use this knowledge.
> > > > > > > >> > >
> > > > > > > >> > > --
> > > > > > > >> > > Chetan
> > > > > > > >> > >
> > > > > > > >> > >
> > > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > > > > > > thomas@datatorrent.com>
> > > > > > > >> > > wrote:
> > > > > > > >> > >
> > > > > > > >> > > > It has not changed the operator execution model. State
> > > > > > > serialization
> > > > > > > >> is
> > > > > > > >> > > > still synchronous, write to HDFS is taken out of the
> > > > operator
> > > > > > > thread.
> > > > > > > >> > > >
> > > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > > > > > amol@datatorrent.com
> > > > > > > >
> > > > > > > >> > > wrote:
> > > > > > > >> > > >
> > > > > > > >> > > > >
> > > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing
> changed
> > > > this?
> > > > > > > >> > > > >
> > > > > > > >> > > > > Amol
> > > > > > > >> > > > >
> > > > > > > >> > > > > Sent from my iPhone
> > > > > > > >> > > > >
> > > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > > > > > >> > > bhupesh@datatorrent.com>
> > > > > > > >> > > > > wrote:
> > > > > > > >> > > > > >
> > > > > > > >> > > > > > Hi All,
> > > > > > > >> > > > > >
> > > > > > > >> > > > > > Is it safe to assume that the checkpointed() and
> the
> > > > > > > >> beginWindow()
> > > > > > > >> > > > calls
> > > > > > > >> > > > > > are sequenced?
> > > > > > > >> > > > > > In other words, are these calls part of the same
> > > thread
> > > > > and
> > > > > > > may
> > > > > > > >> not
> > > > > > > >> > > run
> > > > > > > >> > > > > in
> > > > > > > >> > > > > > parallel?
> > > > > > > >> > > > > >
> > > > > > > >> > > > > > Thanks.
> > > > > > > >> > > > > >
> > > > > > > >> > > > > > --
> > > > > > > >> > > > > > -Bhupesh
> > > > > > > >> > > > >
> > > > > > > >> > > >
> > > > > > > >> > >
> > > > > > > >> >
> > > > > > > >>
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Chandni Singh <ch...@datatorrent.com>.
Chetan,
Looking at the checkpoint(windowId) in Node.java, I don't think the steps
you mentioned are followed.
*if (using AsyncFSStorageAgent) {*
* asyncFSStorageAgent.copyToHdfs(...)*
*}*
*operator.checkpointed(windowId);*
This means even copyToHdfs fails the operator is notified that the window
is check-pointed.
Are we saying that copyToHdfs will never fail with AsyncFSStorageAgent for
a window since the operator is notified that the window is checkpointed?
Chandni
On Tue, Nov 10, 2015 at 11:33 AM, Timothy Farkas <ti...@datatorrent.com>
wrote:
> Will do
>
> On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni <pr...@datatorrent.com>
> wrote:
>
> > Is there a unit test covering it? Otherwise can you write one to test the
> > hypothesis.
> >
> > On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas <ti...@datatorrent.com>
> > wrote:
> >
> > > That is what it is looking like to me. The task is submitted
> > > GenericNode#checkpoint line 504, then at the end of the
> > > GenericNode#checkpoint line 531 checkpointed is called. I am likely
> > missing
> > > something, just would like to know what :)
> > >
> > > Tim
> > >
> > > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> > pramod@datatorrent.com>
> > > wrote:
> > >
> > > > Tim,
> > > >
> > > > Are you suggesting that checkpointed is called before the checkpoint
> is
> > > > completely persisted in the storage.
> > > >
> > > > Thanks
> > > >
> > > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <
> tim@datatorrent.com>
> > > > wrote:
> > > >
> > > > > Chetan,
> > > > >
> > > > > I do not see the process of reporting the checkpoint to stram,
> > > receiving
> > > > > the ack, and then calling checkpointed. The logic I'm seeing in
> > > > GenericNode
> > > > > line 484 is that the checkpoint method is called, it spawns another
> > > > thread
> > > > > that writes to hdfs, and then checkpointed is immediately called
> > > > > afterwards. I am missing something, can you give me some pointers
> so
> > > > that I
> > > > > can better understand the flow?
> > > > >
> > > > > Tim
> > > > >
> > > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> > > ram@datatorrent.com
> > > > >
> > > > > wrote:
> > > > >
> > > > > > Chetan's answer provides a good explanation as well as clarifying
> > > that
> > > > > > the difference can be more than 1.
> > > > > >
> > > > > > Since checkpointing (i.e. "commit notification" as Thomas refers
> to
> > > > > > it) is asynchronous, I'm curious
> > > > > > about whether the window ids in the checkpointed call are
> > guaranteed
> > > > > > to be sequential or if they could
> > > > > > be out of order, i.e. can the checkpointed call see window id 101
> > > > > > before it sees 100 ?
> > > > > >
> > > > > > Ram
> > > > > >
> > > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > > > > > <bh...@datatorrent.com> wrote:
> > > > > > > Hi Tim,
> > > > > > > Thanks for the detailed explanation.
> > > > > > > I understand that the sequence would be
> > > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> > > beginWindow
> > > > > > > (x+1)
> > > > > > >
> > > > > > > However when I try to print out the window ids in beginWindow,
> > > > > endWindow
> > > > > > > and checkpointed calls, I see x and x-1 respectively.
> > > > > > > I.e. If the window just before checkpoint is 100, I see that
> the
> > > > > > > checkpointed call had window id 99.
> > > > > > >
> > > > > > > Note: This is observed in the local mode of Apex.
> > > > > > >
> > > > > > > Thanks
> > > > > > > -Bhupesh
> > > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas" <tim@datatorrent.com
> >
> > > > wrote:
> > > > > > >
> > > > > > >> Hi Bhupesh,
> > > > > > >>
> > > > > > >> The sequencing of checkpoint called in relation to beginWindow
> > and
> > > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> > > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two
> WINDOW_COUNTs
> > > are
> > > > > not
> > > > > > >> configured to be the same then it is possible that
> checkpointed
> > is
> > > > > > called
> > > > > > >> within an application window. So the sequence of events would
> be
> > > > this:
> > > > > > >>
> > > > > > >> beginWindow -> checkpointed -> endWindow
> > > > > > >>
> > > > > > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT
> are
> > > the
> > > > > same
> > > > > > >> (default). Then the sequence of calls would be this:
> > > > > > >>
> > > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed (100)
> ->
> > > > > > beginWindow
> > > > > > >> (101)
> > > > > > >>
> > > > > > >> The numbers in the sequence represent possible streaming
> window
> > > Ids
> > > > > that
> > > > > > >> each call would be associated with.
> > > > > > >>
> > > > > > >> The StateMachine which calls these callbacks for non-input
> > > operators
> > > > > is
> > > > > > in
> > > > > > >> GenericNode.java.
> > > > > > >>
> > > > > > >> Tim
> > > > > > >>
> > > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > > > > > bhupesh@datatorrent.com>
> > > > > > >> wrote:
> > > > > > >>
> > > > > > >> > Hi Chetan / Community,
> > > > > > >> >
> > > > > > >> > Can someone please elaborate on why the window id supplied
> to
> > > > > > >> > CheckpointListener and the Operator would differ.
> > > > > > >> > I tried looking at the window ids of checkpointed() and the
> > > > > > beginWindow()
> > > > > > >> > calls and they differ by 1. Don't know why this should be
> the
> > > > case.
> > > > > > >> >
> > > > > > >> > Thanks.
> > > > > > >> > -Bhupesh
> > > > > > >> >
> > > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > > > > > chetan@datatorrent.com>
> > > > > > >> > wrote:
> > > > > > >> >
> > > > > > >> > > Short answer is yes.
> > > > > > >> > >
> > > > > > >> > > All the control tuples are scheduled to be delivered
> outside
> > > of
> > > > > the
> > > > > > >> > window.
> > > > > > >> > > As checkpointed callback is triggered because of
> CHECKPOINT
> > > > > control
> > > > > > >> > tuple,
> > > > > > >> > > it will happen after endWindow and before the next
> > > beginWindow.
> > > > > > >> > >
> > > > > > >> > > The windowId supplied to CheckpointListener and the one
> > > provided
> > > > > to
> > > > > > >> > > Operator need not match even though the sequence is
> defined.
> > > So
> > > > I
> > > > > am
> > > > > > >> > > curious how you intend to use this knowledge.
> > > > > > >> > >
> > > > > > >> > > --
> > > > > > >> > > Chetan
> > > > > > >> > >
> > > > > > >> > >
> > > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > > > > > thomas@datatorrent.com>
> > > > > > >> > > wrote:
> > > > > > >> > >
> > > > > > >> > > > It has not changed the operator execution model. State
> > > > > > serialization
> > > > > > >> is
> > > > > > >> > > > still synchronous, write to HDFS is taken out of the
> > > operator
> > > > > > thread.
> > > > > > >> > > >
> > > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > > > > amol@datatorrent.com
> > > > > > >
> > > > > > >> > > wrote:
> > > > > > >> > > >
> > > > > > >> > > > >
> > > > > > >> > > > > Sent too soon. Has asynchronous checkpointing changed
> > > this?
> > > > > > >> > > > >
> > > > > > >> > > > > Amol
> > > > > > >> > > > >
> > > > > > >> > > > > Sent from my iPhone
> > > > > > >> > > > >
> > > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > > > > >> > > bhupesh@datatorrent.com>
> > > > > > >> > > > > wrote:
> > > > > > >> > > > > >
> > > > > > >> > > > > > Hi All,
> > > > > > >> > > > > >
> > > > > > >> > > > > > Is it safe to assume that the checkpointed() and the
> > > > > > >> beginWindow()
> > > > > > >> > > > calls
> > > > > > >> > > > > > are sequenced?
> > > > > > >> > > > > > In other words, are these calls part of the same
> > thread
> > > > and
> > > > > > may
> > > > > > >> not
> > > > > > >> > > run
> > > > > > >> > > > > in
> > > > > > >> > > > > > parallel?
> > > > > > >> > > > > >
> > > > > > >> > > > > > Thanks.
> > > > > > >> > > > > >
> > > > > > >> > > > > > --
> > > > > > >> > > > > > -Bhupesh
> > > > > > >> > > > >
> > > > > > >> > > >
> > > > > > >> > >
> > > > > > >> >
> > > > > > >>
> > > > > >
> > > > >
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Timothy Farkas <ti...@datatorrent.com>.
Will do
On Tue, Nov 10, 2015 at 11:01 AM, Pramod Immaneni <pr...@datatorrent.com>
wrote:
> Is there a unit test covering it? Otherwise can you write one to test the
> hypothesis.
>
> On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas <ti...@datatorrent.com>
> wrote:
>
> > That is what it is looking like to me. The task is submitted
> > GenericNode#checkpoint line 504, then at the end of the
> > GenericNode#checkpoint line 531 checkpointed is called. I am likely
> missing
> > something, just would like to know what :)
> >
> > Tim
> >
> > On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <
> pramod@datatorrent.com>
> > wrote:
> >
> > > Tim,
> > >
> > > Are you suggesting that checkpointed is called before the checkpoint is
> > > completely persisted in the storage.
> > >
> > > Thanks
> > >
> > > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <ti...@datatorrent.com>
> > > wrote:
> > >
> > > > Chetan,
> > > >
> > > > I do not see the process of reporting the checkpoint to stram,
> > receiving
> > > > the ack, and then calling checkpointed. The logic I'm seeing in
> > > GenericNode
> > > > line 484 is that the checkpoint method is called, it spawns another
> > > thread
> > > > that writes to hdfs, and then checkpointed is immediately called
> > > > afterwards. I am missing something, can you give me some pointers so
> > > that I
> > > > can better understand the flow?
> > > >
> > > > Tim
> > > >
> > > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> > ram@datatorrent.com
> > > >
> > > > wrote:
> > > >
> > > > > Chetan's answer provides a good explanation as well as clarifying
> > that
> > > > > the difference can be more than 1.
> > > > >
> > > > > Since checkpointing (i.e. "commit notification" as Thomas refers to
> > > > > it) is asynchronous, I'm curious
> > > > > about whether the window ids in the checkpointed call are
> guaranteed
> > > > > to be sequential or if they could
> > > > > be out of order, i.e. can the checkpointed call see window id 101
> > > > > before it sees 100 ?
> > > > >
> > > > > Ram
> > > > >
> > > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > > > > <bh...@datatorrent.com> wrote:
> > > > > > Hi Tim,
> > > > > > Thanks for the detailed explanation.
> > > > > > I understand that the sequence would be
> > > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> > beginWindow
> > > > > > (x+1)
> > > > > >
> > > > > > However when I try to print out the window ids in beginWindow,
> > > > endWindow
> > > > > > and checkpointed calls, I see x and x-1 respectively.
> > > > > > I.e. If the window just before checkpoint is 100, I see that the
> > > > > > checkpointed call had window id 99.
> > > > > >
> > > > > > Note: This is observed in the local mode of Apex.
> > > > > >
> > > > > > Thanks
> > > > > > -Bhupesh
> > > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com>
> > > wrote:
> > > > > >
> > > > > >> Hi Bhupesh,
> > > > > >>
> > > > > >> The sequencing of checkpoint called in relation to beginWindow
> and
> > > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> > > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs
> > are
> > > > not
> > > > > >> configured to be the same then it is possible that checkpointed
> is
> > > > > called
> > > > > >> within an application window. So the sequence of events would be
> > > this:
> > > > > >>
> > > > > >> beginWindow -> checkpointed -> endWindow
> > > > > >>
> > > > > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are
> > the
> > > > same
> > > > > >> (default). Then the sequence of calls would be this:
> > > > > >>
> > > > > >> beginWindow (100) -> endWindow (100) -> checkpointed (100) ->
> > > > > beginWindow
> > > > > >> (101)
> > > > > >>
> > > > > >> The numbers in the sequence represent possible streaming window
> > Ids
> > > > that
> > > > > >> each call would be associated with.
> > > > > >>
> > > > > >> The StateMachine which calls these callbacks for non-input
> > operators
> > > > is
> > > > > in
> > > > > >> GenericNode.java.
> > > > > >>
> > > > > >> Tim
> > > > > >>
> > > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > > > > bhupesh@datatorrent.com>
> > > > > >> wrote:
> > > > > >>
> > > > > >> > Hi Chetan / Community,
> > > > > >> >
> > > > > >> > Can someone please elaborate on why the window id supplied to
> > > > > >> > CheckpointListener and the Operator would differ.
> > > > > >> > I tried looking at the window ids of checkpointed() and the
> > > > > beginWindow()
> > > > > >> > calls and they differ by 1. Don't know why this should be the
> > > case.
> > > > > >> >
> > > > > >> > Thanks.
> > > > > >> > -Bhupesh
> > > > > >> >
> > > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > > > > chetan@datatorrent.com>
> > > > > >> > wrote:
> > > > > >> >
> > > > > >> > > Short answer is yes.
> > > > > >> > >
> > > > > >> > > All the control tuples are scheduled to be delivered outside
> > of
> > > > the
> > > > > >> > window.
> > > > > >> > > As checkpointed callback is triggered because of CHECKPOINT
> > > > control
> > > > > >> > tuple,
> > > > > >> > > it will happen after endWindow and before the next
> > beginWindow.
> > > > > >> > >
> > > > > >> > > The windowId supplied to CheckpointListener and the one
> > provided
> > > > to
> > > > > >> > > Operator need not match even though the sequence is defined.
> > So
> > > I
> > > > am
> > > > > >> > > curious how you intend to use this knowledge.
> > > > > >> > >
> > > > > >> > > --
> > > > > >> > > Chetan
> > > > > >> > >
> > > > > >> > >
> > > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > > > > thomas@datatorrent.com>
> > > > > >> > > wrote:
> > > > > >> > >
> > > > > >> > > > It has not changed the operator execution model. State
> > > > > serialization
> > > > > >> is
> > > > > >> > > > still synchronous, write to HDFS is taken out of the
> > operator
> > > > > thread.
> > > > > >> > > >
> > > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > > > amol@datatorrent.com
> > > > > >
> > > > > >> > > wrote:
> > > > > >> > > >
> > > > > >> > > > >
> > > > > >> > > > > Sent too soon. Has asynchronous checkpointing changed
> > this?
> > > > > >> > > > >
> > > > > >> > > > > Amol
> > > > > >> > > > >
> > > > > >> > > > > Sent from my iPhone
> > > > > >> > > > >
> > > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > > > >> > > bhupesh@datatorrent.com>
> > > > > >> > > > > wrote:
> > > > > >> > > > > >
> > > > > >> > > > > > Hi All,
> > > > > >> > > > > >
> > > > > >> > > > > > Is it safe to assume that the checkpointed() and the
> > > > > >> beginWindow()
> > > > > >> > > > calls
> > > > > >> > > > > > are sequenced?
> > > > > >> > > > > > In other words, are these calls part of the same
> thread
> > > and
> > > > > may
> > > > > >> not
> > > > > >> > > run
> > > > > >> > > > > in
> > > > > >> > > > > > parallel?
> > > > > >> > > > > >
> > > > > >> > > > > > Thanks.
> > > > > >> > > > > >
> > > > > >> > > > > > --
> > > > > >> > > > > > -Bhupesh
> > > > > >> > > > >
> > > > > >> > > >
> > > > > >> > >
> > > > > >> >
> > > > > >>
> > > > >
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Pramod Immaneni <pr...@datatorrent.com>.
Is there a unit test covering it? Otherwise can you write one to test the
hypothesis.
On Tue, Nov 10, 2015 at 11:00 AM, Timothy Farkas <ti...@datatorrent.com>
wrote:
> That is what it is looking like to me. The task is submitted
> GenericNode#checkpoint line 504, then at the end of the
> GenericNode#checkpoint line 531 checkpointed is called. I am likely missing
> something, just would like to know what :)
>
> Tim
>
> On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <pr...@datatorrent.com>
> wrote:
>
> > Tim,
> >
> > Are you suggesting that checkpointed is called before the checkpoint is
> > completely persisted in the storage.
> >
> > Thanks
> >
> > On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <ti...@datatorrent.com>
> > wrote:
> >
> > > Chetan,
> > >
> > > I do not see the process of reporting the checkpoint to stram,
> receiving
> > > the ack, and then calling checkpointed. The logic I'm seeing in
> > GenericNode
> > > line 484 is that the checkpoint method is called, it spawns another
> > thread
> > > that writes to hdfs, and then checkpointed is immediately called
> > > afterwards. I am missing something, can you give me some pointers so
> > that I
> > > can better understand the flow?
> > >
> > > Tim
> > >
> > > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <
> ram@datatorrent.com
> > >
> > > wrote:
> > >
> > > > Chetan's answer provides a good explanation as well as clarifying
> that
> > > > the difference can be more than 1.
> > > >
> > > > Since checkpointing (i.e. "commit notification" as Thomas refers to
> > > > it) is asynchronous, I'm curious
> > > > about whether the window ids in the checkpointed call are guaranteed
> > > > to be sequential or if they could
> > > > be out of order, i.e. can the checkpointed call see window id 101
> > > > before it sees 100 ?
> > > >
> > > > Ram
> > > >
> > > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > > > <bh...@datatorrent.com> wrote:
> > > > > Hi Tim,
> > > > > Thanks for the detailed explanation.
> > > > > I understand that the sequence would be
> > > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) ->
> beginWindow
> > > > > (x+1)
> > > > >
> > > > > However when I try to print out the window ids in beginWindow,
> > > endWindow
> > > > > and checkpointed calls, I see x and x-1 respectively.
> > > > > I.e. If the window just before checkpoint is 100, I see that the
> > > > > checkpointed call had window id 99.
> > > > >
> > > > > Note: This is observed in the local mode of Apex.
> > > > >
> > > > > Thanks
> > > > > -Bhupesh
> > > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com>
> > wrote:
> > > > >
> > > > >> Hi Bhupesh,
> > > > >>
> > > > >> The sequencing of checkpoint called in relation to beginWindow and
> > > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> > > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs
> are
> > > not
> > > > >> configured to be the same then it is possible that checkpointed is
> > > > called
> > > > >> within an application window. So the sequence of events would be
> > this:
> > > > >>
> > > > >> beginWindow -> checkpointed -> endWindow
> > > > >>
> > > > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are
> the
> > > same
> > > > >> (default). Then the sequence of calls would be this:
> > > > >>
> > > > >> beginWindow (100) -> endWindow (100) -> checkpointed (100) ->
> > > > beginWindow
> > > > >> (101)
> > > > >>
> > > > >> The numbers in the sequence represent possible streaming window
> Ids
> > > that
> > > > >> each call would be associated with.
> > > > >>
> > > > >> The StateMachine which calls these callbacks for non-input
> operators
> > > is
> > > > in
> > > > >> GenericNode.java.
> > > > >>
> > > > >> Tim
> > > > >>
> > > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > > > bhupesh@datatorrent.com>
> > > > >> wrote:
> > > > >>
> > > > >> > Hi Chetan / Community,
> > > > >> >
> > > > >> > Can someone please elaborate on why the window id supplied to
> > > > >> > CheckpointListener and the Operator would differ.
> > > > >> > I tried looking at the window ids of checkpointed() and the
> > > > beginWindow()
> > > > >> > calls and they differ by 1. Don't know why this should be the
> > case.
> > > > >> >
> > > > >> > Thanks.
> > > > >> > -Bhupesh
> > > > >> >
> > > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > > > chetan@datatorrent.com>
> > > > >> > wrote:
> > > > >> >
> > > > >> > > Short answer is yes.
> > > > >> > >
> > > > >> > > All the control tuples are scheduled to be delivered outside
> of
> > > the
> > > > >> > window.
> > > > >> > > As checkpointed callback is triggered because of CHECKPOINT
> > > control
> > > > >> > tuple,
> > > > >> > > it will happen after endWindow and before the next
> beginWindow.
> > > > >> > >
> > > > >> > > The windowId supplied to CheckpointListener and the one
> provided
> > > to
> > > > >> > > Operator need not match even though the sequence is defined.
> So
> > I
> > > am
> > > > >> > > curious how you intend to use this knowledge.
> > > > >> > >
> > > > >> > > --
> > > > >> > > Chetan
> > > > >> > >
> > > > >> > >
> > > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > > > thomas@datatorrent.com>
> > > > >> > > wrote:
> > > > >> > >
> > > > >> > > > It has not changed the operator execution model. State
> > > > serialization
> > > > >> is
> > > > >> > > > still synchronous, write to HDFS is taken out of the
> operator
> > > > thread.
> > > > >> > > >
> > > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > > amol@datatorrent.com
> > > > >
> > > > >> > > wrote:
> > > > >> > > >
> > > > >> > > > >
> > > > >> > > > > Sent too soon. Has asynchronous checkpointing changed
> this?
> > > > >> > > > >
> > > > >> > > > > Amol
> > > > >> > > > >
> > > > >> > > > > Sent from my iPhone
> > > > >> > > > >
> > > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > > >> > > bhupesh@datatorrent.com>
> > > > >> > > > > wrote:
> > > > >> > > > > >
> > > > >> > > > > > Hi All,
> > > > >> > > > > >
> > > > >> > > > > > Is it safe to assume that the checkpointed() and the
> > > > >> beginWindow()
> > > > >> > > > calls
> > > > >> > > > > > are sequenced?
> > > > >> > > > > > In other words, are these calls part of the same thread
> > and
> > > > may
> > > > >> not
> > > > >> > > run
> > > > >> > > > > in
> > > > >> > > > > > parallel?
> > > > >> > > > > >
> > > > >> > > > > > Thanks.
> > > > >> > > > > >
> > > > >> > > > > > --
> > > > >> > > > > > -Bhupesh
> > > > >> > > > >
> > > > >> > > >
> > > > >> > >
> > > > >> >
> > > > >>
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Timothy Farkas <ti...@datatorrent.com>.
That is what it is looking like to me. The task is submitted
GenericNode#checkpoint line 504, then at the end of the
GenericNode#checkpoint line 531 checkpointed is called. I am likely missing
something, just would like to know what :)
Tim
On Tue, Nov 10, 2015 at 10:51 AM, Pramod Immaneni <pr...@datatorrent.com>
wrote:
> Tim,
>
> Are you suggesting that checkpointed is called before the checkpoint is
> completely persisted in the storage.
>
> Thanks
>
> On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <ti...@datatorrent.com>
> wrote:
>
> > Chetan,
> >
> > I do not see the process of reporting the checkpoint to stram, receiving
> > the ack, and then calling checkpointed. The logic I'm seeing in
> GenericNode
> > line 484 is that the checkpoint method is called, it spawns another
> thread
> > that writes to hdfs, and then checkpointed is immediately called
> > afterwards. I am missing something, can you give me some pointers so
> that I
> > can better understand the flow?
> >
> > Tim
> >
> > On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <ram@datatorrent.com
> >
> > wrote:
> >
> > > Chetan's answer provides a good explanation as well as clarifying that
> > > the difference can be more than 1.
> > >
> > > Since checkpointing (i.e. "commit notification" as Thomas refers to
> > > it) is asynchronous, I'm curious
> > > about whether the window ids in the checkpointed call are guaranteed
> > > to be sequential or if they could
> > > be out of order, i.e. can the checkpointed call see window id 101
> > > before it sees 100 ?
> > >
> > > Ram
> > >
> > > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > > <bh...@datatorrent.com> wrote:
> > > > Hi Tim,
> > > > Thanks for the detailed explanation.
> > > > I understand that the sequence would be
> > > > beginWindow (x) -> endWindow (x) -> checkpointed (x) -> beginWindow
> > > > (x+1)
> > > >
> > > > However when I try to print out the window ids in beginWindow,
> > endWindow
> > > > and checkpointed calls, I see x and x-1 respectively.
> > > > I.e. If the window just before checkpoint is 100, I see that the
> > > > checkpointed call had window id 99.
> > > >
> > > > Note: This is observed in the local mode of Apex.
> > > >
> > > > Thanks
> > > > -Bhupesh
> > > > On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com>
> wrote:
> > > >
> > > >> Hi Bhupesh,
> > > >>
> > > >> The sequencing of checkpoint called in relation to beginWindow and
> > > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> > > >> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are
> > not
> > > >> configured to be the same then it is possible that checkpointed is
> > > called
> > > >> within an application window. So the sequence of events would be
> this:
> > > >>
> > > >> beginWindow -> checkpointed -> endWindow
> > > >>
> > > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the
> > same
> > > >> (default). Then the sequence of calls would be this:
> > > >>
> > > >> beginWindow (100) -> endWindow (100) -> checkpointed (100) ->
> > > beginWindow
> > > >> (101)
> > > >>
> > > >> The numbers in the sequence represent possible streaming window Ids
> > that
> > > >> each call would be associated with.
> > > >>
> > > >> The StateMachine which calls these callbacks for non-input operators
> > is
> > > in
> > > >> GenericNode.java.
> > > >>
> > > >> Tim
> > > >>
> > > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > > bhupesh@datatorrent.com>
> > > >> wrote:
> > > >>
> > > >> > Hi Chetan / Community,
> > > >> >
> > > >> > Can someone please elaborate on why the window id supplied to
> > > >> > CheckpointListener and the Operator would differ.
> > > >> > I tried looking at the window ids of checkpointed() and the
> > > beginWindow()
> > > >> > calls and they differ by 1. Don't know why this should be the
> case.
> > > >> >
> > > >> > Thanks.
> > > >> > -Bhupesh
> > > >> >
> > > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > > chetan@datatorrent.com>
> > > >> > wrote:
> > > >> >
> > > >> > > Short answer is yes.
> > > >> > >
> > > >> > > All the control tuples are scheduled to be delivered outside of
> > the
> > > >> > window.
> > > >> > > As checkpointed callback is triggered because of CHECKPOINT
> > control
> > > >> > tuple,
> > > >> > > it will happen after endWindow and before the next beginWindow.
> > > >> > >
> > > >> > > The windowId supplied to CheckpointListener and the one provided
> > to
> > > >> > > Operator need not match even though the sequence is defined. So
> I
> > am
> > > >> > > curious how you intend to use this knowledge.
> > > >> > >
> > > >> > > --
> > > >> > > Chetan
> > > >> > >
> > > >> > >
> > > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > > thomas@datatorrent.com>
> > > >> > > wrote:
> > > >> > >
> > > >> > > > It has not changed the operator execution model. State
> > > serialization
> > > >> is
> > > >> > > > still synchronous, write to HDFS is taken out of the operator
> > > thread.
> > > >> > > >
> > > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> > amol@datatorrent.com
> > > >
> > > >> > > wrote:
> > > >> > > >
> > > >> > > > >
> > > >> > > > > Sent too soon. Has asynchronous checkpointing changed this?
> > > >> > > > >
> > > >> > > > > Amol
> > > >> > > > >
> > > >> > > > > Sent from my iPhone
> > > >> > > > >
> > > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > >> > > bhupesh@datatorrent.com>
> > > >> > > > > wrote:
> > > >> > > > > >
> > > >> > > > > > Hi All,
> > > >> > > > > >
> > > >> > > > > > Is it safe to assume that the checkpointed() and the
> > > >> beginWindow()
> > > >> > > > calls
> > > >> > > > > > are sequenced?
> > > >> > > > > > In other words, are these calls part of the same thread
> and
> > > may
> > > >> not
> > > >> > > run
> > > >> > > > > in
> > > >> > > > > > parallel?
> > > >> > > > > >
> > > >> > > > > > Thanks.
> > > >> > > > > >
> > > >> > > > > > --
> > > >> > > > > > -Bhupesh
> > > >> > > > >
> > > >> > > >
> > > >> > >
> > > >> >
> > > >>
> > >
> >
>
Re: Sequencing of operator calls
Posted by Pramod Immaneni <pr...@datatorrent.com>.
Tim,
Are you suggesting that checkpointed is called before the checkpoint is
completely persisted in the storage.
Thanks
On Tue, Nov 10, 2015 at 10:49 AM, Timothy Farkas <ti...@datatorrent.com>
wrote:
> Chetan,
>
> I do not see the process of reporting the checkpoint to stram, receiving
> the ack, and then calling checkpointed. The logic I'm seeing in GenericNode
> line 484 is that the checkpoint method is called, it spawns another thread
> that writes to hdfs, and then checkpointed is immediately called
> afterwards. I am missing something, can you give me some pointers so that I
> can better understand the flow?
>
> Tim
>
> On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <ra...@datatorrent.com>
> wrote:
>
> > Chetan's answer provides a good explanation as well as clarifying that
> > the difference can be more than 1.
> >
> > Since checkpointing (i.e. "commit notification" as Thomas refers to
> > it) is asynchronous, I'm curious
> > about whether the window ids in the checkpointed call are guaranteed
> > to be sequential or if they could
> > be out of order, i.e. can the checkpointed call see window id 101
> > before it sees 100 ?
> >
> > Ram
> >
> > On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> > <bh...@datatorrent.com> wrote:
> > > Hi Tim,
> > > Thanks for the detailed explanation.
> > > I understand that the sequence would be
> > > beginWindow (x) -> endWindow (x) -> checkpointed (x) -> beginWindow
> > > (x+1)
> > >
> > > However when I try to print out the window ids in beginWindow,
> endWindow
> > > and checkpointed calls, I see x and x-1 respectively.
> > > I.e. If the window just before checkpoint is 100, I see that the
> > > checkpointed call had window id 99.
> > >
> > > Note: This is observed in the local mode of Apex.
> > >
> > > Thanks
> > > -Bhupesh
> > > On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com> wrote:
> > >
> > >> Hi Bhupesh,
> > >>
> > >> The sequencing of checkpoint called in relation to beginWindow and
> > >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> > >> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are
> not
> > >> configured to be the same then it is possible that checkpointed is
> > called
> > >> within an application window. So the sequence of events would be this:
> > >>
> > >> beginWindow -> checkpointed -> endWindow
> > >>
> > >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the
> same
> > >> (default). Then the sequence of calls would be this:
> > >>
> > >> beginWindow (100) -> endWindow (100) -> checkpointed (100) ->
> > beginWindow
> > >> (101)
> > >>
> > >> The numbers in the sequence represent possible streaming window Ids
> that
> > >> each call would be associated with.
> > >>
> > >> The StateMachine which calls these callbacks for non-input operators
> is
> > in
> > >> GenericNode.java.
> > >>
> > >> Tim
> > >>
> > >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> > bhupesh@datatorrent.com>
> > >> wrote:
> > >>
> > >> > Hi Chetan / Community,
> > >> >
> > >> > Can someone please elaborate on why the window id supplied to
> > >> > CheckpointListener and the Operator would differ.
> > >> > I tried looking at the window ids of checkpointed() and the
> > beginWindow()
> > >> > calls and they differ by 1. Don't know why this should be the case.
> > >> >
> > >> > Thanks.
> > >> > -Bhupesh
> > >> >
> > >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> > chetan@datatorrent.com>
> > >> > wrote:
> > >> >
> > >> > > Short answer is yes.
> > >> > >
> > >> > > All the control tuples are scheduled to be delivered outside of
> the
> > >> > window.
> > >> > > As checkpointed callback is triggered because of CHECKPOINT
> control
> > >> > tuple,
> > >> > > it will happen after endWindow and before the next beginWindow.
> > >> > >
> > >> > > The windowId supplied to CheckpointListener and the one provided
> to
> > >> > > Operator need not match even though the sequence is defined. So I
> am
> > >> > > curious how you intend to use this knowledge.
> > >> > >
> > >> > > --
> > >> > > Chetan
> > >> > >
> > >> > >
> > >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> > thomas@datatorrent.com>
> > >> > > wrote:
> > >> > >
> > >> > > > It has not changed the operator execution model. State
> > serialization
> > >> is
> > >> > > > still synchronous, write to HDFS is taken out of the operator
> > thread.
> > >> > > >
> > >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <
> amol@datatorrent.com
> > >
> > >> > > wrote:
> > >> > > >
> > >> > > > >
> > >> > > > > Sent too soon. Has asynchronous checkpointing changed this?
> > >> > > > >
> > >> > > > > Amol
> > >> > > > >
> > >> > > > > Sent from my iPhone
> > >> > > > >
> > >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > >> > > bhupesh@datatorrent.com>
> > >> > > > > wrote:
> > >> > > > > >
> > >> > > > > > Hi All,
> > >> > > > > >
> > >> > > > > > Is it safe to assume that the checkpointed() and the
> > >> beginWindow()
> > >> > > > calls
> > >> > > > > > are sequenced?
> > >> > > > > > In other words, are these calls part of the same thread and
> > may
> > >> not
> > >> > > run
> > >> > > > > in
> > >> > > > > > parallel?
> > >> > > > > >
> > >> > > > > > Thanks.
> > >> > > > > >
> > >> > > > > > --
> > >> > > > > > -Bhupesh
> > >> > > > >
> > >> > > >
> > >> > >
> > >> >
> > >>
> >
>
Re: Sequencing of operator calls
Posted by Timothy Farkas <ti...@datatorrent.com>.
Chetan,
I do not see the process of reporting the checkpoint to stram, receiving
the ack, and then calling checkpointed. The logic I'm seeing in GenericNode
line 484 is that the checkpoint method is called, it spawns another thread
that writes to hdfs, and then checkpointed is immediately called
afterwards. I am missing something, can you give me some pointers so that I
can better understand the flow?
Tim
On Tue, Nov 10, 2015 at 10:33 AM, Munagala Ramanath <ra...@datatorrent.com>
wrote:
> Chetan's answer provides a good explanation as well as clarifying that
> the difference can be more than 1.
>
> Since checkpointing (i.e. "commit notification" as Thomas refers to
> it) is asynchronous, I'm curious
> about whether the window ids in the checkpointed call are guaranteed
> to be sequential or if they could
> be out of order, i.e. can the checkpointed call see window id 101
> before it sees 100 ?
>
> Ram
>
> On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
> <bh...@datatorrent.com> wrote:
> > Hi Tim,
> > Thanks for the detailed explanation.
> > I understand that the sequence would be
> > beginWindow (x) -> endWindow (x) -> checkpointed (x) -> beginWindow
> > (x+1)
> >
> > However when I try to print out the window ids in beginWindow, endWindow
> > and checkpointed calls, I see x and x-1 respectively.
> > I.e. If the window just before checkpoint is 100, I see that the
> > checkpointed call had window id 99.
> >
> > Note: This is observed in the local mode of Apex.
> >
> > Thanks
> > -Bhupesh
> > On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com> wrote:
> >
> >> Hi Bhupesh,
> >>
> >> The sequencing of checkpoint called in relation to beginWindow and
> >> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> >> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are not
> >> configured to be the same then it is possible that checkpointed is
> called
> >> within an application window. So the sequence of events would be this:
> >>
> >> beginWindow -> checkpointed -> endWindow
> >>
> >> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the same
> >> (default). Then the sequence of calls would be this:
> >>
> >> beginWindow (100) -> endWindow (100) -> checkpointed (100) ->
> beginWindow
> >> (101)
> >>
> >> The numbers in the sequence represent possible streaming window Ids that
> >> each call would be associated with.
> >>
> >> The StateMachine which calls these callbacks for non-input operators is
> in
> >> GenericNode.java.
> >>
> >> Tim
> >>
> >> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <
> bhupesh@datatorrent.com>
> >> wrote:
> >>
> >> > Hi Chetan / Community,
> >> >
> >> > Can someone please elaborate on why the window id supplied to
> >> > CheckpointListener and the Operator would differ.
> >> > I tried looking at the window ids of checkpointed() and the
> beginWindow()
> >> > calls and they differ by 1. Don't know why this should be the case.
> >> >
> >> > Thanks.
> >> > -Bhupesh
> >> >
> >> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <
> chetan@datatorrent.com>
> >> > wrote:
> >> >
> >> > > Short answer is yes.
> >> > >
> >> > > All the control tuples are scheduled to be delivered outside of the
> >> > window.
> >> > > As checkpointed callback is triggered because of CHECKPOINT control
> >> > tuple,
> >> > > it will happen after endWindow and before the next beginWindow.
> >> > >
> >> > > The windowId supplied to CheckpointListener and the one provided to
> >> > > Operator need not match even though the sequence is defined. So I am
> >> > > curious how you intend to use this knowledge.
> >> > >
> >> > > --
> >> > > Chetan
> >> > >
> >> > >
> >> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> thomas@datatorrent.com>
> >> > > wrote:
> >> > >
> >> > > > It has not changed the operator execution model. State
> serialization
> >> is
> >> > > > still synchronous, write to HDFS is taken out of the operator
> thread.
> >> > > >
> >> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <amol@datatorrent.com
> >
> >> > > wrote:
> >> > > >
> >> > > > >
> >> > > > > Sent too soon. Has asynchronous checkpointing changed this?
> >> > > > >
> >> > > > > Amol
> >> > > > >
> >> > > > > Sent from my iPhone
> >> > > > >
> >> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> >> > > bhupesh@datatorrent.com>
> >> > > > > wrote:
> >> > > > > >
> >> > > > > > Hi All,
> >> > > > > >
> >> > > > > > Is it safe to assume that the checkpointed() and the
> >> beginWindow()
> >> > > > calls
> >> > > > > > are sequenced?
> >> > > > > > In other words, are these calls part of the same thread and
> may
> >> not
> >> > > run
> >> > > > > in
> >> > > > > > parallel?
> >> > > > > >
> >> > > > > > Thanks.
> >> > > > > >
> >> > > > > > --
> >> > > > > > -Bhupesh
> >> > > > >
> >> > > >
> >> > >
> >> >
> >>
>
Re: Sequencing of operator calls
Posted by Munagala Ramanath <ra...@datatorrent.com>.
Chetan's answer provides a good explanation as well as clarifying that
the difference can be more than 1.
Since checkpointing (i.e. "commit notification" as Thomas refers to
it) is asynchronous, I'm curious
about whether the window ids in the checkpointed call are guaranteed
to be sequential or if they could
be out of order, i.e. can the checkpointed call see window id 101
before it sees 100 ?
Ram
On Tue, Nov 10, 2015 at 10:27 AM, Bhupesh Chawda
<bh...@datatorrent.com> wrote:
> Hi Tim,
> Thanks for the detailed explanation.
> I understand that the sequence would be
> beginWindow (x) -> endWindow (x) -> checkpointed (x) -> beginWindow
> (x+1)
>
> However when I try to print out the window ids in beginWindow, endWindow
> and checkpointed calls, I see x and x-1 respectively.
> I.e. If the window just before checkpoint is 100, I see that the
> checkpointed call had window id 99.
>
> Note: This is observed in the local mode of Apex.
>
> Thanks
> -Bhupesh
> On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com> wrote:
>
>> Hi Bhupesh,
>>
>> The sequencing of checkpoint called in relation to beginWindow and
>> endWindow depends on how your APPLICATION_WINDOW_COUNT and
>> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are not
>> configured to be the same then it is possible that checkpointed is called
>> within an application window. So the sequence of events would be this:
>>
>> beginWindow -> checkpointed -> endWindow
>>
>> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the same
>> (default). Then the sequence of calls would be this:
>>
>> beginWindow (100) -> endWindow (100) -> checkpointed (100) -> beginWindow
>> (101)
>>
>> The numbers in the sequence represent possible streaming window Ids that
>> each call would be associated with.
>>
>> The StateMachine which calls these callbacks for non-input operators is in
>> GenericNode.java.
>>
>> Tim
>>
>> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <bh...@datatorrent.com>
>> wrote:
>>
>> > Hi Chetan / Community,
>> >
>> > Can someone please elaborate on why the window id supplied to
>> > CheckpointListener and the Operator would differ.
>> > I tried looking at the window ids of checkpointed() and the beginWindow()
>> > calls and they differ by 1. Don't know why this should be the case.
>> >
>> > Thanks.
>> > -Bhupesh
>> >
>> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <ch...@datatorrent.com>
>> > wrote:
>> >
>> > > Short answer is yes.
>> > >
>> > > All the control tuples are scheduled to be delivered outside of the
>> > window.
>> > > As checkpointed callback is triggered because of CHECKPOINT control
>> > tuple,
>> > > it will happen after endWindow and before the next beginWindow.
>> > >
>> > > The windowId supplied to CheckpointListener and the one provided to
>> > > Operator need not match even though the sequence is defined. So I am
>> > > curious how you intend to use this knowledge.
>> > >
>> > > --
>> > > Chetan
>> > >
>> > >
>> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
>> > > wrote:
>> > >
>> > > > It has not changed the operator execution model. State serialization
>> is
>> > > > still synchronous, write to HDFS is taken out of the operator thread.
>> > > >
>> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
>> > > wrote:
>> > > >
>> > > > >
>> > > > > Sent too soon. Has asynchronous checkpointing changed this?
>> > > > >
>> > > > > Amol
>> > > > >
>> > > > > Sent from my iPhone
>> > > > >
>> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>> > > bhupesh@datatorrent.com>
>> > > > > wrote:
>> > > > > >
>> > > > > > Hi All,
>> > > > > >
>> > > > > > Is it safe to assume that the checkpointed() and the
>> beginWindow()
>> > > > calls
>> > > > > > are sequenced?
>> > > > > > In other words, are these calls part of the same thread and may
>> not
>> > > run
>> > > > > in
>> > > > > > parallel?
>> > > > > >
>> > > > > > Thanks.
>> > > > > >
>> > > > > > --
>> > > > > > -Bhupesh
>> > > > >
>> > > >
>> > >
>> >
>>
Re: Sequencing of operator calls
Posted by Bhupesh Chawda <bh...@datatorrent.com>.
Hi Tim,
Thanks for the detailed explanation.
I understand that the sequence would be
beginWindow (x) -> endWindow (x) -> checkpointed (x) -> beginWindow
(x+1)
However when I try to print out the window ids in beginWindow, endWindow
and checkpointed calls, I see x and x-1 respectively.
I.e. If the window just before checkpoint is 100, I see that the
checkpointed call had window id 99.
Note: This is observed in the local mode of Apex.
Thanks
-Bhupesh
On 10-Nov-2015 11:25 pm, "Timothy Farkas" <ti...@datatorrent.com> wrote:
> Hi Bhupesh,
>
> The sequencing of checkpoint called in relation to beginWindow and
> endWindow depends on how your APPLICATION_WINDOW_COUNT and
> CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are not
> configured to be the same then it is possible that checkpointed is called
> within an application window. So the sequence of events would be this:
>
> beginWindow -> checkpointed -> endWindow
>
> If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the same
> (default). Then the sequence of calls would be this:
>
> beginWindow (100) -> endWindow (100) -> checkpointed (100) -> beginWindow
> (101)
>
> The numbers in the sequence represent possible streaming window Ids that
> each call would be associated with.
>
> The StateMachine which calls these callbacks for non-input operators is in
> GenericNode.java.
>
> Tim
>
> On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <bh...@datatorrent.com>
> wrote:
>
> > Hi Chetan / Community,
> >
> > Can someone please elaborate on why the window id supplied to
> > CheckpointListener and the Operator would differ.
> > I tried looking at the window ids of checkpointed() and the beginWindow()
> > calls and they differ by 1. Don't know why this should be the case.
> >
> > Thanks.
> > -Bhupesh
> >
> > On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <ch...@datatorrent.com>
> > wrote:
> >
> > > Short answer is yes.
> > >
> > > All the control tuples are scheduled to be delivered outside of the
> > window.
> > > As checkpointed callback is triggered because of CHECKPOINT control
> > tuple,
> > > it will happen after endWindow and before the next beginWindow.
> > >
> > > The windowId supplied to CheckpointListener and the one provided to
> > > Operator need not match even though the sequence is defined. So I am
> > > curious how you intend to use this knowledge.
> > >
> > > --
> > > Chetan
> > >
> > >
> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
> > > wrote:
> > >
> > > > It has not changed the operator execution model. State serialization
> is
> > > > still synchronous, write to HDFS is taken out of the operator thread.
> > > >
> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
> > > wrote:
> > > >
> > > > >
> > > > > Sent too soon. Has asynchronous checkpointing changed this?
> > > > >
> > > > > Amol
> > > > >
> > > > > Sent from my iPhone
> > > > >
> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > bhupesh@datatorrent.com>
> > > > > wrote:
> > > > > >
> > > > > > Hi All,
> > > > > >
> > > > > > Is it safe to assume that the checkpointed() and the
> beginWindow()
> > > > calls
> > > > > > are sequenced?
> > > > > > In other words, are these calls part of the same thread and may
> not
> > > run
> > > > > in
> > > > > > parallel?
> > > > > >
> > > > > > Thanks.
> > > > > >
> > > > > > --
> > > > > > -Bhupesh
> > > > >
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Timothy Farkas <ti...@datatorrent.com>.
Hi Bhupesh,
The sequencing of checkpoint called in relation to beginWindow and
endWindow depends on how your APPLICATION_WINDOW_COUNT and
CHECKPOINT_WINDOW_COUNT are configured. If the two WINDOW_COUNTs are not
configured to be the same then it is possible that checkpointed is called
within an application window. So the sequence of events would be this:
beginWindow -> checkpointed -> endWindow
If the APPLICATION_WINDOW_COUNT and CHECKPOINT_WINDOW_COUNT are the same
(default). Then the sequence of calls would be this:
beginWindow (100) -> endWindow (100) -> checkpointed (100) -> beginWindow
(101)
The numbers in the sequence represent possible streaming window Ids that
each call would be associated with.
The StateMachine which calls these callbacks for non-input operators is in
GenericNode.java.
Tim
On Tue, Nov 10, 2015 at 3:36 AM, Bhupesh Chawda <bh...@datatorrent.com>
wrote:
> Hi Chetan / Community,
>
> Can someone please elaborate on why the window id supplied to
> CheckpointListener and the Operator would differ.
> I tried looking at the window ids of checkpointed() and the beginWindow()
> calls and they differ by 1. Don't know why this should be the case.
>
> Thanks.
> -Bhupesh
>
> On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <ch...@datatorrent.com>
> wrote:
>
> > Short answer is yes.
> >
> > All the control tuples are scheduled to be delivered outside of the
> window.
> > As checkpointed callback is triggered because of CHECKPOINT control
> tuple,
> > it will happen after endWindow and before the next beginWindow.
> >
> > The windowId supplied to CheckpointListener and the one provided to
> > Operator need not match even though the sequence is defined. So I am
> > curious how you intend to use this knowledge.
> >
> > --
> > Chetan
> >
> >
> > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
> > wrote:
> >
> > > It has not changed the operator execution model. State serialization is
> > > still synchronous, write to HDFS is taken out of the operator thread.
> > >
> > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
> > wrote:
> > >
> > > >
> > > > Sent too soon. Has asynchronous checkpointing changed this?
> > > >
> > > > Amol
> > > >
> > > > Sent from my iPhone
> > > >
> > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > bhupesh@datatorrent.com>
> > > > wrote:
> > > > >
> > > > > Hi All,
> > > > >
> > > > > Is it safe to assume that the checkpointed() and the beginWindow()
> > > calls
> > > > > are sequenced?
> > > > > In other words, are these calls part of the same thread and may not
> > run
> > > > in
> > > > > parallel?
> > > > >
> > > > > Thanks.
> > > > >
> > > > > --
> > > > > -Bhupesh
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by "Chetan Narsude (cnarsude)" <cn...@cisco.com>.
Checkpointing is an asynchronous operation. The following sub tasks take
place as part of checkpointing:
1 The state of an operator is collected as of the end of the windowId by
Streaming Container
2. The state is dispatched for saving to the persistent media (HDFS by
default)
3. After 2 is done the windowId is reported to Streaming App Master.
4. Streaming App Master updates its books and acks back to the Streaming
Container.
5. Streaming container invokes CheckpointListener interface.
These steps can take unpredictable amount of time and hence immediately
after finishing step 1, the operator moves on to processing next window
while step 2 onwards are executed in different threads and joined in step
5. Hence the windowId provided to CheckpointListener is often behind the
last window Id processed by the operator. The difference is typically 1
but it can be any positive integer.
‹
Chetan
On 11/10/15, 3:36 AM, "Bhupesh Chawda" <bh...@datatorrent.com> wrote:
>Hi Chetan / Community,
>
>Can someone please elaborate on why the window id supplied to
>CheckpointListener and the Operator would differ.
>I tried looking at the window ids of checkpointed() and the beginWindow()
>calls and they differ by 1. Don't know why this should be the case.
>
>Thanks.
>-Bhupesh
>
>On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <ch...@datatorrent.com>
>wrote:
>
>> Short answer is yes.
>>
>> All the control tuples are scheduled to be delivered outside of the
>>window.
>> As checkpointed callback is triggered because of CHECKPOINT control
>>tuple,
>> it will happen after endWindow and before the next beginWindow.
>>
>> The windowId supplied to CheckpointListener and the one provided to
>> Operator need not match even though the sequence is defined. So I am
>> curious how you intend to use this knowledge.
>>
>> --
>> Chetan
>>
>>
>> On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
>> wrote:
>>
>> > It has not changed the operator execution model. State serialization
>>is
>> > still synchronous, write to HDFS is taken out of the operator thread.
>> >
>> > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
>> wrote:
>> >
>> > >
>> > > Sent too soon. Has asynchronous checkpointing changed this?
>> > >
>> > > Amol
>> > >
>> > > Sent from my iPhone
>> > >
>> > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
>> bhupesh@datatorrent.com>
>> > > wrote:
>> > > >
>> > > > Hi All,
>> > > >
>> > > > Is it safe to assume that the checkpointed() and the beginWindow()
>> > calls
>> > > > are sequenced?
>> > > > In other words, are these calls part of the same thread and may
>>not
>> run
>> > > in
>> > > > parallel?
>> > > >
>> > > > Thanks.
>> > > >
>> > > > --
>> > > > -Bhupesh
>> > >
>> >
>>
Re: Sequencing of operator calls
Posted by Bhupesh Chawda <bh...@datatorrent.com>.
Hi Chetan / Community,
Can someone please elaborate on why the window id supplied to
CheckpointListener and the Operator would differ.
I tried looking at the window ids of checkpointed() and the beginWindow()
calls and they differ by 1. Don't know why this should be the case.
Thanks.
-Bhupesh
On Thu, Sep 17, 2015 at 5:56 AM, Chetan Narsude <ch...@datatorrent.com>
wrote:
> Short answer is yes.
>
> All the control tuples are scheduled to be delivered outside of the window.
> As checkpointed callback is triggered because of CHECKPOINT control tuple,
> it will happen after endWindow and before the next beginWindow.
>
> The windowId supplied to CheckpointListener and the one provided to
> Operator need not match even though the sequence is defined. So I am
> curious how you intend to use this knowledge.
>
> --
> Chetan
>
>
> On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
> wrote:
>
> > It has not changed the operator execution model. State serialization is
> > still synchronous, write to HDFS is taken out of the operator thread.
> >
> > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
> wrote:
> >
> > >
> > > Sent too soon. Has asynchronous checkpointing changed this?
> > >
> > > Amol
> > >
> > > Sent from my iPhone
> > >
> > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> bhupesh@datatorrent.com>
> > > wrote:
> > > >
> > > > Hi All,
> > > >
> > > > Is it safe to assume that the checkpointed() and the beginWindow()
> > calls
> > > > are sequenced?
> > > > In other words, are these calls part of the same thread and may not
> run
> > > in
> > > > parallel?
> > > >
> > > > Thanks.
> > > >
> > > > --
> > > > -Bhupesh
> > >
> >
>
Re: Sequencing of operator calls
Posted by Thomas Weise <th...@datatorrent.com>.
There seems to be confusion between "checkpointed" and "committed". I put
my comment on that JIRA.
Serializing the state of the operator is always blocking. Write to HDFS can
be asynchronous. Commit notification is always asynchronous.
Thomas
On Sat, Sep 19, 2015 at 10:19 AM, Amol Kekre <am...@datatorrent.com> wrote:
> Bhupesh,
> If you want to guaranteed that checkpoint is fully done before next tuple
> is processed, you will need to use blocking checkpoint attribute.
>
> Thks
> Amol
>
> On Sat, Sep 19, 2015 at 8:59 AM, Sandesh Hegde <sa...@datatorrent.com>
> wrote:
>
> > Blocking checkpoint is supported, please refer to the following jira,
> > https://malhar.atlassian.net/browse/APEX-19
> >
> > On Fri, Sep 18, 2015 at 9:08 PM Bhupesh Chawda <bh...@datatorrent.com>
> > wrote:
> >
> > > Thanks for the replies.
> > > My intention was to ask if the call to next begin window will start
> only
> > > after the checkpointed call has done it's job.
> > > So if just the call is sequenced, but it's a different thread, then it
> > does
> > > not really help in my case. I need to finish the job which I need to do
> > in
> > > checkpointed call before any new tuples are processed.
> > > On 17-Sep-2015 5:56 am, "Chetan Narsude" <ch...@datatorrent.com>
> wrote:
> > >
> > > > Short answer is yes.
> > > >
> > > > All the control tuples are scheduled to be delivered outside of the
> > > window.
> > > > As checkpointed callback is triggered because of CHECKPOINT control
> > > tuple,
> > > > it will happen after endWindow and before the next beginWindow.
> > > >
> > > > The windowId supplied to CheckpointListener and the one provided to
> > > > Operator need not match even though the sequence is defined. So I am
> > > > curious how you intend to use this knowledge.
> > > >
> > > > --
> > > > Chetan
> > > >
> > > >
> > > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <
> thomas@datatorrent.com>
> > > > wrote:
> > > >
> > > > > It has not changed the operator execution model. State
> serialization
> > is
> > > > > still synchronous, write to HDFS is taken out of the operator
> thread.
> > > > >
> > > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
> > > > wrote:
> > > > >
> > > > > >
> > > > > > Sent too soon. Has asynchronous checkpointing changed this?
> > > > > >
> > > > > > Amol
> > > > > >
> > > > > > Sent from my iPhone
> > > > > >
> > > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > > bhupesh@datatorrent.com>
> > > > > > wrote:
> > > > > > >
> > > > > > > Hi All,
> > > > > > >
> > > > > > > Is it safe to assume that the checkpointed() and the
> > beginWindow()
> > > > > calls
> > > > > > > are sequenced?
> > > > > > > In other words, are these calls part of the same thread and may
> > not
> > > > run
> > > > > > in
> > > > > > > parallel?
> > > > > > >
> > > > > > > Thanks.
> > > > > > >
> > > > > > > --
> > > > > > > -Bhupesh
> > > > > >
> > > > >
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Amol Kekre <am...@datatorrent.com>.
Bhupesh,
If you want to guaranteed that checkpoint is fully done before next tuple
is processed, you will need to use blocking checkpoint attribute.
Thks
Amol
On Sat, Sep 19, 2015 at 8:59 AM, Sandesh Hegde <sa...@datatorrent.com>
wrote:
> Blocking checkpoint is supported, please refer to the following jira,
> https://malhar.atlassian.net/browse/APEX-19
>
> On Fri, Sep 18, 2015 at 9:08 PM Bhupesh Chawda <bh...@datatorrent.com>
> wrote:
>
> > Thanks for the replies.
> > My intention was to ask if the call to next begin window will start only
> > after the checkpointed call has done it's job.
> > So if just the call is sequenced, but it's a different thread, then it
> does
> > not really help in my case. I need to finish the job which I need to do
> in
> > checkpointed call before any new tuples are processed.
> > On 17-Sep-2015 5:56 am, "Chetan Narsude" <ch...@datatorrent.com> wrote:
> >
> > > Short answer is yes.
> > >
> > > All the control tuples are scheduled to be delivered outside of the
> > window.
> > > As checkpointed callback is triggered because of CHECKPOINT control
> > tuple,
> > > it will happen after endWindow and before the next beginWindow.
> > >
> > > The windowId supplied to CheckpointListener and the one provided to
> > > Operator need not match even though the sequence is defined. So I am
> > > curious how you intend to use this knowledge.
> > >
> > > --
> > > Chetan
> > >
> > >
> > > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
> > > wrote:
> > >
> > > > It has not changed the operator execution model. State serialization
> is
> > > > still synchronous, write to HDFS is taken out of the operator thread.
> > > >
> > > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
> > > wrote:
> > > >
> > > > >
> > > > > Sent too soon. Has asynchronous checkpointing changed this?
> > > > >
> > > > > Amol
> > > > >
> > > > > Sent from my iPhone
> > > > >
> > > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > > bhupesh@datatorrent.com>
> > > > > wrote:
> > > > > >
> > > > > > Hi All,
> > > > > >
> > > > > > Is it safe to assume that the checkpointed() and the
> beginWindow()
> > > > calls
> > > > > > are sequenced?
> > > > > > In other words, are these calls part of the same thread and may
> not
> > > run
> > > > > in
> > > > > > parallel?
> > > > > >
> > > > > > Thanks.
> > > > > >
> > > > > > --
> > > > > > -Bhupesh
> > > > >
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Sandesh Hegde <sa...@datatorrent.com>.
Blocking checkpoint is supported, please refer to the following jira,
https://malhar.atlassian.net/browse/APEX-19
On Fri, Sep 18, 2015 at 9:08 PM Bhupesh Chawda <bh...@datatorrent.com>
wrote:
> Thanks for the replies.
> My intention was to ask if the call to next begin window will start only
> after the checkpointed call has done it's job.
> So if just the call is sequenced, but it's a different thread, then it does
> not really help in my case. I need to finish the job which I need to do in
> checkpointed call before any new tuples are processed.
> On 17-Sep-2015 5:56 am, "Chetan Narsude" <ch...@datatorrent.com> wrote:
>
> > Short answer is yes.
> >
> > All the control tuples are scheduled to be delivered outside of the
> window.
> > As checkpointed callback is triggered because of CHECKPOINT control
> tuple,
> > it will happen after endWindow and before the next beginWindow.
> >
> > The windowId supplied to CheckpointListener and the one provided to
> > Operator need not match even though the sequence is defined. So I am
> > curious how you intend to use this knowledge.
> >
> > --
> > Chetan
> >
> >
> > On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
> > wrote:
> >
> > > It has not changed the operator execution model. State serialization is
> > > still synchronous, write to HDFS is taken out of the operator thread.
> > >
> > > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
> > wrote:
> > >
> > > >
> > > > Sent too soon. Has asynchronous checkpointing changed this?
> > > >
> > > > Amol
> > > >
> > > > Sent from my iPhone
> > > >
> > > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> > bhupesh@datatorrent.com>
> > > > wrote:
> > > > >
> > > > > Hi All,
> > > > >
> > > > > Is it safe to assume that the checkpointed() and the beginWindow()
> > > calls
> > > > > are sequenced?
> > > > > In other words, are these calls part of the same thread and may not
> > run
> > > > in
> > > > > parallel?
> > > > >
> > > > > Thanks.
> > > > >
> > > > > --
> > > > > -Bhupesh
> > > >
> > >
> >
>
Re: Sequencing of operator calls
Posted by Bhupesh Chawda <bh...@datatorrent.com>.
Thanks for the replies.
My intention was to ask if the call to next begin window will start only
after the checkpointed call has done it's job.
So if just the call is sequenced, but it's a different thread, then it does
not really help in my case. I need to finish the job which I need to do in
checkpointed call before any new tuples are processed.
On 17-Sep-2015 5:56 am, "Chetan Narsude" <ch...@datatorrent.com> wrote:
> Short answer is yes.
>
> All the control tuples are scheduled to be delivered outside of the window.
> As checkpointed callback is triggered because of CHECKPOINT control tuple,
> it will happen after endWindow and before the next beginWindow.
>
> The windowId supplied to CheckpointListener and the one provided to
> Operator need not match even though the sequence is defined. So I am
> curious how you intend to use this knowledge.
>
> --
> Chetan
>
>
> On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
> wrote:
>
> > It has not changed the operator execution model. State serialization is
> > still synchronous, write to HDFS is taken out of the operator thread.
> >
> > On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com>
> wrote:
> >
> > >
> > > Sent too soon. Has asynchronous checkpointing changed this?
> > >
> > > Amol
> > >
> > > Sent from my iPhone
> > >
> > > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <
> bhupesh@datatorrent.com>
> > > wrote:
> > > >
> > > > Hi All,
> > > >
> > > > Is it safe to assume that the checkpointed() and the beginWindow()
> > calls
> > > > are sequenced?
> > > > In other words, are these calls part of the same thread and may not
> run
> > > in
> > > > parallel?
> > > >
> > > > Thanks.
> > > >
> > > > --
> > > > -Bhupesh
> > >
> >
>
Re: Sequencing of operator calls
Posted by Chetan Narsude <ch...@datatorrent.com>.
Short answer is yes.
All the control tuples are scheduled to be delivered outside of the window.
As checkpointed callback is triggered because of CHECKPOINT control tuple,
it will happen after endWindow and before the next beginWindow.
The windowId supplied to CheckpointListener and the one provided to
Operator need not match even though the sequence is defined. So I am
curious how you intend to use this knowledge.
--
Chetan
On Tue, Sep 15, 2015 at 8:31 AM, Thomas Weise <th...@datatorrent.com>
wrote:
> It has not changed the operator execution model. State serialization is
> still synchronous, write to HDFS is taken out of the operator thread.
>
> On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com> wrote:
>
> >
> > Sent too soon. Has asynchronous checkpointing changed this?
> >
> > Amol
> >
> > Sent from my iPhone
> >
> > > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <bh...@datatorrent.com>
> > wrote:
> > >
> > > Hi All,
> > >
> > > Is it safe to assume that the checkpointed() and the beginWindow()
> calls
> > > are sequenced?
> > > In other words, are these calls part of the same thread and may not run
> > in
> > > parallel?
> > >
> > > Thanks.
> > >
> > > --
> > > -Bhupesh
> >
>
Re: Sequencing of operator calls
Posted by Thomas Weise <th...@datatorrent.com>.
It has not changed the operator execution model. State serialization is
still synchronous, write to HDFS is taken out of the operator thread.
On Tue, Sep 15, 2015 at 8:18 AM, Amol Kekre <am...@datatorrent.com> wrote:
>
> Sent too soon. Has asynchronous checkpointing changed this?
>
> Amol
>
> Sent from my iPhone
>
> > On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <bh...@datatorrent.com>
> wrote:
> >
> > Hi All,
> >
> > Is it safe to assume that the checkpointed() and the beginWindow() calls
> > are sequenced?
> > In other words, are these calls part of the same thread and may not run
> in
> > parallel?
> >
> > Thanks.
> >
> > --
> > -Bhupesh
>
Re: Sequencing of operator calls
Posted by Amol Kekre <am...@datatorrent.com>.
Sent too soon. Has asynchronous checkpointing changed this?
Amol
Sent from my iPhone
> On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <bh...@datatorrent.com> wrote:
>
> Hi All,
>
> Is it safe to assume that the checkpointed() and the beginWindow() calls
> are sequenced?
> In other words, are these calls part of the same thread and may not run in
> parallel?
>
> Thanks.
>
> --
> -Bhupesh
Re: Sequencing of operator calls
Posted by Amol Kekre <am...@datatorrent.com>.
Apex guarantees single thread execution. The answer should be yes
Amol
Sent from my iPhone
> On Sep 15, 2015, at 12:38 AM, Bhupesh Chawda <bh...@datatorrent.com> wrote:
>
> Hi All,
>
> Is it safe to assume that the checkpointed() and the beginWindow() calls
> are sequenced?
> In other words, are these calls part of the same thread and may not run in
> parallel?
>
> Thanks.
>
> --
> -Bhupesh