Yes, that's true.
On Fri, 27 Jan 2017 at 13:16 Nico <nico.franz...@gmail.com> wrote:
> Hi Aljoscha,
>
> got it!!! :) Thank you. So, in order to retain the "original" timestamps,
> it would be necessary to assign the timestemps after the MapFunction
> instead of the kafka source? At lest, this solves the issue in the example.
>
> Best,
> Nico
>
> 2017-01-27 11:49 GMT+01:00 Aljoscha Krettek <aljos...@apache.org>:
>
> Now I see. What you're doing in this example is basically reassigning
> timestamps to other elements in your stateful MapFunction. Flink internally
> keeps track of the timestamp of an element. This can normally not be
> changed, except by using a TimestampAssigner, which you're doing. Now, the
> output from a MapFunction has the same timestamp as the input element. By
> keeping an element in state and emitting it when the next element arrives
> you emit it with the timestamp of that next element and that's the reason
> why the end up in the "wrong" windows.
>
> Does that help?
>
> -
> Aljoscha
>
> On Thu, 26 Jan 2017 at 19:17 Nico <nico.franz...@gmail.com> wrote:
>
> Hi,
>
> can anyone help me with this problem? I don't get it. Forget the examples
> below, I've created a copy / paste example to reproduce the problem of
> incorrect results when using key-value state und windowOperator.
>
>
> public class StreamingJob {
>
> public static void main(String[] args) throws Exception {
> // set up the streaming execution environment
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
>
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>
> DataStream<Tuple2<String,Long>> stream = env.fromElements(
> new Tuple2<>("1",1485446260994L),
> new Tuple2<>("1",1485446266012L),
> new Tuple2<>("1",1485446271031L),
> new Tuple2<>("1",1485446276040L),
> new Tuple2<>("1",1485446281045L),
> new Tuple2<>("1",1485446286049L),
> new Tuple2<>("1",1485446291062L),
> new Tuple2<>("1",1485446296066L),
> new Tuple2<>("1",1485446302019L)
> );
>
> stream
> .assignTimestampsAndWatermarks(new
> BoundedOutOfOrdernessTimestampExtractor<Tuple2<String,
> Long>>(Time.seconds(0)) {
> @Override
> public long extractTimestamp(Tuple2<String, Long> stringLongTuple2) {
> return stringLongTuple2.f1;
> }
> })
> .keyBy("f0")
> .map(new MapTest())
> .keyBy("f0")
> .window(TumblingEventTimeWindows.of(Time.seconds(20)))
> .apply(new WindowFunction<Tuple2<String,Long>, Object, Tuple,
> TimeWindow>() {
> @Override
> public void apply(Tuple tuple, TimeWindow timeWindow,
> Iterable<Tuple2<String, Long>> iterable, Collector<Object> collector)
> throws Exception {
>
> Set<Long> set = new HashSet<>();
> for(Tuple2<String,Long> t : iterable){
> set.add(t.f1);
> }
>
> StringBuilder sb = new StringBuilder();
>
> sb.append("Window [" +timeWindow.getStart() +" " +timeWindow.getEnd() +"]
> ");
> sb.append("Set " +set.toString());
> System.out.println(sb.toString());
> }
> })
> .print();
>
>
> // execute program
> env.execute("Flink Streaming Java API Skeleton");
> }
>
> private static class MapTest extends
> RichMapFunction<Tuple2<String,Long>,Tuple2<String,Long>> {
>
> private transient ValueState<Tuple2<String, Long>> state;
>
> @Override
> public Tuple2<String, Long> map(Tuple2<String, Long> stringLongTuple2)
> throws Exception {
>
> Tuple2<String,Long> t = state.value();
>
> state.update(stringLongTuple2);
>
> if(t == null) return stringLongTuple2;
>
> return t;
> }
>
> @Override
> public void open(Configuration parameters) throws Exception {
>
> ValueStateDescriptor<Tuple2<String,Long>> vsd = new ValueStateDescriptor<>(
> "lastEvent",
> TypeInformation.of(new TypeHint<Tuple2<String, Long>>() {}),
> null
> );
>
> state = getRuntimeContext().getState(vsd);
> }
> }
> }
>
>
> Output:
>
> Window [1485446260000 1485446280000] Set [1485446271031, 1485446260994,
> 1485446266012]
> Window [1485446280000 1485446300000] Set [1485446291062, 1485446281045,
> 1485446286049, 1485446276040]
> Window [1485446300000 1485446320000] Set [1485446296066]
>
> Best,
> Nico
>
> BTW ... I am using Flink 1.1.3.
>
>
> 2017-01-16 12:18 GMT+01:00 Nico <nico.franz...@gmail.com>:
>
> Hi Aljoscha,
>
> is was able to identify the root cause of the problem. It is my first map
> function using the ValueState. But first, the
> assignTimestampsAndWatermarks() is called after the connector to Kafka is
> generated:
>
> FlinkKafkaConsumer09<Car> carFlinkKafkaConsumer09 =
> new FlinkKafkaConsumer09<>("Traffic", new Car(), properties);
>
> // Extrahieren der Timestamps mit max. Delay von 2s
> carFlinkKafkaConsumer09.assignTimestampsAndWatermarks(new
> TimestampGenerator(Time.seconds(0)));
>
> In the map function I try to calculate the direction between two GPS data
> points. For this, I store the last event in ValueState. The function looks
> like this:
>
> private static class BearingMap extends RichMapFunction<Car, Car> {
>
> private transient ValueState<Car> state;
> private final double maxdiff = 12; // in Sekunden
>
> @Override
> public Car map(Car destination) throws Exception {
>
> Car origin = state.value();
> double olat, olon, dlat, dlon;
>
> /**
> * Wenn State leer, berechne keine Richtung, sondern speichere Event
> nur in State
> */
> if (origin == null){
> state.update(destination);
> // gebe Car ohne Aenderungen zurueck
> return destination;
> }
>
> double diff = origin.getTimestamp()-destination.getTimestamp();
>
> System.out.println("Differenz: " +diff);
>
> if(Math.abs(diff) <= maxdiff*1000){
>
> /*
> * Bei spaeten Events, die jedoch noch in den erlaubten Delay fallen
> */
> if(diff > 0){
> Car tmp = destination;
> destination = origin;
> origin = tmp;
> }
>
> /*
> * Car tmp ist immer der Ursprung
> */
>
> double bearing = Helper.calculateBearing(
>
> origin.getLat(),origin.getLon(),destination.getLat(),destination.getLon());
>
> // Update des State's
> state.update(destination);
>
> origin.setDirection(bearing);
> return origin;
>
> }
>
> // Bei zu spaeten Events behalte jetzigen Status und gebe diesen ohne
> Richtung zurück
> return origin;
>
> }
>
>
> @Override
> public void open(Configuration parameters) throws Exception {
>
> ValueStateDescriptor<Car> vsd = new ValueStateDescriptor<>(
> "lastEvent",
> Car.class,
> null
> );
>
> state = getRuntimeContext().getState(vsd);
> }
>
> }
>
> Together with the window function:
>
>
> private static class TimeWindowTest implements WindowFunction<Car,
> Tuple9<Double, Double, Double, Double, Double, Double, Double, Integer,
> List<String>>, Tuple, TimeWindow> {
> @Override
> public void apply(Tuple tuple, TimeWindow timeWindow, Iterable<Car>
> iterable, Collector<Tuple9<Double, Double, Double, Double, Double, Double,
> Double, Integer, List<String>>> collector) throws Exception {
> String s = "Zeitfenster: " +timeWindow.getStart() +" - " +
> timeWindow.getEnd() +"\n";
> Set<Long> timestamps = new HashSet<Long>();
>
> for( Car c : iterable){
> timestamps.add(c.getTimestamp());
> }
>
> System.out.println( s +timestamps +"\n\n");
> }
> }
>
> I get for :
>
> stream
> .filter(new FilterFunction<Car>() {
> @Override
> public boolean filter(Car car) throws Exception {
> return car.getId().equals("car.330");
> }
> })
> .keyBy("id")
> .map(new BearingMap())
> .keyBy("id")
> .window(TumblingEventTimeWindows.of(Time.seconds(10)))
> .apply(new TimeWindowTest());
>
> So actually, when an event e1 arrives the Map Operator, it is stored in
> ValueState and after the next element e2 arrives, e1
> will be forwarded. This is after 5 seconds. This generates the following
> outcome. One Element is always around 5 seconds before the
> start of the window.
>
> Differenz: -5013.0
> Differenz: -5014.0
> Zeitfenster: 1484564690000 - 1484564700000 (Window times start and end)
> [1484564686236, 1484564691260]
>
>
> Differenz: -5009.0
> Differenz: -5007.0
> Zeitfenster: 1484564700000 - 1484564710000
> [1484564696273, 1484564701287]
>
>
> Differenz: -5005.0
> Differenz: -5014.0
> Zeitfenster: 1484564710000 - 1484564720000
> [1484564706296, 1484564711303]
>
>
> Best,
>
> Nico
>
>
>
> 2017-01-09 16:10 GMT+01:00 Aljoscha Krettek <aljos...@apache.org>:
>
> Hi,
> I'm assuming you also have the call to assignTimestampsAndWatermarks()
> somewhere in there as well, as in:
>
> stream
> .assignTimestampsAndWatermarks(new TimestampGenerator()) // or
> somewhere else in the pipeline
> .keyBy("id")
> .map(...)
> .filter(...)
> .map(...)
> .keyBy("areaID")
> .map(new KeyExtractor())
> .keyBy("f1.areaID","f0.sinterval")
> .window(TumblingEventTimeWindows.of(Time.seconds(20)))
> .apply(new TrafficInformation());
>
> Just checking, to make sure. If you have this we might have to dig a
> little deeper. Could you also please trying to bring the whole output of
> your apply() method in one go, i.e. collect all the output in a String and
> then have one call to System.out.println(), it could be that the output in
> the terminal is not completely in order.
>
> Cheers,
> Aljoscha
>
> On Mon, 2 Jan 2017 at 15:04 Nico <nico.franz...@gmail.com> wrote:
>
> Hi Aljoscha,
>
> thank you for having a look. Actually there is not too much code based on
> timestamps:
>
> stream
> .keyBy("id")
> .map(...)
> .filter(...)
> .map(...)
> .keyBy("areaID")
> .map(new KeyExtractor())
> .keyBy("f1.areaID","f0.sinterval")
> .window(TumblingEventTimeWindows.of(Time.seconds(20)))
> .apply(new TrafficInformation());
>
> The map functions only enrich the data and don't change anything related
> to the timestamp.
>
> the apply function is:
>
> @Override
> public void apply(
> Tuple key,
> TimeWindow timeWindow,
> Iterable<Tuple2<DirectionInterval, Car>> cars,
> Collector<Tuple3<String, Double, Double>> out) throws Exception {
>
> System.out.println("Start: " +timeWindow.getStart());
> System.out.println("End: " +timeWindow.getEnd());
>
> for(Tuple2<DirectionInterval, Car> t : cars){
> System.out.println(t.f1);
> }
>
> System.out.println(t.f1) prints all information about a car, in which the
> timestep is embedded. The System gets the timestamp with the class:
>
> public class TimestampGenerator extends
> BoundedOutOfOrdernessTimestampExtractor <Car> {
>
>
> public TimestampGenerator(Time maxOutOfOrderness){
> super(maxOutOfOrderness);
> }
>
> @Override
> public long extractTimestamp(Car car) {
> return car.getTimestamp();
> }
>
>
> Example output is presented in the previous post... it looks like the
> timestamp is rounded... I am confused :-/
>
> Best,
> Nico
>
> 2016-12-23 19:41 GMT+01:00 Aljoscha Krettek <aljos...@apache.org>:
>
> Hi,
> could you please share code (and example data) for producing this output.
> I'd like to have a look.
>
> Cheers,
> Aljoscha
>
> On Wed, 21 Dec 2016 at 16:29 Nico <nico.franz...@gmail.com> wrote:
>
> Hi @all,
>
> I am using a TumblingEventTimeWindows.of(Time.seconds(20)) for testing.
> During this I found a strange behavior (at least for me) in the assignment
> of events.
>
> The first element of a new window is actually always part of the old
> window. I thought the events are late, but then they they would be dropped
> instead of assigned to the new window. Even with a allowedLateness of 10s
> the behavior remains the same.
>
> The used timeWindow.getStart() and getEnd in order to get the boundaries
> of the window.
>
> Can someone explain this?
>
> Best,
> Nico
>
>
> TimeWindows with Elements:
>
> Start: 1482332940000 - End: 1482332960000
> timestamp=1482332952907
>
> Start: 1482332960000 - End: 1482332980000
> timestamp=1482332958929
> timestamp=1482332963995
> timestamp=1482332969027
> timestamp=1482332974039
>
> Start: 1482332980000 - End: 1482333000000
> timestamp=1482332979059
> timestamp=1482332984072
> timestamp=1482332989081
> timestamp=1482332994089
>
> Start: 1482333000000 - End: 1482333020000
> timestamp=1482332999113
> timestamp=1482333004123
> timestamp=1482333009132
> timestamp=1482333014144
>
>
>
>
>
>
