Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@flink.apache.org
Message-ID: <55D72B5B.7050702@tu-ilmenau.de>
Date: Fri, 21 Aug 2015 15:44:59 +0200
From: Philipp Goetze <philipp.goetze@tu-ilmenau.de>
User-Agent: Mozilla/5.0 (X11; Linux x86_64;
 rv:31.0) Gecko/20100101 Thunderbird/31.8.0
MIME-Version: 1.0
To: user@flink.apache.org
Subject: Re: Custom Aggregate - Example
References: <55D71EBF.4080006@tu-ilmenau.de>
 <CANMXwW1qsTVSrVAVg8HAJUZ1kL8VyhhHjbNCE+g0gWjirjmx+Q@mail.gmail.com>
In-Reply-To: 
 <CANMXwW1qsTVSrVAVg8HAJUZ1kL8VyhhHjbNCE+g0gWjirjmx+Q@mail.gmail.com>
Content-Type: multipart/alternative;
 boundary="------------020002080100070000050600"

This is a multi-part message in MIME format.
--------------020002080100070000050600
Content-Type: text/plain; charset=utf-8; format=flowed
Content-Transfer-Encoding: 7bit

Thank you Aljoscha,

I guessed that I should use the reduce method. However, I do not look 
for window aggregations. I want to do this on a grouped stream.

The problem is we work with Lists instead of tuples and thus we can not 
use the pre-implemented aggregates.

So the idea is to call it like that:

    val aggr = source.groupBy(_(0)).reduce(new customReducer(1))

And this is the signature of the class:

    class customReducer(field: Int) extends RichReduceFunction[List[Any]]


How do I have to implement this class now, so that it is working 
correctly even with parallelism > 1?

I hope you understand what I try to do. =)

Kind Regards,
Philipp


On 21.08.2015 15:28, Aljoscha Krettek wrote:
> Hi,
> with the current API this should do what you are after:
>
> val input = ...
> val result = input
>   .window(...)
>   .groupBy(...)
>   .reduceWindow( /* your reduce function */ )
>
> With the reduce function you should be able to implement any custom 
> aggregations. You can also use foldWindow() if you want to do a 
> functional fold over the window.
>
> I hope this helps.
>
> Cheers,
> Aljoscha
>
> On Fri, 21 Aug 2015 at 14:51 Philipp Goetze 
> <philipp.goetze@tu-ilmenau.de <mailto:philipp.goetze@tu-ilmenau.de>> 
> wrote:
>
>     Hello community,
>
>     how do I define a custom aggregate function in Flink Streaming
>     (Scala)?
>     Could you please provide an example on how to do that?
>
>     Thank you and best regards,
>     Philipp
>


--------------020002080100070000050600
Content-Type: text/html; charset=utf-8
Content-Transfer-Encoding: 8bit

<html>
  <head>
    <meta content="text/html; charset=utf-8" http-equiv="Content-Type">
  </head>
  <body bgcolor="#FFFFFF" text="#000000">
    Thank you Aljoscha,<br>
    <br>
    I guessed that I should use the reduce method. However, I do not
    look for window aggregations. I want to do this on a grouped stream.
    <br>
    <br>
    The problem is we work with Lists instead of tuples and thus we can
    not use the pre-implemented aggregates.<br>
    <br>
    So the idea is to call it like that:<br>
    <blockquote><tt>val aggr = source.groupBy(_(0)).reduce(new
        customReducer(1))</tt><br>
    </blockquote>
    And this is the signature of the class:<br>
    <blockquote><tt>class customReducer(field: Int) extends
        RichReduceFunction[List[Any]] </tt><br>
    </blockquote>
    <br>
    How do I have to implement this class now, so that it is working
    correctly even with parallelism &gt; 1?<br>
    <br>
    I hope you understand what I try to do. =)<br>
    <br>
    Kind Regards,<br>
    Philipp<br>
    <br>
    <br>
    <div class="moz-cite-prefix">On 21.08.2015 15:28, Aljoscha Krettek
      wrote:<br>
    </div>
    <blockquote
cite="mid:CANMXwW1qsTVSrVAVg8HAJUZ1kL8VyhhHjbNCE+g0gWjirjmx+Q@mail.gmail.com"
      type="cite">
      <div dir="ltr">Hi,
        <div>with the current API this should do what you are after:</div>
        <div><br>
        </div>
        <div>val input = ...</div>
        <div> </div>
        <div>val result = input</div>
        <div>  .window(...)</div>
        <div>  .groupBy(...)</div>
        <div>  .reduceWindow( /* your reduce function */ )</div>
        <div><br>
        </div>
        <div>With the reduce function you should be able to implement
          any custom aggregations. You can also use foldWindow() if you
          want to do a functional fold over the window.</div>
        <div><br>
        </div>
        <div>I hope this helps.</div>
        <div><br>
        </div>
        <div>Cheers,</div>
        <div>Aljoscha </div>
      </div>
      <br>
      <div class="gmail_quote">
        <div dir="ltr">On Fri, 21 Aug 2015 at 14:51 Philipp Goetze &lt;<a
            moz-do-not-send="true"
            href="mailto:philipp.goetze@tu-ilmenau.de">philipp.goetze@tu-ilmenau.de</a>&gt;
          wrote:<br>
        </div>
        <blockquote class="gmail_quote" style="margin:0 0 0
          .8ex;border-left:1px #ccc solid;padding-left:1ex">Hello
          community,<br>
          <br>
          how do I define a custom aggregate function in Flink Streaming
          (Scala)?<br>
          Could you please provide an example on how to do that?<br>
          <br>
          Thank you and best regards,<br>
          Philipp<br>
        </blockquote>
      </div>
    </blockquote>
    <br>
  </body>
</html>

--------------020002080100070000050600--