Thanks Chiwan.
I think this example still creates a lazy-evaluated plan. And if i need to
collect statistics to front end (and use it in subsequent iteration
evaluation) as my example with computing column-wise averages suggests?
problem generally is, what if I need to eagerly evaluate the statistics
inside the iteration in order to proceed with further computations (and
even plan construction). typically, that would be result of M-step in EM
algorithm.
On Sun, Mar 27, 2016 at 3:26 AM, Chiwan Park <chiwanp...@apache.org> wrote:
> Hi Dmitriy,
>
> I think you can implement it with iterative API with custom convergence
> criterion. You can express the convergence criterion by two methods. One is
> using a convergence criterion data set [1][2] and the other is registering
> an aggregator with custom implementation of `ConvergenceCriterion`
> interface [3].
>
> Here is an example using a convergence criterion data set in Scala API:
>
> ```
> package flink.sample
>
> import org.apache.flink.api.scala._
>
> import scala.util.Random
>
> object SampleApp extends App {
> val env = ExecutionEnvironment.getExecutionEnvironment
>
> val data = env.fromElements[Double](1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
>
> val result = data.iterateWithTermination(5000) { prev =>
> // calculate sub solution
> val rand = Random.nextDouble()
> val subSolution = prev.map(_ * rand)
>
> // calculate convergent condition
> val convergence = subSolution.reduce(_ + _).map(_ / 10).filter(_ > 8)
>
> (subSolution, convergence)
> }
>
> result.print()
> }
> ```
>
> Regards,
> Chiwan Park
>
> [1]:
> https://ci.apache.org/projects/flink/flink-docs-release-1.0/api/java/org/apache/flink/api/java/operators/IterativeDataSet.html#closeWith%28org.apache.flink.api.java.DataSet,%20org.apache.flink.api.java.DataSet%29
> [2]: iterateWithTermination method in
> https://ci.apache.org/projects/flink/flink-docs-release-1.0/api/scala/index.html#org.apache.flink.api.scala.DataSet
> [3]:
> https://ci.apache.org/projects/flink/flink-docs-release-1.0/api/java/org/apache/flink/api/java/operators/IterativeDataSet.html#registerAggregationConvergenceCriterion%28java.lang.String,%20org.apache.flink.api.common.aggregators.Aggregator,%20org.apache.flink.api.common.aggregators.ConvergenceCriterion%29
>
> > On Mar 26, 2016, at 2:51 AM, Dmitriy Lyubimov <dlie...@gmail.com> wrote:
> >
> > Thank you, all :)
> >
> > yes, that's my question. How do we construct such a loop with a concrete
> > example?
> >
> > Let's take something nonsensical yet specific.
> >
> > Say, in samsara terms we do something like that :
> >
> > var avg = Double.PositiveInfinity
> > var drmA = ... (construct elsewhere)
> >
> >
> >
> > do {
> > avg = drmA.colMeans.mean // average of col-wise means
> > drmA = drmA - avg // elementwise subtract of average
> >
> > } while (avg > 1e-10)
> >
> > (which probably does not converge in reality).
> >
> > How would we implement that with native iterations in flink?
> >
> >
> >
> > On Wed, Mar 23, 2016 at 2:50 AM, Till Rohrmann <trohrm...@apache.org>
> wrote:
> >
> >> Hi Dmitriy,
> >>
> >> I’m not sure whether I’ve understood your question correctly, so please
> >> correct me if I’m wrong.
> >>
> >> So you’re asking whether it is a problem that
> >>
> >> stat1 = A.map.reduce
> >> A = A.update.map(stat1)
> >>
> >> are executed on the same input data set A and whether we have to cache A
> >> for that, right? I assume you’re worried that A is calculated twice.
> >>
> >> Since you don’t have a API call which triggers eager execution of the
> data
> >> flow, the map.reduce and map(stat1) call will only construct the data
> flow
> >> of your program. Both operators will depend on the result of A which is
> >> only once calculated (when execute, collect or count is called) and then
> >> sent to the map.reduce and map(stat1) operator.
> >>
> >> However, it is not recommended using an explicit loop to do iterative
> >> computations with Flink. The problem here is that you will basically
> unroll
> >> the loop and construct a long pipeline with the operations of each
> >> iterations. Once you execute this long pipeline you will face
> considerable
> >> memory fragmentation, because every operator will get a proportional
> >> fraction of the available memory assigned. Even worse, if you trigger
> the
> >> execution of your data flow to evaluate the convergence criterion, you
> will
> >> execute for each iteration the complete pipeline which has been built
> up so
> >> far. Thus, you’ll end up with a quadratic complexity in the number of
> >> iterations. Therefore, I would highly recommend using Flink’s built in
> >> support for native iterations which won’t suffer from this problem or to
> >> materialize at least for every n iterations the intermediate result. At
> the
> >> moment this would mean to write the data to some sink and then reading
> it
> >> from there again.
> >>
> >> I hope this answers your question. If not, then don’t hesitate to ask me
> >> again.
> >>
> >> Cheers,
> >> Till
> >>
> >>
> >> On Wed, Mar 23, 2016 at 10:19 AM, Theodore Vasiloudis <
> >> theodoros.vasilou...@gmail.com> wrote:
> >>
> >>> Hello Dmitriy,
> >>>
> >>> If I understood correctly what you are basically talking about
> modifying
> >> a
> >>> DataSet as you iterate over it.
> >>>
> >>> AFAIK this is currently not possible in Flink, and indeed it's a real
> >>> bottleneck for ML algorithms. This is the reason our current
> >>> SGD implementation does a pass over the whole dataset at each
> iteration,
> >>> since we cannot take a sample from the dataset
> >>> and iterate only over that (so it's not really stochastic).
> >>>
> >>> The relevant JIRA is here:
> >>> https://issues.apache.org/jira/browse/FLINK-2396
> >>>
> >>> I would love to start a discussion on how we can proceed to fix this.
> >>>
> >>> Regards,
> >>> Theodore
> >>>
> >>> On Tue, Mar 22, 2016 at 9:56 PM, Dmitriy Lyubimov <dlie...@gmail.com>
> >>> wrote:
> >>>
> >>>> Hi,
> >>>>
> >>>> probably more of a question for Till:
> >>>>
> >>>> Imagine a common ML algorithm flow that runs until convergence.
> >>>>
> >>>> typical distributed flow would be something like that (e.g. GMM EM
> >> would
> >>> be
> >>>> exactly like that):
> >>>>
> >>>> A: input
> >>>>
> >>>> do {
> >>>>
> >>>> stat1 = A.map.reduce
> >>>> A = A.update-map(stat1)
> >>>> conv = A.map.reduce
> >>>> } until conv > convThreshold
> >>>>
> >>>> There probably could be 1 map-reduce step originating on A to compute
> >>> both
> >>>> convergence criteria statistics and udpate statistics in one step. not
> >>> the
> >>>> point.
> >>>>
> >>>> The point is that update and map.reduce originate on the same dataset
> >>>> intermittently.
> >>>>
> >>>> In spark we would normally commit A to a object tree cache so that
> data
> >>> is
> >>>> available to subsequent map passes without any I/O or serialization
> >>>> operations, thus insuring high rate of iterations.
> >>>>
> >>>> We observe the same pattern pretty much everywhere. clustering,
> >>>> probabilistic algorithms, even batch gradient descent of quasi newton
> >>>> algorithms fitting.
> >>>>
> >>>> How do we do something like that, for example, in FlinkML?
> >>>>
> >>>> Thoughts?
> >>>>
> >>>> thanks.
> >>>>
> >>>> -Dmitriy
> >>>>
> >>>
> >>
>
>
