flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Nirmalya Sengupta <sengupta.nirma...@gmail.com>
Subject Confusion about multiple use of one ValueState
Date Fri, 13 May 2016 03:28:00 GMT
Hello all,

Let's say I want to hold some state value derived during one
transformation, and then use that same state value in a subsequent
transformation? For example:

myStream
.keyBy(fieldID) // Some field ID, may be 0
.map(new MyStatefulMapper())
.map(new MySubsequentMapper())
....

Now, I define MyStatefulMapper in the usual fashion:

public class MyStatefulMapper extends RichFlatMapFunction<Tuple2<Long,
Long>, Tuple2<Long, Long>> {

    /**     * The ValueState handle. The first field is the count, the
second field a running sum.     */
    private transient ValueState<Tuple2<Long, Long>> sum;

    @Override
    public void flatMap(Tuple2<Long, Long> input,
Collector<Tuple2<Long, Long>> out) throws Exception {

       // logic of accessing and updating the ValueState 'sum' above
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
                new ValueStateDescriptor<>(
                        "mySum", // the state name
                        TypeInformation.of(new TypeHint<Tuple2<Long,
Long>>() {}), // type information
                        Tuple2.of(0L, 0L)); // default value of the
state, if nothing was set
        sum = getRuntimeContext().getState(descriptor);
    }}


So, by now, RuntimeContext has registered a State holder named 'mySum'.

In the implementation of 'MySubsequentMapper', I need to access this State
holder named 'mySum', perhaps thus (my thinking, I may be wrong):

public class MySubsequentMapper extends
RichFlatMapFunction<Tuple2<Long, Long>, Tuple2<Long, Long>> {

    /**     * The ValueState handle. The first field is the count, the
second field a running sum.     */
    private transient ValueState<Tuple2<Long, Long>> aSubsequentSum;

    private transient ValueState<Tuple2<Long, Long>> sum; // defined earlier


    @Override
    public void flatMap(Tuple2<Long, Long> input,
Collector<Tuple2<Long, Long>> out) throws Exception {

       // logic of accessing and updating the ValueState 'aSubsequentSum' above

       // but this logic depends on the current contents of ValueState
'sum' created earlier
    }

    @Override
    public void open(Configuration config) {
        // Logic to create ValueDescriptor for 'aSubsequentSum' which
is owned by this operator

        // ...

        // Question: now, how do I prepare for accessing 'sum' which
is a State holder, but created inside an earlier operator?
        sum = getRuntimeContext().getState(descriptor) // how can I
pass the name 'mySum' (used in StateDescriptor)?
    }}

I have two questions:

1) What I am trying to achieve: is that possible and even, advisable? If
not, then what is the alternative?
2) Is there a guarantee that Flink will execute MyStatefulOperator.open()
always before MySubsequentOperator.open() because of the lexical order of
appearance in the source code?

-- Nirmalya




-- 
Software Technologist
http://www.linkedin.com/in/nirmalyasengupta
"If you have built castles in the air, your work need not be lost. That is
where they should be.
Now put the foundation under them."

Mime
View raw message