Mailing-List: contact dev-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
MIME-Version: 1.0
Sender: ewenstephan@gmail.com
In-Reply-To: <CA+3U0Zm551_62K6rKuUCrvUycPSc-5DyHCeK_5Zx_rmxrrHXvQ@mail.gmail.com>
References: <BLU437-SMTP7220A1C64AA32F44EBD8FBD1440@phx.gbl>
	<CAKADb_PdxLw_NxMKBG9wW5vUeLm9cZpE8Z+ucoY7KUBu0oYFkg@mail.gmail.com>
	<BLU436-SMTP975D4A31448106E3A23299D1450@phx.gbl>
	<CANC1h_vjnWesDethK-2NQ6GWAd1Mx=yGzDU36sLdg3pze7Tj1g@mail.gmail.com>
	<CA+3U0Zm551_62K6rKuUCrvUycPSc-5DyHCeK_5Zx_rmxrrHXvQ@mail.gmail.com>
Date: Tue, 31 May 2016 11:27:13 +0200
Message-ID: <CANC1h_v5dxvjYhWJt=FwBXm0zvOQ1__Quy08Bqodf5V9ew2_Gw@mail.gmail.com>
Subject: Re: Side-effects of DataSet::count
From: Stephan Ewen <sewen@apache.org>
To: "dev@flink.apache.org" <dev@flink.apache.org>
Content-Type: multipart/alternative; boundary=001a1144b952686d0d05341ffa53
archived-at: Tue, 31 May 2016 09:27:25 -0000

--001a1144b952686d0d05341ffa53
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

Hi!

There was some preliminary work on this. By now, the requirements have
grown a bit. The backtracking needs to handle

  - Scheduling for execution (the here raised point), possibly resuming
from available intermediate results
  - Recovery from partially executed programs, where operators execute
whole or not (batch style)
  - Recover from intermediate result since latest completed checkpoint
  - Eventually even recover superstep-based iterations.

So the design needs to be extended slightly. We do not have a design
writeup for this, but I agree, it would be great to have one.
I have a pretty good general idea about this, let me see if I can get to
that next week.

In general, for such things (long standing ideas and designs), we should
have something like Kafka has with its KIPs (Kafka Improvement Proposal) -
a place where to collect them, refine them over time, and
see how people react to them or step up to implement them. We could call
them 3Fs (Flink Feature Forms) ;-)

Greetings,
Stephan


On Tue, May 31, 2016 at 1:02 AM, Greg Hogan <code@greghogan.com> wrote:

> Hi Stephan,
>
> Is there a design document, prior discussion, or background material on
> this enhancement? Am I correct in understanding that this only applies to
> DataSet since streams run indefinitely?
>
> Thanks,
> Greg
>
> On Mon, May 30, 2016 at 5:49 PM, Stephan Ewen <sewen@apache.org> wrote:
>
> > Hi Eron!
> >
> > Yes, the idea is to actually switch all executions to a backtracking
> > scheduling mode. That simultaneously solves both fine grained recovery
> and
> > lazy execution, where later stages build on prior stages.
> >
> > With all the work around streaming, we have not gotten to this so far,
> but
> > it is one feature still in the list...
> >
> > Greetings,
> > Stephan
> >
> >
> > On Mon, May 30, 2016 at 9:55 PM, Eron Wright <ewright@live.com> wrote:
> >
> > > Thinking out loud now=E2=80=A6
> > >
> > > Is the job graph fully mutable?   Can it be cleared?   For example,
> > > shouldn=E2=80=99t the count method remove the sink after execution co=
mpletes?
> > >
> > > Can numerous job graphs co-exist within a single driver program?    H=
ow
> > > would that relate to the session concept?
> > >
> > > Seems the count method should use =E2=80=98backtracking=E2=80=99 sche=
dule mode, and
> only
> > > execute the minimum needed to materialize the count sink.
> > >
> > > > On May 29, 2016, at 3:08 PM, M=C3=A1rton Balassi <
> balassi.marton@gmail.com>
> > > wrote:
> > > >
> > > > Hey Eron,
> > > >
> > > > Yes, DataSet#collect and count methods implicitly trigger a JobGrap=
h
> > > > execution, thus they also trigger writing to any previously defined
> > > sinks.
> > > > The idea behind this behavior is to enable interactive querying (th=
e
> > one
> > > > that you are used to get from a shell environment) and it is also a
> > great
> > > > debugging tool.
> > > >
> > > > Best,
> > > >
> > > > Marton
> > > >
> > > > On Sun, May 29, 2016 at 11:28 PM, Eron Wright <ewright@live.com>
> > wrote:
> > > >
> > > >> I was curious as to how the `count` method on DataSet worked, and
> was
> > > >> surprised to see that it executes the entire program graph.
>  Wouldn=E2=80=99t
> > > this
> > > >> cause undesirable side-effects like writing to sinks?    Also
> strange
> > > that
> > > >> the graph is mutated with the addition of a sink (that isn=E2=80=
=99t
> > > subsequently
> > > >> removed).
> > > >>
> > > >> Surveying the Flink code, there aren=E2=80=99t many situations whe=
re the
> > program
> > > >> graph is implicitly executed (`collect` is another).   Nonetheless=
,
> > this
> > > >> has deepened my appreciation for how dynamic the application might
> be.
> > > >>
> > > >> // DataSet.java
> > > >> public long count() throws Exception {
> > > >>   final String id =3D new AbstractID().toString();
> > > >>
> > > >>   output(new Utils.CountHelper<T>(id)).name("count()");
> > > >>
> > > >>   JobExecutionResult res =3D getExecutionEnvironment().execute();
> > > >>   return res.<Long> getAccumulatorResult(id);
> > > >> }
> > > >> Eron
> > >
> > >
> >
>

--001a1144b952686d0d05341ffa53--