Thanks Roland a lot for the clarification .. I can do this ..
val netTxn: RunnableGraph[Future[akka.Done]] =
transactions.map(validate)
.groupBy(MaxGroupCount, _.accountNo)
.fold(TransactionMonoid.zero)(_ |+| _)
.mergeSubstreams
Only difference with the 1.0 API is that I cannot do the fold in the Sink -
instead I have to do it here. I don't mind, it may even be better this way.
Just one clarification .. In the above snippet are the folds on individual
substreams done in parallel ?
Thanks again for the help ..
Regards.
On Wed, Apr 27, 2016 at 2:53 AM, Roland Kuhn <goo...@rkuhn.info> wrote:
>
> 26 apr 2016 kl. 20:14 skrev Debasish Ghosh <ghosh.debas...@gmail.com>:
>
> Just for the sake of completeness, this works ..
>
> transactions.map(validate)
> .groupBy(MaxGroupCount, _.accountNo)
> .fold(Map.empty[String, Transaction])((l, r) => l |+|
> Map(r.accountNo -> r))
> .mergeSubstreams
>
>
> Yes, this is what I proposed a while ago :-) (more or less, see below)
>
> Since I cannot access the sub-streams I need to simulate this through a
> Map ..
>
>
> I don’t fully understand: the fold is running on each of the substreams,
> there is nothing that is inaccessible. The only difference to the old API
> is that the key is not available in pre-extracted form (i.e. things are not
> wrapped in tuples). You also don’t need to use a Map, you could also keep
> the last r.accountNo in the first position of a tuple and the current
> Transaction in the second—the result then looks exactly the same as it did
> with streams 1.0. OTOH I suspect that your monoid leaves the accountNo
> intact, so there is not even a need to allocate those tuples.
>
> Would you please explain how our groupBy loses the semantics from SQL? I’m
> a bit rusty, but the group “name” is also not made available in a dedicated
> fashion there, AFAIR. The current shape of the API was inspired by BigData
> analytics tools that model this operation in the same way, because the
> previous scheme was simply too powerful (it allowed arbitrary dynamic
> treatment of each of the substreams—SQL grouping is also homogenous and not
> different for each group). Dynamic elements also defeat optimizations like
> fusing and graph simplification.
>
> Regards,
>
> Roland
>
>
> Thanks.
>
> On Tue, Apr 26, 2016 at 11:32 PM, Debasish Ghosh <ghosh.debas...@gmail.com
> > wrote:
>
>> Thanks Viktor .. my main issue is that we lose the natural semantics of
>> groupBy as we learnt from SQL. We cannot access the substreams as separate
>> abstractions that groupBy creates.
>>
>> On Tue, Apr 26, 2016 at 11:28 PM, Viktor Klang <viktor.kl...@gmail.com>
>> wrote:
>>
>>> Debasish,
>>>
>>> The problem with groupBy was that it was too easy to create leaks and
>>> silently broken solutions with it, but I sympathize with your situation, I
>>> also felt the original as being more ergonomic. I think there is room for
>>> improvement in the current solution.
>>> --
>>> Cheers,
>>> √
>>> On Apr 26, 2016 7:20 PM, "Debasish Ghosh" <ghosh.debas...@gmail.com>
>>> wrote:
>>>
>>> Roland -
>>>
>>> The problem is I cannot access the substreams separately within the
>>> fold. In the earlier solution the fold was on individual substreams (not on
>>> the whole stream). Here the fold gets *all* the elements from all
>>> substreams and I lose the ability to process substreams separately. Hence I
>>> lose the ability to compose with a Monoid. Possibly I need to create a
>>> separate Map and then do the aggregation. It will be clunky .. I am already
>>> starting to feel the loss of the 1.0 API, which I thought was very
>>> idiomatic from groupBy point of view.
>>>
>>> Let me see if I can at all get a solution for this ..
>>>
>>> regards.
>>>
>>> On Tue, Apr 26, 2016 at 10:33 PM, Roland Kuhn <goo...@rkuhn.info> wrote:
>>>
>>>> Instead of using a fold sink which materializes to a Future you’ll need
>>>> to use a fold combinator which produces the result as its only value after
>>>> the substream has completed, i.e. you keep the computation results within
>>>> the streaming domain a bit longer instead of going to the Future domain
>>>> immediately. Your 1.0 solution must have done something very similar, only
>>>> with a different API (i.e. you folded the substreams and then probably used
>>>> mapAsync to flatten the resulting stream of Futures).
>>>>
>>>> Regards,
>>>>
>>>> Roland
>>>>
>>>> 26 apr 2016 kl. 18:05 skrev Debasish Ghosh <ghosh.debas...@gmail.com>:
>>>>
>>>> Roland -
>>>>
>>>> I need to merge into a fold sink w/ a Monoid. Consider the following
>>>> example ..
>>>>
>>>> case class Transaction(id: String, accountNo: String, debitCredit:
>>>> TransactionType, amount: Amount, date: Date = today)
>>>>
>>>> and I have the following list of Transactions ..
>>>>
>>>> val txns =
>>>> Seq(
>>>> Transaction("t-1", "a-1", Debit, 1000),
>>>> Transaction("t-2", "a-2", Debit, 1000),
>>>> Transaction("t-3", "a-3", Credit, 1000),
>>>> Transaction("t-4", "a-1", Credit, 1000),
>>>> Transaction("t-5", "a-1", Debit, 1000),
>>>> Transaction("t-6", "a-2", Debit, 1000),
>>>> Transaction("t-7", "a-3", Credit, 1000),
>>>> Transaction("t-8", "a-3", Debit, 1000),
>>>> Transaction("t-9", "a-2", Credit, 1000),
>>>> Transaction("t-10", "a-2", Debit, 1000),
>>>> Transaction("t-11", "a-1", Credit, 1000),
>>>> Transaction("t-12", "a-3", Debit, 1000)
>>>> )
>>>>
>>>> I do a group by accountNo, which gives me 3 substreams. Each substream
>>>> needs to go into a fold sink where I fold using a Monoid. The Monoid has
>>>> the logic of merging transactions belonging to the same accountNo. The
>>>> logic that u suggest will not do this. I need to access the substreams
>>>> separately. That's why I did a groupBy (similar to SQL groupBy). And I
>>>> could do this in the 1.0 version.
>>>>
>>>> Any workaround that u suggest ?
>>>>
>>>> Thanks.
>>>>
>>>> On Tue, Apr 26, 2016 at 9:18 PM, Roland Kuhn <goo...@rkuhn.info> wrote:
>>>>
>>>>> If you need the results from the substreams you’ll have to merge them
>>>>> back into the mainstream and aggregate them there:
>>>>>
>>>>> transactions.groupBy(100,
>>>>> ...).fold(...).mergeSubstreams.grouped(100).to(Sink.head)
>>>>>
>>>>> Regards,
>>>>>
>>>>> Roland
>>>>>
>>>>> 26 apr 2016 kl. 17:36 skrev debasish <ghosh.debas...@gmail.com>:
>>>>>
>>>>> Viktor -
>>>>>
>>>>> Here's the same stuff that works for akka-streams version
>>>>> akka-stream-experimental 1.0 RC4. ..
>>>>> https://gist.github.com/debasishg/4d596c1f26d4759ed65e281bb2e6fd2c ..
>>>>>
>>>>> The upgrade that I am having trouble with is defining netTxn (pls see
>>>>> the gist). The groupBy works like a charm in the older version. But in the
>>>>> newer version it returns a Subflow and I was stumped how to get hold
>>>>> of each of the substreams and fold them using a Monoid on a fold
>>>>> sink. Konrad's solution was to use the to method of subflow. But
>>>>> somehow it's not giving the desired result .. help ?
>>>>>
>>>>> Thanks.
>>>>>
>>>>>
>>>>> On Tuesday, April 26, 2016 at 8:46:11 PM UTC+5:30, √ wrote:
>>>>>>
>>>>>> What are you expecting to be returned?
>>>>>>
>>>>>> --
>>>>>> Cheers,
>>>>>> √
>>>>>> On Apr 26, 2016 3:49 PM, "Debasish Ghosh" <ghosh.d...@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> Thanks Konrad for the pointer .. when I run the graph I get a
>>>>>>> NotUsed .. That's not hwat I get with the earlier implementation.
>>>>>>> Please have a look at the gist ..
>>>>>>> https://gist.github.com/debasishg/a42e867bb2bc8ad18243597178bbce93 ..
>>>>>>> what am I doing wrong.
>>>>>>>
>>>>>>> Thanks.
>>>>>>>
>>>>>>> On Tue, Apr 26, 2016 at 6:22 PM, Konrad Malawski <
>>>>>>> konrad....@typesafe.com> wrote:
>>>>>>>
>>>>>>>> (I did a quick mock Transaction type, your Monoid should work fine
>>>>>>>> there ofc).
>>>>>>>>
>>>>>>>> -- Konrad
>>>>>>>>
>>>>>>>> On Sunday, 24 April 2016 21:42:43 UTC+2, debasish wrote:
>>>>>>>>
>>>>>>>>> Hi -
>>>>>>>>>
>>>>>>>>> I am trying to migrate some akka-streams code from an earlier
>>>>>>>>> version (akka-stream-experimental 1.0.RC4) .. please have a look at
>>>>>>>>> the
>>>>>>>>> following ..
>>>>>>>>>
>>>>>>>>> /**
>>>>>>>>> * Create multiple streams out of a single stream. The stream
>>>>>>>>> "transactions" is being
>>>>>>>>> * demultiplexed into many streams split by account number. Each
>>>>>>>>> of the sub-streams are
>>>>>>>>> * then materialized to the fold sink "txnSink", which folds each
>>>>>>>>> of the transaction
>>>>>>>>> * substreams to compute the net value of the transaction for that
>>>>>>>>> account
>>>>>>>>> */
>>>>>>>>> val netTxn: Source[RunnableGraph[Future[Transaction]], Unit] =
>>>>>>>>> transactions.map(validate).groupBy(_.accountNo).map { case (a,
>>>>>>>>> s) => s.toMat(txnSink)(Keep.right) }
>>>>>>>>>
>>>>>>>>> I am trying to create substreams based on a field accountNo within
>>>>>>>>> a Transaction. Then I pass each substream to toMat with a Sink
>>>>>>>>> which is a fold Sink defined as below ..
>>>>>>>>>
>>>>>>>>> /**
>>>>>>>>> * Would like to fold transactions through monoid append
>>>>>>>>> */
>>>>>>>>> val txnSink: Sink[Transaction, Future[Transaction]] =
>>>>>>>>> Sink.fold[Transaction, Transaction](TransactionMonoid.zero)(_
>>>>>>>>> |+| _)
>>>>>>>>>
>>>>>>>>> The idea is to apply a monoid over each substream and do some
>>>>>>>>> netting on transactions belonging to the same accountNo.
>>>>>>>>>
>>>>>>>>> How can I migrate this to 2.4.4 ? Now groupBy returns a Subflow and
>>>>>>>>> I am not sure how to apply each of the substreams to the
>>>>>>>>> materializer. One
>>>>>>>>> option that works will be to create a Map[String, Transaction] upfront
>>>>>>>>> and then directly apply the monoid to the Map in the main stream
>>>>>>>>> and not create any substream. But I would like to have the substreams
>>>>>>>>> fro
>>>>>>>>> some other purpose as well.
>>>>>>>>>
>>>>>>>>> Any help will be appreciated ..
>>>>>>>>>
>>>>>>>>> Regards.
>>>>>>>>> - Debasish
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>>>
>>>>>>>> --
>>>>>>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>> >>>>>>>>>> Check the FAQ:
>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>> >>>>>>>>>> Search the archives:
>>>>>>>> https://groups.google.com/group/akka-user
>>>>>>>> ---
>>>>>>>> You received this message because you are subscribed to a topic in
>>>>>>>> the Google Groups "Akka User List" group.
>>>>>>>> To unsubscribe from this topic, visit
>>>>>>>> https://groups.google.com/d/topic/akka-user/YhJfghLqnhw/unsubscribe
>>>>>>>> .
>>>>>>>> To unsubscribe from this group and all its topics, send an email to
>>>>>>>> akka-user+...@googlegroups.com.
>>>>>>>> To post to this group, send email to akka...@googlegroups.com.
>>>>>>>> Visit this group at https://groups.google.com/group/akka-user.
>>>>>>>> For more options, visit https://groups.google.com/d/optout.
>>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>>
>>>>>>> --
>>>>>>> Debasish Ghosh
>>>>>>> http://manning.com/ghosh2
>>>>>>> http://manning.com/ghosh
>>>>>>>
>>>>>>> Twttr: @debasishg
>>>>>>> Blog: http://debasishg.blogspot.com
>>>>>>> Code: http://github.com/debasishg
>>>>>>>
>>>>>>> --
>>>>>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>> >>>>>>>>>> Check the FAQ:
>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>> >>>>>>>>>> Search the archives:
>>>>>>> https://groups.google.com/group/akka-user
>>>>>>> ---
>>>>>>> You received this message because you are subscribed to the Google
>>>>>>> Groups "Akka User List" group.
>>>>>>> To unsubscribe from this group and stop receiving emails from it,
>>>>>>> send an email to akka-user+...@googlegroups.com.
>>>>>>> To post to this group, send email to akka...@googlegroups.com.
>>>>>>> Visit this group at https://groups.google.com/group/akka-user.
>>>>>>> For more options, visit https://groups.google.com/d/optout.
>>>>>>>
>>>>>>
>>>>> --
>>>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>> >>>>>>>>>> Check the FAQ:
>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>> >>>>>>>>>> Search the archives:
>>>>> https://groups.google.com/group/akka-user
>>>>> ---
>>>>> You received this message because you are subscribed to the Google
>>>>> Groups "Akka User List" group.
>>>>> To unsubscribe from this group and stop receiving emails from it, send
>>>>> an email to akka-user+unsubscr...@googlegroups.com.
>>>>> To post to this group, send email to akka-user@googlegroups.com.
>>>>> Visit this group at https://groups.google.com/group/akka-user.
>>>>> For more options, visit https://groups.google.com/d/optout.
>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>> >>>>>>>>>> Check the FAQ:
>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>> >>>>>>>>>> Search the archives:
>>>>> https://groups.google.com/group/akka-user
>>>>> ---
>>>>> You received this message because you are subscribed to a topic in the
>>>>> Google Groups "Akka User List" group.
>>>>> To unsubscribe from this topic, visit
>>>>> https://groups.google.com/d/topic/akka-user/YhJfghLqnhw/unsubscribe.
>>>>> To unsubscribe from this group and all its topics, send an email to
>>>>> akka-user+unsubscr...@googlegroups.com.
>>>>> To post to this group, send email to akka-user@googlegroups.com.
>>>>> Visit this group at https://groups.google.com/group/akka-user.
>>>>> For more options, visit https://groups.google.com/d/optout.
>>>>>
>>>>
>>>>
>>>>
>>>> --
>>>> Debasish Ghosh
>>>> http://manning.com/ghosh2
>>>> http://manning.com/ghosh
>>>>
>>>> Twttr: @debasishg
>>>> Blog: http://debasishg.blogspot.com
>>>> Code: http://github.com/debasishg
>>>>
>>>> --
>>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>>> >>>>>>>>>> Check the FAQ:
>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>> >>>>>>>>>> Search the archives:
>>>> https://groups.google.com/group/akka-user
>>>> ---
>>>> You received this message because you are subscribed to the Google
>>>> Groups "Akka User List" group.
>>>> To unsubscribe from this group and stop receiving emails from it, send
>>>> an email to akka-user+unsubscr...@googlegroups.com.
>>>> To post to this group, send email to akka-user@googlegroups.com.
>>>> Visit this group at https://groups.google.com/group/akka-user.
>>>> For more options, visit https://groups.google.com/d/optout.
>>>>
>>>>
>>>>
>>>> --
>>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>>> >>>>>>>>>> Check the FAQ:
>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>> >>>>>>>>>> Search the archives:
>>>> https://groups.google.com/group/akka-user
>>>> ---
>>>> You received this message because you are subscribed to a topic in the
>>>> Google Groups "Akka User List" group.
>>>> To unsubscribe from this topic, visit
>>>> https://groups.google.com/d/topic/akka-user/YhJfghLqnhw/unsubscribe.
>>>> To unsubscribe from this group and all its topics, send an email to
>>>> akka-user+unsubscr...@googlegroups.com.
>>>> To post to this group, send email to akka-user@googlegroups.com.
>>>> Visit this group at https://groups.google.com/group/akka-user.
>>>> For more options, visit https://groups.google.com/d/optout.
>>>>
>>>
>>>
>>>
>>> --
>>> Debasish Ghosh
>>> http://manning.com/ghosh2
>>> http://manning.com/ghosh
>>>
>>> Twttr: @debasishg
>>> Blog: http://debasishg.blogspot.com
>>> Code: http://github.com/debasishg
>>>
>>> --
>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>> >>>>>>>>>> Check the FAQ:
>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>> >>>>>>>>>> Search the archives:
>>> https://groups.google.com/group/akka-user
>>> ---
>>> You received this message because you are subscribed to the Google
>>> Groups "Akka User List" group.
>>> To unsubscribe from this group and stop receiving emails from it, send
>>> an email to akka-user+unsubscr...@googlegroups.com.
>>> To post to this group, send email to akka-user@googlegroups.com.
>>> Visit this group at https://groups.google.com/group/akka-user.
>>> For more options, visit https://groups.google.com/d/optout.
>>>
>>>
>>> --
>>> >>>>>>>>>> Read the docs: http://akka.io/docs/
>>> >>>>>>>>>> Check the FAQ:
>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>> >>>>>>>>>> Search the archives:
>>> https://groups.google.com/group/akka-user
>>> ---
>>> You received this message because you are subscribed to a topic in the
>>> Google Groups "Akka User List" group.
>>> To unsubscribe from this topic, visit
>>> https://groups.google.com/d/topic/akka-user/YhJfghLqnhw/unsubscribe.
>>> To unsubscribe from this group and all its topics, send an email to
>>> akka-user+unsubscr...@googlegroups.com.
>>> To post to this group, send email to akka-user@googlegroups.com.
>>> Visit this group at https://groups.google.com/group/akka-user.
>>> For more options, visit https://groups.google.com/d/optout.
>>>
>>
>>
>>
>> --
>> Debasish Ghosh
>> http://manning.com/ghosh2
>> http://manning.com/ghosh
>>
>> Twttr: @debasishg
>> Blog: http://debasishg.blogspot.com
>> Code: http://github.com/debasishg
>>
>
>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to the Google Groups
> "Akka User List" group.
> To unsubscribe from this group and stop receiving emails from it, send an
> email to akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
>
> --
> >>>>>>>>>> Read the docs: http://akka.io/docs/
> >>>>>>>>>> Check the FAQ:
> http://doc.akka.io/docs/akka/current/additional/faq.html
> >>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
> ---
> You received this message because you are subscribed to a topic in the
> Google Groups "Akka User List" group.
> To unsubscribe from this topic, visit
> https://groups.google.com/d/topic/akka-user/YhJfghLqnhw/unsubscribe.
> To unsubscribe from this group and all its topics, send an email to
> akka-user+unsubscr...@googlegroups.com.
> To post to this group, send email to akka-user@googlegroups.com.
> Visit this group at https://groups.google.com/group/akka-user.
> For more options, visit https://groups.google.com/d/optout.
>
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
--
>>>>>>>>>> Read the docs: http://akka.io/docs/
>>>>>>>>>> Check the FAQ:
>>>>>>>>>> http://doc.akka.io/docs/akka/current/additional/faq.html
>>>>>>>>>> Search the archives: https://groups.google.com/group/akka-user
---
You received this message because you are subscribed to the Google Groups "Akka
User List" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to akka-user+unsubscr...@googlegroups.com.
To post to this group, send email to akka-user@googlegroups.com.
Visit this group at https://groups.google.com/group/akka-user.
For more options, visit https://groups.google.com/d/optout.
