You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by izual <iz...@163.com> on 2020/09/01 06:43:43 UTC
Re:Re: Questions of "State Processing API in Scala"
I tried to fix the small mistake of sample code in State-Processor-API doc[1], could someone do a doc review[2] for me, thank you.
1: https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/state_processor_api.html#keyed-state
2: https://github.com/apache/flink/pull/13266
At 2020-01-21 15:54:56, "Tzu-Li (Gordon) Tai" <tz...@apache.org> wrote:
>Hi Izual,
>
>Thanks for reporting this! I'm also forwarding this to the user mailing
>list, as that is the more suitable place for this question.
>
>I think the usability of the State Processor API in Scala is indeed
>something that hasn’t been looked at closely yet.
>
>On Tue, Jan 21, 2020 at 8:12 AM izual <iz...@163.com> wrote:
>
>> Hi community,
>>
>> When I use state in Scala, something makes confused, I followed these
>> steps to generate and read states:
>>
>> a. implements the example[1] `CountWindowAverage` in Scala(exactly same),
>> and run jobA => that makes good.
>>
>> b. execute `flink cancel -s ${JobID}` => savepoints was generated as
>> expected.
>>
>> c. implements the example[2] `StatefulFunctionWithTime` in Scala(code
>> below), and run jobB => failed, exceptions shows that "Caused by:
>> org.apache.flink.util.StateMigrationException: The new key serializer must
>> be compatible."
>>
>>
>> ReaderFunction code as below:
>>
>> ```
>>
>> class ReaderFunction extends KeyedStateReaderFunction[Long, (Long,
>> Long)] {
>>
>> var countState: ValueState[(Long, Long)] = _
>>
>> override def open(parameters: Configuration): Unit = {
>>
>> val stateDescriptor = new ValueStateDescriptor("average",
>> createTypeInformation[(Long, Long)])
>>
>> countState = getRuntimeContext().getState(stateDescriptor)
>>
>> }
>>
>> override def readKey(key: Long, ctx: KeyedStateReaderFunction.Context,
>> out: Collector[(Long, Long)]): Unit = {
>>
>> out.collect(countState.value())
>>
>> }
>>
>> }
>>
>> ```
>>
>> d. then I try to use java.lang.Long instead of Long in key-type, and run
>> jobB => exception just disappeared and that makes good.
>>
>> This makes me confused. Did I miss some features in State-Processing-API,
>> such as `magic-implicits`?
>>
>
>This part is explainable. The "magic-implicits" actually happen in the
>DataStream Scala API.
>Any primitive Scala types will inferred and serialized as their Java
>counterparts.
>AFAIK, this would not happen in the State Processor API yet and therefore
>why you are getting the StateMigrationException.
>When using Scala types directly with the State Processor API, I would guess
>that Kryo (as a generic fallback) was being used to access state.
>This can probably be confirmed by looking at the exception stack trace. Can
>you post a full copy of that?
>
>This should be resolvable by properly supporting Scala for the State
>Processor API, but it's just that up to this point, we didn't have a plan
>for that yet.
>Can you open a JIRA for this? I think it'll be a reasonable extension to
>the API.
>
>
>>
>> And when I change `xxx.keyBy(_._1)` to `xxx.keyBy(0)`，same exception comes
>> again，this time I tried to use Tuple(java.lang.Long) or something else, but
>> does not work.
>>
>
>I'm not sure what you mean here. Where is this keyBy happening? In the
>Scala DataStream job, or the State Processor API?
>
>
>>
>> Hope.
>>
>> 1:
>> https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/state/state.html#using-managed-keyed-state
>>
>> 2:
>> https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/state/state.html#using-managed-keyed-state
>
>
>Cheers,
>Gordon
Re: Re: Questions of "State Processing API in Scala"
Posted by Matthias Pohl <ma...@ververica.com>.
Hi Izual,
thanks for contributing and improving the documentation. The PR will be
picked up as part of our regular maintenance work. The communication will
happen through PR conversations as soon as someone picks it up.
Best,
Matthias
On Tue, Sep 1, 2020 at 8:44 AM izual <iz...@163.com> wrote:
> I tried to fix the small mistake of sample code in State-Processor-API
> doc[1], could someone do a doc review[2] for me, thank you.
>
> 1:
> https://ci.apache.org/projects/flink/flink-docs-master/dev/libs/state_processor_api.html#keyed-state
> 2: https://github.com/apache/flink/pull/13266
>
>
>
> At 2020-01-21 15:54:56, "Tzu-Li (Gordon) Tai" <tz...@apache.org> wrote:
> >Hi Izual,
> >
> >Thanks for reporting this! I'm also forwarding this to the user mailing
> >list, as that is the more suitable place for this question.
> >
> >I think the usability of the State Processor API in Scala is indeed
> >something that hasn’t been looked at closely yet.
> >
> >On Tue, Jan 21, 2020 at 8:12 AM izual <iz...@163.com> wrote:
> >
> >> Hi community,
> >>
> >> When I use state in Scala, something makes confused, I followed these
> >> steps to generate and read states:
> >>
> >> a. implements the example[1] `CountWindowAverage` in Scala(exactly same),
> >> and run jobA => that makes good.
> >>
> >> b. execute `flink cancel -s ${JobID}` => savepoints was generated as
> >> expected.
> >>
> >> c. implements the example[2] `StatefulFunctionWithTime` in Scala(code
> >> below), and run jobB => failed, exceptions shows that "Caused by:
> >> org.apache.flink.util.StateMigrationException: The new key serializer must
> >> be compatible."
> >>
> >>
> >> ReaderFunction code as below:
> >>
> >> ```
> >>
> >> class ReaderFunction extends KeyedStateReaderFunction[Long, (Long,
> >> Long)] {
> >>
> >> var countState: ValueState[(Long, Long)] = _
> >>
> >> override def open(parameters: Configuration): Unit = {
> >>
> >> val stateDescriptor = new ValueStateDescriptor("average",
> >> createTypeInformation[(Long, Long)])
> >>
> >> countState = getRuntimeContext().getState(stateDescriptor)
> >>
> >> }
> >>
> >> override def readKey(key: Long, ctx: KeyedStateReaderFunction.Context,
> >> out: Collector[(Long, Long)]): Unit = {
> >>
> >> out.collect(countState.value())
> >>
> >> }
> >>
> >> }
> >>
> >> ```
> >>
> >> d. then I try to use java.lang.Long instead of Long in key-type, and run
> >> jobB => exception just disappeared and that makes good.
> >>
> >> This makes me confused. Did I miss some features in State-Processing-API,
> >> such as `magic-implicits`?
> >>
> >
> >This part is explainable. The "magic-implicits" actually happen in the
> >DataStream Scala API.
> >Any primitive Scala types will inferred and serialized as their Java
> >counterparts.
> >AFAIK, this would not happen in the State Processor API yet and therefore
> >why you are getting the StateMigrationException.
> >When using Scala types directly with the State Processor API, I would guess
> >that Kryo (as a generic fallback) was being used to access state.
> >This can probably be confirmed by looking at the exception stack trace. Can
> >you post a full copy of that?
> >
> >This should be resolvable by properly supporting Scala for the State
> >Processor API, but it's just that up to this point, we didn't have a plan
> >for that yet.
> >Can you open a JIRA for this? I think it'll be a reasonable extension to
> >the API.
> >
> >
> >>
> >> And when I change `xxx.keyBy(_._1)` to `xxx.keyBy(0)`，same exception comes
> >> again，this time I tried to use Tuple(java.lang.Long) or something else, but
> >> does not work.
> >>
> >
> >I'm not sure what you mean here. Where is this keyBy happening? In the
> >Scala DataStream job, or the State Processor API?
> >
> >
> >>
> >> Hope.
> >>
> >> 1:
> >> https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/state/state.html#using-managed-keyed-state
> >>
> >> 2:
> >> https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/stream/state/state.html#using-managed-keyed-state
> >
> >
> >Cheers,
> >Gordon
>
>
>
>
>
--
Matthias Pohl | Engineer
Follow us @VervericaData Ververica <https://www.ververica.com/>
--
Join Flink Forward <https://flink-forward.org/> - The Apache Flink
Conference
Stream Processing | Event Driven | Real Time
--
Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany
--
Ververica GmbH
Registered at Amtsgericht Charlottenburg: HRB 158244 B
Managing Directors: Yip Park Tung Jason, Jinwei (Kevin) Zhang, Karl Anton
Wehner