Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
MIME-Version: 1.0
From: Victor Godoy Poluceno <victorpoluceno@gmail.com>
Date: Tue, 25 Jul 2017 09:45:49 -0300
Message-ID: <CAGYsAp-8jXzNCCPJ6Wu58JyoD9gHxi6Kn_rHsxPJ+x9V76NY-Q@mail.gmail.com>
Subject: Unable to make mapWithState work correctly
To: user@flink.apache.org
Content-Type: multipart/alternative; boundary="001a1147396039e7d9055523b7fb"

--001a1147396039e7d9055523b7fb
Content-Type: text/plain; charset="UTF-8"

Hi,

I am trying to write a simple streaming program to count values from a
Kafka topic in a fault tolerant manner, like this
<https://gist.github.com/victorpoluceno/8690df8459bf3afd60477f83ec78f7a8>:

<code>
val config: Configuration = new Configuration()
config.setString(ConfigConstants.STATE_BACKEND, "filesystem")
config.setString("state.backend.fs.checkpointdir", "file:///tmp/flink")

val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.enableCheckpointing(10)

val properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "test");

val stream = env
    .addSource(new FlinkKafkaConsumer010[String]("test", new
SimpleStringSchema(), properties))
    .map((_, 1))
    .keyBy(_._1)
    .mapWithState((in: (String, Int), count: Option[Int]) => {
      val newCount = in._2 + count.getOrElse(0)
      ((in._1, newCount), Some(newCount))
    }).print

env.execute("Job")
</code>

The idea is to use the filesystem state backend to persist the computation
state (count) and to restore the computation state in case of failure or
restart. I have a program that inject the same key on Kafka. But I am
unable to make Flink work correctly, every time the Flink restarts the
value from state is empty, so the count starts from zero. What am I missing
here?

I am running this on a local environment (sbt run) with Flink 1.3.1, Java
1.8.0_131, and Ubuntu 16.04.

-- 
hooray!

--
Victor Godoy Poluceno

--001a1147396039e7d9055523b7fb
Content-Type: text/html; charset="UTF-8"
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr"><div><div><div>Hi,<br><br></div>I am trying to write a sim=
ple streaming program to count values from a Kafka topic in a fault toleran=
t manner, like <a href=3D"https://gist.github.com/victorpoluceno/8690df8459=
bf3afd60477f83ec78f7a8">this</a>:<br><br></div>&lt;code&gt;<br>val config: =
Configuration =3D new Configuration()<br>config.setString(ConfigConstants.S=
TATE_BACKEND, &quot;filesystem&quot;)<br>config.setString(&quot;state.backe=
nd.fs.checkpointdir&quot;, &quot;file:///tmp/flink&quot;)<br><br>val env =
=3D StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(config)<br>e=
nv.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)=
;<br>env.enableCheckpointing(10)<br><br>val properties =3D new Properties()=
;<br>properties.setProperty(&quot;bootstrap.servers&quot;, &quot;localhost:=
9092&quot;);<br>properties.setProperty(&quot;<a href=3D"http://group.id">gr=
oup.id</a>&quot;, &quot;test&quot;);<br><br>val stream =3D env<br>=C2=A0=C2=
=A0=C2=A0 .addSource(new FlinkKafkaConsumer010[String](&quot;test&quot;, ne=
w SimpleStringSchema(), properties))<br>=C2=A0=C2=A0=C2=A0 .map((_, 1))<br>=
=C2=A0=C2=A0=C2=A0 .keyBy(_._1)<br>=C2=A0=C2=A0=C2=A0 .mapWithState((in: (S=
tring, Int), count: Option[Int]) =3D&gt; {<br>=C2=A0=C2=A0=C2=A0=C2=A0=C2=
=A0 val newCount =3D in._2 + count.getOrElse(0)<br>=C2=A0=C2=A0=C2=A0=C2=A0=
=C2=A0 ((in._1, newCount), Some(newCount))<br>=C2=A0=C2=A0=C2=A0 }).print<b=
r><br>env.execute(&quot;Job&quot;)<br></div>&lt;/code&gt;<br><div><div><br>=
The idea is to use the filesystem state backend to persist the computation =
state (count) and to restore the computation state in case of failure or re=
start. I have a program that inject the same key on Kafka. But I am unable =
to make Flink work correctly, every time the Flink restarts the value from =
state is empty, so the count starts from zero. What am I missing here?<br><=
br></div><div>I am running this on a local environment (sbt run) with Flink=
 1.3.1, Java 1.8.0_131, and Ubuntu 16.04.<br clear=3D"all"></div><div><div>=
<div><div><br>-- <br><div class=3D"gmail_signature">hooray!<br><br>--<br>Vi=
ctor Godoy Poluceno</div>
</div></div></div></div></div></div>

--001a1147396039e7d9055523b7fb--