flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Stephan Ewen <se...@apache.org>
Subject Re: serialization error when using multiple metrics counters
Date Mon, 09 Oct 2017 18:44:08 GMT
Interesting, is there a quirk in Scala that using multiple lazy variables
results possibly in eager initialization of some?

On Mon, Oct 9, 2017 at 4:37 PM, Kostas Kloudas <k.kloudas@data-artisans.com>
wrote:

> Hi Colin,
>
> Are you initializing your counters from within the open() method of you
> rich function?
> In other words, are you calling
>
> counter = getRuntimeContext.getMetricGroup.counter(“my counter”)
>
> from within the open().
>
> The counter interface is not serializable. So if you instantiate the
> counters outside the open(),
> when Flink tries to ship your code to the cluster, it cannot so you get
> the exception.
>
> You can have a look at the docs for an example:
> https://ci.apache.org/projects/flink/flink-docs-release-1.3/
> monitoring/metrics.html
>
> Thanks,
> Kostas
>
> On Oct 7, 2017, at 11:34 PM, Colin Williams <colin.williams.seattle@gmail.
> com> wrote:
>
> I've created a RichMapFunction in scala with multiple counters like:
>
>    lazy val successCounter = getRuntimeContext.getMetricGro
> up.counter("successfulParse")
>    lazy val failedCounter = getRuntimeContext.getMetricGro
> up.counter("failedParse")
>    lazy val errorCounter = getRuntimeContext.getMetricGro
> up.counter("errorParse")
>
> which I increment in the map function. While testing I noticed that I have
> no issues with using a single counter. However with multiple counters I get
> a serialization error using more than one counter.
>
> Does anyone know how I can use multiple counters from my RichMapFunction,
> or what I'm doing wrong?
>
> [info]   org.apache.flink.api.common.InvalidProgramException: The
> implementation of the RichMapFunction is not serializable. The object
> probably contains or references non serializable fields.
> [info]   at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleane
> r.java:100)
> [info]   at org.apache.flink.streaming.api.environment.StreamExecutionEn
> vironment.clean(StreamExecutionEnvironment.java:1548)
> [info]   at org.apache.flink.streaming.api.datastream.DataStream.clean(D
> ataStream.java:183)
> [info]   at org.apache.flink.streaming.api.datastream.DataStream.map(Dat
> aStream.java:527)
> [info]   at org.apache.flink.streaming.api.scala.DataStream.map(DataStre
> am.scala:581)
> [info]   at ParsedResultUnwrapperTest$$anonfun$2.apply(ParsedResultUnwra
> pperTest.scala:27)
> [info]   at ParsedResultUnwrapperTest$$anonfun$2.apply(ParsedResultUnwra
> pperTest.scala:23)
> [info]   at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
> [info]   at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
> [info]   at org.scalatest.Transformer.apply(Transformer.scala:22)
> [info]   ...
> [info]   Cause: java.io.NotSerializableException:
> org.apache.flink.metrics.SimpleCounter
> [info]   at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.j
> ava:1184)
> [info]   at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputSt
> ream.java:1548)
> [info]   at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStrea
> m.java:1509)
> [info]   at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputS
> tream.java:1432)
> [info]   at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.j
> ava:1178)
> [info]   at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.ja
> va:348)
> [info]   at org.apache.flink.util.InstantiationUtil.serializeObject(Inst
> antiationUtil.java:315)
> [info]   at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleane
> r.java:81)
> [info]   at org.apache.flink.streaming.api.environment.StreamExecutionEn
> vironment.clean(StreamExecutionEnvironment.java:1548)
> [info]   at org.apache.flink.streaming.api.datastream.DataStream.clean(D
> ataStream.java:183)
> [info]   ...
> [info] - ParseResultUnwrapper.errorCounter.getCount should return 1L for
> a Error -> ParseResult[LineProtocol] *** FAILED ***
> [info]   org.apache.flink.api.common.InvalidProgramException: The
> implementation of the RichMapFunction is not serializable. The object
> probably contains or references non serializable fields.
> [info]   at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleane
> r.java:100)
> [info]   at org.apache.flink.streaming.api.environment.StreamExecutionEn
> vironment.clean(StreamExecutionEnvironment.java:1548)
> [info]   at org.apache.flink.streaming.api.datastream.DataStream.clean(D
> ataStream.java:183)
> [info]   at org.apache.flink.streaming.api.datastream.DataStream.map(Dat
> aStream.java:527)
> [info]   at org.apache.flink.streaming.api.scala.DataStream.map(DataStre
> am.scala:581)
> [info]   at ParsedResultUnwrapperTest$$anonfun$3.apply(ParsedResultUnwra
> pperTest.scala:37)
> [info]   at ParsedResultUnwrapperTest$$anonfun$3.apply(ParsedResultUnwra
> pperTest.scala:32)
> [info]   at org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
> [info]   at org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
> [info]   at org.scalatest.Transformer.apply(Transformer.scala:22)
> [info]   ...
> [info]   Cause: java.io.NotSerializableException:
> org.apache.flink.metrics.SimpleCounter
> [info]   at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.j
> ava:1184)
> [info]   at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputSt
> ream.java:1548)
> [info]   at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStrea
> m.java:1509)
> [info]   at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputS
> tream.java:1432)
> [info]   at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.j
> ava:1178)
> [info]   at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.ja
> va:348)
> [info]   at org.apache.flink.util.InstantiationUtil.serializeObject(Inst
> antiationUtil.java:315)
> [info]   at org.apache.flink.api.java.ClosureCleaner.clean(ClosureCleane
> r.java:81)
> [info]   at org.apache.flink.streaming.api.environment.StreamExecutionEn
> vironment.clean(StreamExecutionEnvironment.java:1548)
> [info]   at org.apache.flink.streaming.api.datastream.DataStream.clean(D
> ataStream.java:183)
> [info]   ...
>
>
>

Mime
View raw message