How do you declare WindowedOperator?
Thank you,
Vlad
On 4/28/17 10:35, AJAY GUPTA wrote:
Vlad,
The approach you suggested doesn't work because the CSVParser outputs
Object Data Type irrespective of the POJO class being emitted.
Ajay
On Fri, Apr 28, 2017 at 8:13 PM, Vlad Rozov <v.ro...@datatorrent.com> wrote:
Make your POJO class implement WindowedOperator Tuple interface (it may
return itself in getValue()).
Thank you,
Vlad
On 4/28/17 02:44, AJAY GUPTA wrote:
Hi All,
I am creating an application which is using Windowed Operator. This
application involves CsvParser operator emitting a POJO object which is to
be passed as input to WindowedOperator. The WindowedOperator requires an
instance of Tuple class as input :
*public final transient DefaultInputPort<Tuple<InputT>>
input = new DefaultInputPort<Tuple<InputT>>() *
Due to this, the addStream cannot work as the output of CsvParser's output
port is not compatible with input port type of WindowedOperator.
One way to solve this problem is to have an operator between the above two
operators as a convertor.
I would like to know if there is any other more generic approach to solve
this problem without writing a new Operator for every new application
using
Windowed Operators.
Thanks,
Ajay
On Thu, Mar 23, 2017 at 5:25 PM, Bhupesh Chawda <bhup...@datatorrent.com>
wrote:
Hi All,
I think we have some agreement on the way we should use control tuples
for
File I/O operators to support batch.
In order to have more operators in Malhar, support this paradigm, I think
we should also look at store operators - JDBC, Cassandra, HBase etc.
The case with these operators is simpler as most of these do not poll the
sources (except JDBC poller operator) and just stop once they have read a
fixed amount of data. In other words, these are inherently batch sources.
The only change that we should add to these operators is to shut down the
DAG once the reading of data is done. For a windowed operator this would
mean a Global window with a final watermark before the DAG is shut down.
~ Bhupesh
_______________________________________________________
Bhupesh Chawda
E: bhup...@datatorrent.com | Twitter: @bhupeshsc
www.datatorrent.com | apex.apache.org
On Tue, Feb 28, 2017 at 10:59 PM, Bhupesh Chawda <
bhup...@datatorrent.com>
wrote:
Hi Thomas,
Even though the windowing operator is not just "event time", it seems it
is too much dependent on the "time" attribute of the incoming tuple.
This
is the reason we had to model the file index as a timestamp to solve the
batch case for files.
Perhaps we should work on increasing the scope of the windowed operator
to
consider other types of windows as well. The Sequence option suggested
by
David seems to be something in that direction.
~ Bhupesh
_______________________________________________________
Bhupesh Chawda
E: bhup...@datatorrent.com | Twitter: @bhupeshsc
www.datatorrent.com | apex.apache.org
On Tue, Feb 28, 2017 at 10:48 PM, Thomas Weise <t...@apache.org> wrote:
That's correct, we are looking at a generalized approach for state
management vs. a series of special cases.
And to be clear, windowing does not imply event time, otherwise it
would
be
"EventTimeOperator" :-)
Thomas
On Tue, Feb 28, 2017 at 9:11 AM, Bhupesh Chawda <
bhup...@datatorrent.com>
wrote:
Hi David,
I went through the discussion, but it seems like it is more on the
event
time watermark handling as opposed to batches. What we are trying to
do
is
have watermarks serve the purpose of demarcating batches using control
tuples. Since each batch is separate from others, we would like to
have
stateful processing within a batch, but not across batches.
At the same time, we would like to do this in a manner which is
consistent
with the windowing mechanism provided by the windowed operator. This
will
allow us to treat a single batch as a (bounded) stream and apply all
the
event time windowing concepts in that time span.
For example, let's say I need to process data for a day (24 hours) as
a
single batch. The application is still streaming in nature: it would
end
the batch after a day and start a new batch the next day. At the same
time,
I would be able to have early trigger firings every minute as well as
drop
any data which is say, 5 mins late. All this within a single day.
~ Bhupesh
_______________________________________________________
Bhupesh Chawda
E: bhup...@datatorrent.com | Twitter: @bhupeshsc
www.datatorrent.com | apex.apache.org
On Tue, Feb 28, 2017 at 9:27 PM, David Yan <david...@gmail.com>
wrote:
There is a discussion in the Flink mailing list about key-based
watermarks.
I think it's relevant to our use case here.
https://lists.apache.org/thread.html/2b90d5b1d5e2654212cfbbcc6510ef
424bbafc4fadb164bd5aff9216@%3Cdev.flink.apache.org%3E
David
On Tue, Feb 28, 2017 at 2:13 AM, Bhupesh Chawda <
bhup...@datatorrent.com
wrote:
Hi David,
If using time window does not seem appropriate, we can have
another
class
which is more suited for such sequential and distinct windows.
Perhaps, a
CustomWindow option can be introduced which takes in a window id.
The
purpose of this window option could be to translate the window id
into
appropriate timestamps.
Another option would be to go with a custom timestampExtractor for
such
tuples which translates the each unique file name to a distinct
timestamp
while using time windows in the windowed operator.
~ Bhupesh
_______________________________________________________
Bhupesh Chawda
E: bhup...@datatorrent.com | Twitter: @bhupeshsc
www.datatorrent.com | apex.apache.org
On Tue, Feb 28, 2017 at 12:28 AM, David Yan <david...@gmail.com>
wrote:
I now see your rationale on putting the filename in the window.
As far as I understand, the reasons why the filename is not part
of
the
key
and the Global Window is not used are:
1) The files are processed in sequence, not in parallel
2) The windowed operator should not keep the state associated
with
the
file
when the processing of the file is done
3) The trigger should be fired for the file when a file is done
processing.
However, if the file is just a sequence has nothing to do with a
timestamp,
assigning a timestamp to a file is not an intuitive thing to do
and
would
just create confusions to the users, especially when it's used
as
an
example for new users.
How about having a separate class called SequenceWindow? And
perhaps
TimeWindow can inherit from it?
David
On Mon, Feb 27, 2017 at 8:58 AM, Thomas Weise <t...@apache.org>
wrote:
On Mon, Feb 27, 2017 at 8:50 AM, Bhupesh Chawda <
bhup...@datatorrent.com
wrote:
I think my comments related to count based windows might be
causing
confusion. Let's not discuss count based scenarios for now.
Just want to make sure we are on the same page wrt. the
"each
file
is a
batch" use case. As mentioned by Thomas, the each tuple from
the
same
file
has the same timestamp (which is just a sequence number) and
that
helps
keep tuples from each file in a separate window.
Yes, in this case it is a sequence number, but it could be a
time
stamp
also, depending on the file naming convention. And if it was
event
time
processing, the watermark would be derived from records within
the
file.
Agreed, the source should have a mechanism to control the time
stamp
extraction along with everything else pertaining to the
watermark
generation.
We could also implement a "timestampExtractor" interface to
identify
the
timestamp (sequence number) for a file.
~ Bhupesh
_______________________________________________________
Bhupesh Chawda
E: bhup...@datatorrent.com | Twitter: @bhupeshsc
www.datatorrent.com | apex.apache.org
On Mon, Feb 27, 2017 at 9:52 PM, Thomas Weise <
t...@apache.org
wrote:
I don't think this is a use case for count based window.
We have multiple files that are retrieved in a sequence
and
there
is
no
knowledge of the number of records per file. The
requirement is
to
aggregate each file separately and emit the aggregate when
the
file
is
read
fully. There is no concept of "end of something" for an
individual
key
and
global window isn't applicable.
However, as already explained and implemented by Bhupesh,
this
can
be
solved using watermark and window (in this case the window
timestamp
isn't
a timestamp, but a file sequence, but that doesn't matter.
Thomas
On Mon, Feb 27, 2017 at 8:05 AM, David Yan <
david...@gmail.com
wrote:
I don't think this is the way to go. Global Window only
means
the
timestamp
does not matter (or that there is no timestamp). It does
not
necessarily
mean it's a large batch. Unless there is some notion of
event
time
for
each
file, you don't want to embed the file into the window
itself.
If you want the result broken up by file name, and if
the
files
are
to
be
processed in parallel, I think making the file name be
part
of
the
key
is
the way to go. I think it's very confusing if we somehow
make
the
file
to
be part of the window.
For count-based window, it's not implemented yet and
you're
welcome
to
add
that feature. In case of count-based windows, there
would
be
no
notion
of
time and you probably only trigger at the end of each
window.
In
the
case
of count-based windows, the watermark only matters for
batch
since
you
need
a way to know when the batch has ended (if the count is
10,
the
number
of
tuples in the batch is let's say 105, you need a way to
end
the
last
window
with 5 tuples).
David
On Mon, Feb 27, 2017 at 2:41 AM, Bhupesh Chawda <
bhup...@datatorrent.com
wrote:
Hi David,
Thanks for your comments.
The wordcount example that I created based on the
windowed
operator
does
processing of word counts per file (each file as a
separate
batch),
i.e.
process counts for each file and dump into separate
files.
As I understand Global window is for one large batch;
i.e.
all
incoming
data falls into the same batch. This could not be
processed
using
GlobalWindow option as we need more than one windows.
In
this
case, I
configured the windowed operator to have time windows
of
1ms
each
and
passed data for each file with increasing timestamps:
(file1,
1),
(file2,
2) and so on. Is there a better way of handling this
scenario?
Regarding (2 - count based windows), I think there is
a
trigger
option
to
process count based windows. In case I want to process
every
1000
tuples
as
a batch, I could set the Trigger option to
CountTrigger
with
the
accumulation set to Discarding. Is this correct?
I agree that (4. Final Watermark) can be done using
Global
window.
~ Bhupesh
______________________________
_________________________
Bhupesh Chawda
E: bhup...@datatorrent.com | Twitter: @bhupeshsc
www.datatorrent.com | apex.apache.org
On Mon, Feb 27, 2017 at 12:18 PM, David Yan <
david...@gmail.com>
wrote:
I'm worried that we are making the watermark concept
too
complicated.
Watermarks should simply just tell you what windows
can
be
considered
complete.
Point 2 is basically a count-based window.
Watermarks
do
not
play a
role
here because the window is always complete at the
n-th
tuple.
If I understand correctly, point 3 is for batch
processing
of
files.
Unless
the files contain timed events, it sounds to be that
this
can
be
achieved
with just a Global Window. For signaling EOF, a
watermark
with
a
+infinity
timestamp can be used so that triggers will be fired
upon
receipt
of
that
watermark.
For point 4, just like what I mentioned above, can
be
achieved
with a
watermark with a +infinity timestamp.
David
On Sat, Feb 18, 2017 at 8:04 AM, Bhupesh Chawda <
bhup...@datatorrent.com
wrote:
Hi Thomas,
For an input operator which is supposed to
generate
watermarks
for
downstream operators, I can think about the
following
watermarks
that
the
operator can emit:
1. Time based watermarks (the high watermark / low
watermark)
2. Number of tuple based watermarks (Every n
tuples)
3. File based watermarks (Start file, end file)
4. Final watermark
File based watermarks seem to be applicable for
batch
(file
based)
as
well,
and hence I thought of looking at these first.
Does
this
seem
to
be
in
line
with the thought process?
~ Bhupesh
______________________________
_________________________
Bhupesh Chawda
Software Engineer
E: bhup...@datatorrent.com | Twitter: @bhupeshsc
www.datatorrent.com | apex.apache.org
On Thu, Feb 16, 2017 at 10:37 AM, Thomas Weise <
t...@apache.org
wrote:
I don't think this should be designed based on a
simplistic
file
input-output scenario. It would be good to
include a
stateful
transformation based on event time.
More complex pipelines contain stateful
transformations
that
depend
on
windowing and watermarks. I think we need a
watermark
concept
that
is
based
on progress in event time (or other monotonic
increasing
sequence)
that
other operators can generically work with.
Note that even file input in many cases can
produce
time
based
watermarks,
for example when you read part files that are
bound
by
event
time.
Thanks,
Thomas
On Wed, Feb 15, 2017 at 4:02 AM, Bhupesh Chawda
<
bhup...@datatorrent.com
