Re: Flink streaming questions

Posted by Jamie Grier on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Flink-streaming-questions-tp10808p10816.html

Hi Henri,

#1 - This is by design.  Event time advances with the slowest input source.  If there are input sources that generate no data this is indistinguishable from a slow source.  Kafka topics where some partitions receive no data are a problem in this regard -- but there isn't a simple solution.  If possible I would address it at the source.

#2 - If it's possible to run these init functions just once when you submit the job you can run them in the constructor of your FoldFunction.  This init will then happen exactly once (on the client) and the constructed FoldFunction is then serialized and distributed around the cluster.  If this doesn't work because you need something truly dynamic you could also accomplish this with a simple local variable in your function.

class MyFoldFunction extends FoldFunction {
  private var initialized = false
  def fold(accumulator: T, value: O): T = {
    if(!initialized){
      doInitStuff()
      initialized = true
    }
    
    doNormalStuff()
  }
}

#3 - One way to do this is as you've said which is to attach the profile information to the event, using a mapper, before it enters the window operations.

On Mon, Jan 2, 2017 at 1:25 AM, Henri Heiskanen <[hidden email]> wrote:

Hi,

I have few questions related to Flink streaming. I am on 1.2-SNAPSHOT and what I would like to accomplish is to have a stream that reads data from multiple kafka topics, identifies user sessions, uses an external user user profile to enrich the data, evaluates an script to produce session aggregates and then create updated profiles from session aggregates. I am working with high volume data and user sessions may be long, so using generic window apply might not work. Below is the simplification of the stream.

stream = createKafkaStreams(...);

env.setParallelism(4);

env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

stream

                .keyBy(2)

                .window(EventTimeSessionWindows.withGap(Time.minutes(10)))

                .fold(new SessionData(), new SessionFold(), new ProfilerApply())

                .print();

The questions:

1. Initially when I used event time windowing I could not get any of my windows to close. The reason seemed to be that I had 6 partitions in my test kafka setup and only 4 of them generated traffic. If I used parallelism above 4, then no windows were closed. Is this by design or a defect? We use flink-connector-kafka-0.10 because earlier versions did not commit the offsets correctly.

2. Rich fold functions are not supported. However I would like execute a piece of custom script in the fold function that requires initialisation part. I would have used the open and close lifecycle methods of rich functions but they are not available now in fold. What would be the preferred way to run some initialisation routines (and closing the gracefully) when using fold?

3. Kind of related to above. I would also like to fetch a user profile from external source in the beginning of the session. What would be a best practice for that kind of operation? If I would be using the generic window apply I could fetch in in the beginning of the apply method. I was thinking of introducing a mapper that fetches this profiler periodically and caches it to flink state. However, with this setup I would not be able to tie this to user sessions identified for windows.

4. I also may have an additional requirement of writing out each event enriched with current session and profile data. I basically could do this again with generic window function and write out each event with collector when iterating, but would there be a better pattern to use? Maybe sharing state with functions or something.

Br,

Henri H

--