Return-Path: X-Original-To: apmail-flink-user-archive@minotaur.apache.org Delivered-To: apmail-flink-user-archive@minotaur.apache.org Received: from mail.apache.org (hermes.apache.org [140.211.11.3]) by minotaur.apache.org (Postfix) with SMTP id A768519741 for ; Thu, 7 Apr 2016 16:40:20 +0000 (UTC) Received: (qmail 48760 invoked by uid 500); 7 Apr 2016 16:40:20 -0000 Delivered-To: apmail-flink-user-archive@flink.apache.org Received: (qmail 48670 invoked by uid 500); 7 Apr 2016 16:40:20 -0000 Mailing-List: contact user-help@flink.apache.org; run by ezmlm Precedence: bulk List-Help: List-Unsubscribe: List-Post: List-Id: Reply-To: user@flink.apache.org Delivered-To: mailing list user@flink.apache.org Received: (qmail 48661 invoked by uid 99); 7 Apr 2016 16:40:20 -0000 Received: from mail-relay.apache.org (HELO mail-relay.apache.org) (140.211.11.15) by apache.org (qpsmtpd/0.29) with ESMTP; Thu, 07 Apr 2016 16:40:20 +0000 Received: from mail-lf0-f47.google.com (mail-lf0-f47.google.com [209.85.215.47]) by mail-relay.apache.org (ASF Mail Server at mail-relay.apache.org) with ESMTPSA id 995991A041D for ; Thu, 7 Apr 2016 16:40:19 +0000 (UTC) Received: by mail-lf0-f47.google.com with SMTP id e190so59428271lfe.0 for ; Thu, 07 Apr 2016 09:40:19 -0700 (PDT) X-Gm-Message-State: AD7BkJJmpYn+osuXaovp3LZBWMYrwo9Mc+ikNy7WGB8GibQ0hkD65OWMIR1EIWjSWmD4i11lT74lbQTkscPPHA== X-Received: by 10.25.144.204 with SMTP id s195mr1840742lfd.120.1460047218238; Thu, 07 Apr 2016 09:40:18 -0700 (PDT) MIME-Version: 1.0 References: <1460036466.4119473.571808673.019EFE9C@webmail.messagingengine.com> In-Reply-To: From: Aljoscha Krettek Date: Thu, 07 Apr 2016 16:40:08 +0000 X-Gmail-Original-Message-ID: Message-ID: Subject: Re: mutable hashmap outside of stream, does it get snapshotted ? To: user@flink.apache.org Content-Type: multipart/alternative; boundary=001a114014facc6e0a052fe7bb87 --001a114014facc6e0a052fe7bb87 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: quoted-printable Hi, good explanation and pointers! I just want to add that the uriLookup table in your example is not really shared between your operator instances in a distributed setting. When serializing your transformations the current state of the HashMap is serialized with them because it is in the closure of the transformations. Then, on the cluster, the HashMap is serialized and all parallel instances basically work on their now local copy of the empty HashMap. Cheers, Aljoscha On Thu, 7 Apr 2016 at 18:30 Stefano Baghino wrote: > Hi Bart, > > to make sure that the state is checkpointed you have to: > > 1. configure your Flink installation with a reliable state backend > (optional for development, you can read more about it here > > ) > 2. explicitly enable checkpointing in your program (see how here > > =E2=80=94 it's just a couple of lines of code) > 3. extend your operators so that they checkpoint data, by implementing > the `Checkpointed` interface or using an instance field (the semantics= of > the two approaches are slightly different, you can read more about it > here > > ) > > When your data is checkpointed you can access the state if your operator > implements the `RichFunction` interface (via an abstract class that wraps > the operator you need to implement, like `RichMapFunction`). > > For your need in particular, I don't know a way to checkpoint state share= d > between different operators; perhaps you can you refactor your code so th= at > the state is encapsulated in an operator implementation and then moved > through your pipeline as a parameter of the following operators. Would th= at > work? > > I apologize for just providing pointers to the docs in my reply but > checkpointing deserves a good explanation and I feel the docs get the job > done pretty well. I will gladly help you if you have any doubt. > > Hope I've been of some help. > > On Thu, Apr 7, 2016 at 3:41 PM, Bart van Deenen > wrote: > >> Hi all >> >> I'm having a datastream transformation, that updates a mutable >> hashmap that exists outside of the stream. >> >> So it's something like >> >> object FlinkJob { >> val uriLookup =3D mutable.HashMap.empty[String, Int] >> >> >> def main(args: Array[String]) { >> >> val stream: DataStream =3D ... >> >> stream.keybBy(1).timeWindow(..).fold(..) >> .window(..) >> .map(..).fold(..) >> .addSink(..) >> } >> } >> >> where the uriLookup hashmap gets updated inside the stream >> transformation, >> and is serialized in the step before the addSink >> >> It works fine, however >> >> Does the snapshotting mechanism in case of a node failure actually >> serialize this map? >> >> And out of curiousity, can I actually see what data exists inside the >> snapshot data? >> >> Thanks. >> >> Bart >> >> > > > -- > BR, > Stefano Baghino > > Software Engineer @ Radicalbit > --001a114014facc6e0a052fe7bb87 Content-Type: text/html; charset=UTF-8 Content-Transfer-Encoding: quoted-printable
Hi,
good explanation and pointers!

I just want to add that the uriLookup table in your example is not re= ally shared between your operator instances in a distributed setting. When = serializing your transformations the current state of the HashMap is serial= ized with them because it is in the closure of the transformations. Then, o= n the cluster, the HashMap is serialized and all parallel instances basical= ly work on their now local copy of the empty HashMap.

<= div>Cheers,
Aljoscha

On Thu, 7 Apr 2016 at 18:30 Stefano Baghino <stefano.baghino@radicalbit.io> = wrote:
Hi Bart,
to make sure that the state is checkpointed you have to:
  1. configure your Flink installation with a reliable state ba= ckend (optional for development, you can read more about it here)
  2. exp= licitly enable checkpointing in your program (see how here =E2=80=94 it's just a couple of = lines of code)
  3. extend your operators so that they checkpoint data, = by implementing the `Checkpointed` interface or using an instance field (th= e semantics of the two approaches are slightly different, you can read more= about it here)
= When your data is checkpointed you can access the state if your operator im= plements the `RichFunction` interface (via an abstract class that wraps the= operator you need to implement, like `RichMapFunction`).

For your need in particular, I don't know a way to checkpoint s= tate shared between different operators; perhaps you can you refactor your = code so that the state is encapsulated in an operator implementation and th= en moved through your pipeline as a parameter of the following operators. W= ould that work?

I apologize for just providi= ng pointers to the docs in my reply but checkpointing deserves a good expla= nation and I feel the docs get the job done pretty well. I will gladly help= you if you have any doubt.

Hope I've been of = some help.

On Thu, Apr 7, 2016 at 3:41 PM, Bart = van Deenen <bartvandeenen@fastmail.fm> wrote:
Hi all

I'm having a datastream transformation, that updates a mutable
hashmap that exists outside of the stream.

So it's something like

object FlinkJob {
=C2=A0 val uriLookup =3D mutable.HashMap.empty[String, Int]


=C2=A0 def main(args: Array[String]) {

=C2=A0 =C2=A0 val stream: DataStream =3D ...

=C2=A0 =C2=A0 stream.keybBy(1).timeWindow(..).fold(..)
=C2=A0 =C2=A0 .window(..)
=C2=A0 =C2=A0 .map(..).fold(..)
=C2=A0 =C2=A0 .addSink(..)
=C2=A0 }
}

where the uriLookup hashmap gets updated inside the stream
transformation,
and is serialized in the step before the addSink

It works fine, however

Does the snapshotting mechanism in case of a node failure actually
serialize this map?

And out of curiousity, can I actually see what data exists inside the
snapshot data?

Thanks.

Bart




--
BR,
<= /div>Stefano Baghino

Software Engineer @ Radicalbit
<= /div>
--001a114014facc6e0a052fe7bb87--