This only works for keyed streams, you have to use keyBy().
You can use the Checkpointed interface instead
(https://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/state.html#checkpointing-instance-fields).
On Thu, Aug 11, 2016 at 3:35 PM, Ramanan, Buvana (Nokia - US)
<buvana.rama...@nokia-bell-labs.com> wrote:
> Hi Kostas,
>
>
>
> Here is my code. All I am trying to compute is (x[t] – x[t-1]), where x[t]
> is the current value of the incoming sample and x[t-1] is the previous value
> of the incoming sample. I store the current value in state store
> (‘prev_tuple’) so that I can use it for computation in next cycle. As you
> may observe, I am not using keyBy. I am simply printing out the resultant
> tuple.
>
>
>
> It appears from the error message that I have to set the key serializer (and
> possibly value serializer) for the state store. I am not sure how to do
> that…
>
>
>
> Thanks for your interest in helping,
>
>
>
>
>
> Regards,
>
> Buvana
>
>
>
> public class stateful {
>
> private static String INPUT_KAFKA_TOPIC = null;
>
> private static int TIME_WINDOW = 0;
>
>
>
> public static void main(String[] args) throws Exception {
>
>
>
> if (args.length < 2) {
>
> throw new IllegalArgumentException("The application needs two
> arguments. The first is the name of the kafka topic from which it has to \n"
>
> + "fetch the data. The second argument is the size of
> the window, in seconds, to which the aggregation function must be applied.
> \n");
>
> }
>
>
>
> INPUT_KAFKA_TOPIC = args[0];
>
> TIME_WINDOW = Integer.parseInt(args[1]);
>
>
>
> Properties properties = null;
>
>
>
> properties = new Properties();
>
> properties.setProperty("bootstrap.servers", "localhost:9092");
>
> properties.setProperty("zookeeper.connect", "localhost:2181");
>
> properties.setProperty("group.id", "test");
>
>
>
> StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
>
> //env.setStateBackend(new
> FsStateBackend("file://home/buvana/flink/checkpoints"));
>
>
>
> DataStreamSource<String> stream = env
>
> .addSource(new FlinkKafkaConsumer09<>(INPUT_KAFKA_TOPIC, new
> SimpleStringSchema(), properties));
>
>
>
> // maps the data into Flink tuples
>
> DataStream<Tuple2<String,Double>> streamTuples = stream.flatMap(new
> Rec2Tuple2());
>
>
>
> // write the result to the console or in a Kafka topic
>
> streamTuples.print();
>
>
>
> env.execute("plus one");
>
>
>
> }
>
>
>
> public static class Rec2Tuple2 extends RichFlatMapFunction<String,
> Tuple2<String,Double> > {
>
> private transient ValueState<Tuple2<String, Double>> prev_tuple;
>
>
>
> @Override
>
> public void flatMap(String incString, Collector<Tuple2<String,
> Double>> out) throws Exception {
>
> try {
>
> Double value = Double.parseDouble(incString);
>
> System.out.println("value = " + value);
>
> Tuple2<String, Double> prev_stored_tp = prev_tuple.value();
>
> System.out.println(prev_stored_tp);
>
>
>
> Double value2 = value - prev_stored_tp.f1;
>
> prev_stored_tp.f1 = value;
>
> prev_stored_tp.f0 = INPUT_KAFKA_TOPIC;
>
> prev_tuple.update(prev_stored_tp);
>
>
>
> Tuple2<String, Double> tp = new Tuple2<String, Double>();
>
> tp.setField(INPUT_KAFKA_TOPIC, 0);
>
> tp.setField(value2, 1);
>
> out.collect(tp);
>
>
>
> } catch (NumberFormatException e) {
>
> System.out.println("Could not convert to Float" +
> incString);
>
> System.err.println("Could not convert to Float" +
> incString);
>
> }
>
> }
>
>
>
> @Override
>
> public void open(Configuration config) {
>
> ValueStateDescriptor<Tuple2<String, Double>> descriptor =
>
> new ValueStateDescriptor<>(
>
> "previous input value", // the state name
>
> TypeInformation.of(new TypeHint<Tuple2<String,
> Double>>() {}), // type information
>
> Tuple2.of("test topic", 0.0)); // default value
> of the state, if nothing was set
>
> prev_tuple = getRuntimeContext().getState(descriptor);
>
> }
>
> }
>
> }
>
>
>
> From: Kostas Kloudas [mailto:k.klou...@data-artisans.com]
> Sent: Thursday, August 11, 2016 5:45 AM
> To: user@flink.apache.org
> Subject: Re: flink - Working with State example
>
>
>
> Hello Buvana,
>
>
>
> Can you share a bit more details on your operator and how you are using it?
>
> For example, are you using keyBy before using you custom operator?
>
>
>
> Thanks a lot,
>
> Kostas
>
>
>
> On Aug 10, 2016, at 10:03 PM, Ramanan, Buvana (Nokia - US)
> <buvana.rama...@nokia-bell-labs.com> wrote:
>
>
>
> Hello,
>
>
>
> I am utilizing the code snippet in:
> https://ci.apache.org/projects/flink/flink-docs-master/apis/streaming/state.html
> and particularly ‘open’ function in my code:
>
> @Override
>
> public void open(Configuration config) {
>
> ValueStateDescriptor<Tuple2<Long, Long>> descriptor =
>
> new ValueStateDescriptor<>(
>
> "average", // 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);
>
> }
>
>
>
> When I run, I get the following error:
>
> Caused by: java.lang.RuntimeException: Error while getting state
>
> at
> org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getState(StreamingRuntimeContext.java:120)
>
> at wikiedits.stateful$Rec2Tuple2.open(stateful.java:103)
>
> at
> org.apache.flink.api.common.functions.util.FunctionUtils.openFunction(FunctionUtils.java:38)
>
> at
> org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.open(AbstractUdfStreamOperator.java:91)
>
> at
> org.apache.flink.streaming.api.operators.StreamFlatMap.open(StreamFlatMap.java:41)
>
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.openAllOperators(StreamTask.java:314)
>
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:214)
>
> at
> org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
>
> at java.lang.Thread.run(Thread.java:745)
>
> Caused by: java.lang.Exception: State key serializer has not been configured
> in the config. This operation cannot use partitioned state.
>
> at
> org.apache.flink.runtime.state.AbstractStateBackend.getPartitionedState(AbstractStateBackend.java:199)
>
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator.getPartitionedState(AbstractStreamOperator.java:260)
>
> at
> org.apache.flink.streaming.api.operators.StreamingRuntimeContext.getState(StreamingRuntimeContext.java:118)
>
> ... 8 more
>
>
>
> Where do I define the key & value serializer for state?
>
>
>
> Thanks,
>
> Buvana
>
>
