Hi Aljoscha,
This is an excellent question! And the answer is, we don't need any new
concepts like "SDF executor" and can rely on the per-key state and timers
machinery that already exists in all runners because it's necessary to
implement windowing/triggering properly.
Note that this is already somewhat addressed in the previously posted State
and Timers proposal https://s.apache.org/beam-state , under "per-key
workflows".
Think of it this way, using the Kafka example: we'll expand it into a
transform:
(1) ParDo { topic -> (unique key, topic, partition, [0, inf))) for
partition in topic.listPartitions() }
(2) GroupByKey
(3) ParDo { key, topic, partition, R -> Kafka reader code in the
proposal/slides }
- R is the OffsetRange restriction which in this case will be always of
the form [startOffset, inf).
- there'll be just 1 value per key, but we use GBK to just get access to
the per-key state/timers machinery. This may be runner-specific; maybe some
runners don't need a GBK to do that.
Now suppose the topic has two partitions, P1 and P2, and they get assigned
unique keys K1, K2.
Then the input to (3) will be a collection of: (K1, topic, P1, [0, inf)),
(K2, topic, P2, [0, inf)).
Suppose we have just 1 worker with just 1 thread. Now, how will this thread
be able to produce elements from both P1 and P2? here's how.
The thread will process (K1, topic, P1, [0, inf)), checkpoint after a
certain time or after a certain number of elements are output (just like
with the current UnboundedSource reading code) producing a residual
restriction R1' (basically a new start timestamp), put R11 into the per-key
state and set a timer T1 to resume.
Then it will process (K2, topic, P2, [0, inf)), do the same producing a
residual restriction R2' and setting a timer T2 to resume.
Then timer T1 will fire in the context of the key K1. The thread will call
processElement again, this time supplying R1' as the restriction; the
process repeats and after a while it checkpoints and stores R1'' into state
of K1.
Then timer T2 will fire in the context of K2, run processElement for a
while, set a new timer and store R2'' into the state of K2.
Etc.
If partition 1 goes away, the processElement call will return "do not
resume", so a timer will not be set and instead the state associated with
K1 will be GC'd.
So basically it's almost like cooperative thread scheduling: things run for
a while, until the runner tells them to checkpoint, then they set a timer
to resume themselves, and the runner fires the timers, and the process
repeats. And, again, this only requires things that runners can already do
- state and timers, but no new concept of SDF executor (and consequently no
necessity to choose/tune how many you need).
Makes sense?
On Sat, Aug 20, 2016 at 9:46 AM Aljoscha Krettek <aljos...@apache.org>
wrote:
Hi,
I have another question that I think wasn't addressed in the meeting. At
least it wasn't mentioned in the notes.
In the context of replacing sources by a combination of to SDFs, how do you
determine how many "SDF executor" instances you need downstream? For the
sake of argument assume that both SDFs are executed with parallelism 1 (or
one per worker). Now, if you have a file source that reads from a static
set of files the first SDF would emit the filenames while the second SDF
would receive the filenames and emit their contents. This works well and
the downstream SDF can process one filename after the other. Now, think of
something like a Kafka source. The first SDF would emit the partitions (say
4 partitions, in this example) and the second SDF would be responsible for
reading from a topic and emitting elements. Reading from one topic never
finishes so you can't process the topics in series. I think you would need
to have 4 downstream "SDF executor" instances. The question now is: how do
you determine whether you are in the first or the second situation?
Probably I'm just overlooking something and this is already dealt with
somewhere... :-)
Cheers,
Aljoscha
On Fri, 19 Aug 2016 at 21:02 Ismaël Mejía <ieme...@gmail.com> wrote:
Hello,
Thanks for the notes both Dan and Eugene, and for taking the time to do
the
presentation and answer our questions.
I mentioned the ongoing work on dynamic scaling on Flink because I
suppose
that it will address dynamic rebalancing eventually (there are multiple
changes going on for dynamic scaling).
https://docs.google.com/document/d/1G1OS1z3xEBOrYD4wSu-LuBCyPUWyFd9l3T9WyssQ63w/edit#heading=h.2suhu1fjxmp4
https://lists.apache.org/list.html?d...@flink.apache.org:lte=1M:FLIP-8
Anyway I am far from an expert on flink, but probably the flink guys can
give their opinion about this and refer to a more precise document that
the
ones I mentioned..
Thanks again,
Ismaël
On Fri, Aug 19, 2016 at 8:52 PM, Jean-Baptiste Onofré <j...@nanthrax.net>
wrote:
Great summary Eugene and Dan.
And thanks again for the details, explanation, and discussion.
Regards
JB
On 08/19/2016 08:16 PM, Eugene Kirpichov wrote:
Thanks for attending, everybody!
Here are meeting notes (thanks Dan!).
Q: Will SplittableDoFn enable better repartitioning of the
input/output
data?
A: Not really; repartitioning is orthogonal to SDF.
Current Source API suffers from lack of composition and scalability
because
we treat sources too much as metadata, not enough as data.
Q(slide with transform expansion): who does the "magic"?
A: The runner. Checkpointing and dynamically splitting restrictions
will
require collaboration with the runner.
Q: How does the runner interact with the DoFn to control the
restrictions?
Is it related to the centralized job tracker etc.?
A: RestrictionTracker is a simple helper object, that exists purely on
the
worker while executing a single partition, and interacts with the
worker
harness part of the runner. Not to be confused with the centralized
job
tracker (master) - completely unrelated. Worker harness, of course,
interacts with the master in some relevant ways (e.g. Dataflow master
can
tell "you're a straggler, you should split").
Q: Is this a new DoFn subclass, or how will this integrate with the
existing code?
A: It's a feature of reflection-based DoFn (
https://s.apache.org/a-new-do
fn)
- just another optional parameter of type RestrictionTracker to
processElement() which is dynamically bound via reflection, so fully
backward/forward compatible, and looks to users like a regular DoFn.
Q: why is fractionClaimed a double?
A: It's a number in 0..1. Job-wide magic (e.g. autoscaling, dynamic
rebalancing) requires a uniform way to represent progress through
different
sources.
Q: Spark runner is microbatch-based, so this seems to map well onto
checkpoint/resume, right?
A: Yes; actually the Dataflow runner is, at a worker level, also
microbatch-based. The way SDF interacts with a runner will be very
similar
to how a Bounded/UnboundedSource interacts with a runner.
Q: Using SDF, what would be the "packaging" of the IO?
A: Same as currently: package IO's as PTransforms and their
implementation
under the hood can be anything: Source, simple ParDo's, SDF, etc. E.g.
Datastore was recently refactored from BoundedSource to ParDo (ended
up
simpler and more scalable), transparently to users.
Q: What's the timeline; what to do with the IOs currently in
development?
A: Timeline is O(months). Keep doing what you're doing and working on
top
of Source APIs when necessary and simple ParDo's otherwise.
Q: What's the impact for the runner writers?
A: Tentatively expected that most of the code for running an SDF will
be
common to runners, with some amount of per-runner glue code, just like
GBK/windowing/triggering. Impact on Dataflow runner is larger since it
supports dynamic rebalancing in batch mode and this is the hardest
part,
but for other runners shouldn't be too hard.
JB: Talend has people who can help with this: e.g. help integrate into
Spark runner, refactor IOs etc. Amit also willing to chat about
supporting
SDF in Spark runner.
Ismael: There's a Flink proposal about dynamic rebalancing. Ismael
will
send a link.
On Fri, Aug 19, 2016 at 7:21 AM Jean-Baptiste Onofré <j...@nanthrax.net
wrote:
Hi Eugene,
thanks for the reminder.
Just to prepare some topics for the call, please find some points:
1. Using SDF, what would be the "packaging" of the IO ? It sounds to
me
that we can keep the IO packaging style (using with* setters for the
IO
configuration) and replace PTransform, Source, Reader, ... directly
with
SDF. Correct ?
2. What's your plan in term of release to include SDF ? We have
several
IOs in preparation and I wonder if it's worth to start to use the new
SDF API or not.
3. What's the impact for the runner writers ? The runners will have
to
support SDF, that could be tricky depending of the execution engine.
In
the worst case where the runner can't fully support SDF, does it mean
that most of our IOs will be useless ?
Just my dumb topics ;)
Thanks,
See you at 8am !
Regards
JB
On 08/19/2016 02:29 AM, Eugene Kirpichov wrote:
Hello everybody,
Just a reminder:
The meeting is happening tomorrow - Friday Aug 19th, 8am-9am PST, to
join
the call go to
https://hangouts.google.com/hangouts/_/google.com/splittabledofn .
I intend to go over the proposed design and then have a free-form
discussion.
Please have a skim through the proposal doc: https://s.apache.org/
splittable-do-fn
I also made some slides that are basically a trimmed-down version of
the
doc to use as a guide when conducting the meeting,
https://docs.google.com/presentation/d/1DfCp9VRLH-5Ap5nbz047
Fiv3NfjrywJVSdef5dRY4aE/edit?usp=sharing
.
I will post notes from the meeting on this thread afterwards.
Thanks, looking forward.
On Fri, Aug 12, 2016 at 5:35 PM Dan Halperin
<dhalp...@google.com.invalid
wrote:
This is pretty cool! I'll be there too. (unless the hangout gets too
full
-- if so, I'll drop out in favor of others who aren't lucky enough
to
get
to talk to Eugene all the time.)
On Fri, Aug 12, 2016 at 4:03 PM, Andrew Psaltis <
psaltis.and...@gmail.com>
wrote:
+1 I'll join
On Friday, August 12, 2016, Aparup Banerjee (apbanerj) <
apban...@cisco.com
wrote:
+ 1, me2
On 8/12/16, 9:27 AM, "Amit Sela" <amitsel...@gmail.com
<javascript:;>>
wrote:
+1 as in I'll join ;-)
On Fri, Aug 12, 2016, 19:14 Eugene Kirpichov
<kirpic...@google.com.invalid
wrote:
Sounds good, thanks!
Then Friday Aug 19th it is, 8am-9am PST,
https://staging.talkgadget.google.com/hangouts/_/google.
com/splittabledofn
On Thu, Aug 11, 2016 at 11:12 PM Jean-Baptiste Onofré <
j...@nanthrax.net
<javascript:;>>
wrote:
Hi
Unfortunately I will be in Ireland on August 15th. What about
Friday
19th ?
Regards
JB
On Aug 11, 2016, 23:22, at 23:22, Eugene Kirpichov
<kirpic...@google.com.INVALID> wrote:
Hi JB,
Sounds great, does the suggested time over videoconference
work
for
you?
On Thu, Aug 11, 2016 at 11:59 AM Jean-Baptiste Onofré <
j...@nanthrax.net <javascript:;>>
wrote:
Hi Eugene
May we talk together next week ? I like the proposal. I
would
just
need
some details for my understanding.
Thanks
Regards
JB
On Aug 11, 2016, 19:46, at 19:46, Eugene Kirpichov
<kirpic...@google.com.INVALID> wrote:
Hi JB,
What are your thoughts on this?
I'm also thinking of having a virtual meeting to explain
more
about
this
proposal if necessary, since I understand it is a lot to
digest.
How about: Monday Aug 15, 8am-9am Pacific time, over
Hangouts?
(link:
https://staging.talkgadget.google.com/hangouts/_/google.
com/splittabledofn
-
I confirmed that it can be joined without being logged
into a
Google
account)
Who'd be interested in attending, and does this time/date
work
for
people?
On Fri, Aug 5, 2016 at 10:48 AM Eugene Kirpichov
<kirpic...@google.com <javascript:;>>
wrote:
Hi JB, thanks for reading and for your comments!
It sounds like you are concerned about continued support
for
existing
IO's
people have developed, and about backward compatibility?
We do not need to remove the Source API, and all existing
Source-based
connectors will continue to work [though the document
proposes
at
some
point to make Read.from(Source) to translate to a wrapper
SDF
under
the
hood, to exercise the feature more and to make sure that
it
is
strictly
more powerful - but this is an optional implementation
detail].
Perhaps the document phrases this too strongly -
"replacing
the
Source
API": a better phrasing would be "introducing a new API so
powerful
and
easy-to-use that hopefully people will choose it over the
Source
API
all
the time, even though they don't have to" :) And we can
discuss
whether or
not to actually deprecate/remove the Source API at some
point
down
the
road, once it becomes clear whether this is the case or
not.
To give more context: this proposal came out of
discussions
within
the SDK
team over the past ~1.5 years, before the Beam project
existed,
on
how to
make major improvements to the Source API; perhaps it will
clarify
things
if I give a history of the ideas discussed:
- The first idea was to introduce a
Read.from(PCollection<Source>)
transform while keeping the Source API intact - this, given
appropriate
implementation, would solve most of the scalability and
composability
issues of IO's. Then most connectors would look like :
ParDo<A,
Source<B>>
+ Read.from().
- Then we figured that the Source class is an unnecessary
abstraction, as
it simply holds data. What if we only had a Reader<S, B>
class
where
S is
the source type and B the output type? Then connectors
would
be
something
like: ParDo<A, S> + hypothetical Read.using(Reader<S, B>).
- Then somebody remarked that some of the features of
Source
are
useful to
ParDo's as well: e.g. ability to report progress when
processing a
very
heavy element, or ability to produce very large output in
parallel.
- The two previous bullets were already hinting that the
Read.using()
primitive might not be so special: it just takes S and
produces
B:
isn't
that what a ParDo does, plus some source magic, minus the
convenience
of
c.output() vs. the start/advance() state machine?
- At this point it became clear that we should explore
unifying
sources
and ParDo's, in particular: can we bring the magic of
sources
to
ParDo's
but without the limitations and coding inconveniences? And
this
is
how
SplittableDoFn was born: bringing source magic to a DoFn
by
providing
a
RangeTracker.
- Once the idea of "splittable DoFn's" was born, it became
clear
that
it
is strictly more general than sources; at least, in the
respect
that
sources have to produce output, while DoFn's don't: an SDF
may
very
well
produce no output at all, and simply perform a side effect
in
a
parallel/resumable way.
- Then there were countless hours of discussions on
unifying
the
bounded/unbounded cases, on the particulars of RangeTracker
APIs
reconciling parallelization and checkpointing, what the
relation
between
SDF and DF should be, etc. They culminated in the current
proposal.
The
proposal comes at a time when a couple of key ingredients
are
(almost)
ready: NewDoFn to make SDF look like a regular DoFn, and
the
State/Timers
proposal to enable unbounded work per element.
To put it shortly:
- Yes, we will support existing Source connectors, and
will
support
writing new ones, possibly forever. There is no interference
with
current
users of Source.
- The new API is an attempt to improve the Source API,
taken
to
its
logical limit where it turns out that users' goals can be
accomplished
easier and more generically entirely within ParDo's.
Let me know what you think, and thanks again!
On Fri, Aug 5, 2016 at 2:39 AM Jean-Baptiste Onofré
<j...@nanthrax.net <javascript:;>>
wrote:
Hi Eugene,
Just a question: why is it in DoFn and note an
improvement
of
Source
?
If I understand correctly, it means that we will have to
refactore
all
existing IO: basically, what you propose is to remove all
Source
to
replace with NewDoFn.
I'm concern with this approach, especially in term of
timing:
clearly,
the IO is the area where we have to move forward in Beam
as
it
will
allow new users to start in their projects.
So, we started to bring new IOs: Kafka, JMS, Cassandra,
MongoDB,
JDBC,
... and some people started to learn the IO API
(Bounded/Unbouded
source, etc).
I think it would make more sense to enhance the IO API
(Source)
instead
of introducing a NewDoFn.
What are your thoughts for IO writer like me ? ;)
Regards
JB
On 08/04/2016 07:45 PM, Eugene Kirpichov wrote:
Hello Beam community,
We (myself, Daniel Mills and Robert Bradshaw) would like
to
propose
"Splittable DoFn" - a major generalization of DoFn, which
allows
processing
of a single element to be non-monolithic, i.e.
checkpointable
and
parallelizable, as well as doing an unbounded amount of
work
per
element.
This allows effectively replacing the current
Bounded/UnboundedSource
APIs
with DoFn's that are much easier to code, more scalable
and
composable
with
the rest of the Beam programming model, and enables many
use
cases
that
were previously difficult or impossible, as well as some
non-obvious new
use cases.
This proposal has been mentioned before in JIRA
[BEAM-65]
and
some
Beam
meetings, and now the whole thing is written up in a
document:
https://s.apache.org/splittable-do-fn
Here are some things that become possible with
Splittable
DoFn:
- Efficiently read a filepattern matching millions of
files
- Read a collection of files that are produced by an
earlier
step
in the
pipeline (e.g. easily implement a connector to a storage
system
that can
export itself to files)
- Implement a Kafka reader by composing a "list
partitions"
DoFn
with a
DoFn that simply polls a consumer and outputs new records
in
a
while()
loop
- Implement a log tailer by composing a DoFn that
incrementally
returns
new
files in a directory and a DoFn that tails a file
- Implement a parallel "count friends in common"
algorithm
(matrix
squaring) with good work balancing
Here is the meaningful part of a hypothetical Kafka
reader
written
against
this API:
ProcessContinuation processElement(
ProcessContext context, OffsetRangeTracker
tracker)
{
try (KafkaConsumer<String, String> consumer =
Kafka.subscribe(context.element().topic,
context.element().partition)) {
consumer.seek(tracker.start());
while (true) {
ConsumerRecords<String, String> records =
consumer.poll(100ms);
if (records == null) return done();
for (ConsumerRecord<String, String> record :
records)
{
if (!tracker.tryClaim(record.offset())) {
return
resume().withFutureOutputWatermark(record.timestamp());
}
context.output(record);
}
}
}
}
The document describes in detail the motivations behind
this
feature,
the
basic idea and API, open questions, and outlines an
incremental
delivery
plan.
The proposed API builds on the reflection-based new DoFn
[new-do-fn]
and is
loosely related to "State and Timers for DoFn"
[beam-state].
Please take a look and comment!
Thanks.
[BEAM-65] https://issues.apache.org/jira/browse/BEAM-65
[new-do-fn] https://s.apache.org/a-new-do-fn
[beam-state] https://s.apache.org/beam-state
--
Jean-Baptiste Onofré
jbono...@apache.org <javascript:;>
http://blog.nanthrax.net
Talend - http://www.talend.com
--
Thanks,
Andrew
Subscribe to my book: Streaming Data <http://manning.com/psaltis>
<https://www.linkedin.com/pub/andrew-psaltis/1/17b/306>
twiiter: @itmdata <
http://twitter.com/intent/user?screen_name=itmdata
--
Jean-Baptiste Onofré
jbono...@apache.org
http://blog.nanthrax.net
Talend - http://www.talend.com
--
Jean-Baptiste Onofré
jbono...@apache.org
http://blog.nanthrax.net
Talend - http://www.talend.com
