(DEPRECATED) Apache Flink User Mailing List archive.

Finding the Maximum Value Received so far in a Stream

classic Classic list List threaded Threaded
♦
♦
 6 messages Options
Komal Mariam
Reply | Threaded
Open this post in threaded view
♦
♦
|

Finding the Maximum Value Received so far in a Stream

Komal Mariam
Hello all,
I'm trying to do a fairly simple task that is to find the maximum value (Double) received so far in a stream. This is what I implemented:

POJO class:
public class Fish{
    public Integer fish_id;
    public Point coordinate;   //position
  
public Fish() {};

public Fish(fish_id,double x, double y) {
   //assign to fish object
}

Java_main.java
 DataStream text  = env
                .addSource(new FlinkKafkaConsumer<>("test", new JSONKeyValueDeserializationSchema(false),properties).setStartFromLatest());

      DataStream<Fish> fishes = text.map(new MapFunction<ObjectNode, Fish>() {
            @Override
            public Fish map(ObjectNode json) throws Exception {
               Fish fishes = new Fish( json.get("value").get("id").asInteger() ,json.get("value").get("x_begin").asDouble(),json.get("value").get("y_begin").asDouble());
                return fishes;
            }
        });


 I can't seem to apply aggregations on the Point class without extracting the x coordinates in a separate stream so here are the 2 methods I have applied:

Method 1: Simple reduce

KeyedStream<Fish, Integer> keyed = fishes.keyBy(s->s.fish_id);
        keyed.reduce(new ReduceFunction<Fish>() {
            @Override
            public Fish reduce(Fish t, Fish t1) throws Exception {
                if (t.X > t1.X) {
                    return t;
                } else
                    return t1;
            }
        }).map(new MapFunction<Fish, Double>() {
                    @Override
                    public Double map(Fish t) throws Exception {
                        return t.point.getX();
                    }
                }).print();


Result: 1> -73.8517632
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.851446       //smaller than previous value!
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.8517632
1> -73.851446
3> -73.85012          
3> -73.850212
3> -73.851979     //smaller than previous value
3> -73.850212
3> -73.8512969
3> -73.8512969  

1) I'm trying to compute the max so why do I see smaller values after the latest maximum value. I think this is because order of the outputs is not preserved as same as inputs?
If this is so how do I ensure that the order is preserved and I only see the latest maximum value?

2) Another thing I have noticed is that if instance 1 produces the max say -73.8505642 but then instance 2 would produce  -73.9064 which is again smaller than the value produced by instance 1. I'm assuming its because there is no communication between parallel instances hence they produce two value. If this is so how do I get ONE maximum value from all the parallel instances combined? 

Method 2: Using States

public class GetMaximum extends RichMapFunction<Fish, Fish> {

    private transient ValueState<Fish> max;

    @Override
    public Fish map(Fish input) throws Exception {

        // access the state value
       Fish currentMaximum = max.value();

        if (input.point.getX() > currentMaximum.point.getX()) {
            currentMaximum.objid = input.objid;
            currentMaximum.point = (org.locationtech.jts.geom.Point) input.point.clone();
            max.update(currentMaximum);
        }
        return currentMaximum;
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Fish> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Fish>() {}), // type information
                        new Fish(-100,0)); // default value of the state, if nothing was set
        max = getRuntimeContext().getState(descriptor);
    }
}

3) I'm getting the same results as method 1. isn't ValueState shared between all instances of the same operator? 

4) Out of the two methodologies which is a better choice?

Would really appreciate your help!

Best Regards,
Komal
Jörn Franke
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Finding the Maximum Value Received so far in a Stream

Jörn Franke
You can not compare doubles in Java (and other languages). Reason is that double has
a limited precision and is rounded. See here for some examples and discussion:


Am 03.10.2019 um 08:26 schrieb Komal Mariam <[hidden email]>:


Hello all,
I'm trying to do a fairly simple task that is to find the maximum value (Double) received so far in a stream. This is what I implemented:

POJO class:
public class Fish{
    public Integer fish_id;
    public Point coordinate;   //position
  
public Fish() {};

public Fish(fish_id,double x, double y) {
   //assign to fish object
}

Java_main.java
 DataStream text  = env
                .addSource(new FlinkKafkaConsumer<>("test", new JSONKeyValueDeserializationSchema(false),properties).setStartFromLatest());

      DataStream<Fish> fishes = text.map(new MapFunction<ObjectNode, Fish>() {
            @Override
            public Fish map(ObjectNode json) throws Exception {
               Fish fishes = new Fish( json.get("value").get("id").asInteger() ,json.get("value").get("x_begin").asDouble(),json.get("value").get("y_begin").asDouble());
                return fishes;
            }
        });


 I can't seem to apply aggregations on the Point class without extracting the x coordinates in a separate stream so here are the 2 methods I have applied:

Method 1: Simple reduce

KeyedStream<Fish, Integer> keyed = fishes.keyBy(s->s.fish_id);
        keyed.reduce(new ReduceFunction<Fish>() {
            @Override
            public Fish reduce(Fish t, Fish t1) throws Exception {
                if (t.X > t1.X) {
                    return t;
                } else
                    return t1;
            }
        }).map(new MapFunction<Fish, Double>() {
                    @Override
                    public Double map(Fish t) throws Exception {
                        return t.point.getX();
                    }
                }).print();


Result: 1> -73.8517632
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.851446       //smaller than previous value!
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.8517632
1> -73.851446
3> -73.85012          
3> -73.850212
3> -73.851979     //smaller than previous value
3> -73.850212
3> -73.8512969
3> -73.8512969  

1) I'm trying to compute the max so why do I see smaller values after the latest maximum value. I think this is because order of the outputs is not preserved as same as inputs?
If this is so how do I ensure that the order is preserved and I only see the latest maximum value?

2) Another thing I have noticed is that if instance 1 produces the max say -73.8505642 but then instance 2 would produce  -73.9064 which is again smaller than the value produced by instance 1. I'm assuming its because there is no communication between parallel instances hence they produce two value. If this is so how do I get ONE maximum value from all the parallel instances combined? 

Method 2: Using States

public class GetMaximum extends RichMapFunction<Fish, Fish> {

    private transient ValueState<Fish> max;

    @Override
    public Fish map(Fish input) throws Exception {

        // access the state value
       Fish currentMaximum = max.value();

        if (input.point.getX() > currentMaximum.point.getX()) {
            currentMaximum.objid = input.objid;
            currentMaximum.point = (org.locationtech.jts.geom.Point) input.point.clone();
            max.update(currentMaximum);
        }
        return currentMaximum;
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Fish> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Fish>() {}), // type information
                        new Fish(-100,0)); // default value of the state, if nothing was set
        max = getRuntimeContext().getState(descriptor);
    }
}

3) I'm getting the same results as method 1. isn't ValueState shared between all instances of the same operator? 

4) Out of the two methodologies which is a better choice?

Would really appreciate your help!

Best Regards,
Komal
Jörn Franke
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Finding the Maximum Value Received so far in a Stream

Jörn Franke
In reply to this post by Komal Mariam
Btw. Why don’t you use the max method?  


See here about the state solution:


About your reduce function:
You execute it by fish_id if I see it correctly. This will create one result by fish_id . I propose to map first all fish coordinates under a single key and then reduce by this single key. 

Am 03.10.2019 um 08:26 schrieb Komal Mariam <[hidden email]>:


Hello all,
I'm trying to do a fairly simple task that is to find the maximum value (Double) received so far in a stream. This is what I implemented:

POJO class:
public class Fish{
    public Integer fish_id;
    public Point coordinate;   //position
  
public Fish() {};

public Fish(fish_id,double x, double y) {
   //assign to fish object
}

Java_main.java
 DataStream text  = env
                .addSource(new FlinkKafkaConsumer<>("test", new JSONKeyValueDeserializationSchema(false),properties).setStartFromLatest());

      DataStream<Fish> fishes = text.map(new MapFunction<ObjectNode, Fish>() {
            @Override
            public Fish map(ObjectNode json) throws Exception {
               Fish fishes = new Fish( json.get("value").get("id").asInteger() ,json.get("value").get("x_begin").asDouble(),json.get("value").get("y_begin").asDouble());
                return fishes;
            }
        });


 I can't seem to apply aggregations on the Point class without extracting the x coordinates in a separate stream so here are the 2 methods I have applied:

Method 1: Simple reduce

KeyedStream<Fish, Integer> keyed = fishes.keyBy(s->s.fish_id);
        keyed.reduce(new ReduceFunction<Fish>() {
            @Override
            public Fish reduce(Fish t, Fish t1) throws Exception {
                if (t.X > t1.X) {
                    return t;
                } else
                    return t1;
            }
        }).map(new MapFunction<Fish, Double>() {
                    @Override
                    public Double map(Fish t) throws Exception {
                        return t.point.getX();
                    }
                }).print();


Result: 1> -73.8517632
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.851446       //smaller than previous value!
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.8517632
1> -73.851446
3> -73.85012          
3> -73.850212
3> -73.851979     //smaller than previous value
3> -73.850212
3> -73.8512969
3> -73.8512969  

1) I'm trying to compute the max so why do I see smaller values after the latest maximum value. I think this is because order of the outputs is not preserved as same as inputs?
If this is so how do I ensure that the order is preserved and I only see the latest maximum value?

2) Another thing I have noticed is that if instance 1 produces the max say -73.8505642 but then instance 2 would produce  -73.9064 which is again smaller than the value produced by instance 1. I'm assuming its because there is no communication between parallel instances hence they produce two value. If this is so how do I get ONE maximum value from all the parallel instances combined? 

Method 2: Using States

public class GetMaximum extends RichMapFunction<Fish, Fish> {

    private transient ValueState<Fish> max;

    @Override
    public Fish map(Fish input) throws Exception {

        // access the state value
       Fish currentMaximum = max.value();

        if (input.point.getX() > currentMaximum.point.getX()) {
            currentMaximum.objid = input.objid;
            currentMaximum.point = (org.locationtech.jts.geom.Point) input.point.clone();
            max.update(currentMaximum);
        }
        return currentMaximum;
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Fish> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Fish>() {}), // type information
                        new Fish(-100,0)); // default value of the state, if nothing was set
        max = getRuntimeContext().getState(descriptor);
    }
}

3) I'm getting the same results as method 1. isn't ValueState shared between all instances of the same operator? 

4) Out of the two methodologies which is a better choice?

Would really appreciate your help!

Best Regards,
Komal
Komal Mariam
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Finding the Maximum Value Received so far in a Stream

Komal Mariam
Hi Jörn!

Thanks for your suggestions. 

Btw just a correction in the Fish class it's "public Point point" not  "public Point coordinate".

For double type comparison, I implemented what you suggested and used BigDecimal for their comparison. I'm still getting the same results where I see smaller values after the latest maximum value. 

I would prefer using the max method but I need to do max on the x-value inside the Point object contained inside the Fish object. (I retrieve it using fish.point.getX();)

The max method on keyed stream works by providing the field name in the form of a string  keyedStream.max("point")  so is there a way I can reference the x coordinate inside the point object to max on?
Is there a way I can implement  keyedStream.max(fish.point.getX())?

About your reduce function:
You execute it by fish_id if I see it correctly. This will create one result by fish_id . I propose to map first all fish coordinates under a single key and then reduce by this single key. 

^Will try this next. 

Regards,
Komal





On Thu, 3 Oct 2019 at 16:42, Jörn Franke <[hidden email]> wrote:
Btw. Why don’t you use the max method?  


See here about the state solution:


About your reduce function:
You execute it by fish_id if I see it correctly. This will create one result by fish_id . I propose to map first all fish coordinates under a single key and then reduce by this single key. 

Am 03.10.2019 um 08:26 schrieb Komal Mariam <[hidden email]>:


Hello all,
I'm trying to do a fairly simple task that is to find the maximum value (Double) received so far in a stream. This is what I implemented:

POJO class:
public class Fish{
    public Integer fish_id;
    public Point coordinate;   //position
  
public Fish() {};

public Fish(fish_id,double x, double y) {
   //assign to fish object
}

Java_main.java
 DataStream text  = env
                .addSource(new FlinkKafkaConsumer<>("test", new JSONKeyValueDeserializationSchema(false),properties).setStartFromLatest());

      DataStream<Fish> fishes = text.map(new MapFunction<ObjectNode, Fish>() {
            @Override
            public Fish map(ObjectNode json) throws Exception {
               Fish fishes = new Fish( json.get("value").get("id").asInteger() ,json.get("value").get("x_begin").asDouble(),json.get("value").get("y_begin").asDouble());
                return fishes;
            }
        });


 I can't seem to apply aggregations on the Point class without extracting the x coordinates in a separate stream so here are the 2 methods I have applied:

Method 1: Simple reduce

KeyedStream<Fish, Integer> keyed = fishes.keyBy(s->s.fish_id);
        keyed.reduce(new ReduceFunction<Fish>() {
            @Override
            public Fish reduce(Fish t, Fish t1) throws Exception {
                if (t.X > t1.X) {
                    return t;
                } else
                    return t1;
            }
        }).map(new MapFunction<Fish, Double>() {
                    @Override
                    public Double map(Fish t) throws Exception {
                        return t.point.getX();
                    }
                }).print();


Result: 1> -73.8517632
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.851446       //smaller than previous value!
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.8517632
1> -73.851446
3> -73.85012          
3> -73.850212
3> -73.851979     //smaller than previous value
3> -73.850212
3> -73.8512969
3> -73.8512969  

1) I'm trying to compute the max so why do I see smaller values after the latest maximum value. I think this is because order of the outputs is not preserved as same as inputs?
If this is so how do I ensure that the order is preserved and I only see the latest maximum value?

2) Another thing I have noticed is that if instance 1 produces the max say -73.8505642 but then instance 2 would produce  -73.9064 which is again smaller than the value produced by instance 1. I'm assuming its because there is no communication between parallel instances hence they produce two value. If this is so how do I get ONE maximum value from all the parallel instances combined? 

Method 2: Using States

public class GetMaximum extends RichMapFunction<Fish, Fish> {

    private transient ValueState<Fish> max;

    @Override
    public Fish map(Fish input) throws Exception {

        // access the state value
       Fish currentMaximum = max.value();

        if (input.point.getX() > currentMaximum.point.getX()) {
            currentMaximum.objid = input.objid;
            currentMaximum.point = (org.locationtech.jts.geom.Point) input.point.clone();
            max.update(currentMaximum);
        }
        return currentMaximum;
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Fish> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Fish>() {}), // type information
                        new Fish(-100,0)); // default value of the state, if nothing was set
        max = getRuntimeContext().getState(descriptor);
    }
}

3) I'm getting the same results as method 1. isn't ValueState shared between all instances of the same operator? 

4) Out of the two methodologies which is a better choice?

Would really appreciate your help!

Best Regards,
Komal
Theo
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Finding the Maximum Value Received so far in a Stream

Theo
Hi Komal,

regarding using max Method: You can call .map() on your stream and convert the POJO to another stream/type, e.g. having only the x coordinate of the POJO and then apply the max operator.

And as the others said: You are working on a keyed stream per fish_id, so you will get one maximum per fish-id. If you want to have a global maximum, you should not key by fish_id first.

Best regards
Theo

Von: "Komal Mariam" <[hidden email]>
An: "Jörn Franke" <[hidden email]>
CC: "user" <[hidden email]>
Gesendet: Donnerstag, 3. Oktober 2019 11:01:54
Betreff: Re: Finding the Maximum Value Received so far in a Stream

Hi Jörn!

Thanks for your suggestions. 

Btw just a correction in the Fish class it's "public Point point" not  "public Point coordinate".

For double type comparison, I implemented what you suggested and used BigDecimal for their comparison. I'm still getting the same results where I see smaller values after the latest maximum value. 

I would prefer using the max method but I need to do max on the x-value inside the Point object contained inside the Fish object. (I retrieve it using fish.point.getX();)

The max method on keyed stream works by providing the field name in the form of a string  keyedStream.max("point")  so is there a way I can reference the x coordinate inside the point object to max on?
Is there a way I can implement  keyedStream.max(fish.point.getX())?

About your reduce function:
You execute it by fish_id if I see it correctly. This will create one result by fish_id . I propose to map first all fish coordinates under a single key and then reduce by this single key. 

^Will try this next. 

Regards,
Komal





On Thu, 3 Oct 2019 at 16:42, Jörn Franke <[hidden email]> wrote:
Btw. Why don’t you use the max method?  


See here about the state solution:


About your reduce function:
You execute it by fish_id if I see it correctly. This will create one result by fish_id . I propose to map first all fish coordinates under a single key and then reduce by this single key. 

Am 03.10.2019 um 08:26 schrieb Komal Mariam <[hidden email]>:

Hello all,
I'm trying to do a fairly simple task that is to find the maximum value (Double) received so far in a stream. This is what I implemented:

POJO class:
public class Fish{
    public Integer fish_id;
    public Point coordinate;   //position
  
public Fish() {};

public Fish(fish_id,double x, double y) {
   //assign to fish object
}

Java_main.java
 DataStream text  = env
                .addSource(new FlinkKafkaConsumer<>("test", new JSONKeyValueDeserializationSchema(false),properties).setStartFromLatest());

      DataStream<Fish> fishes = text.map(new MapFunction<ObjectNode, Fish>() {
            @Override
            public Fish map(ObjectNode json) throws Exception {
               Fish fishes = new Fish( json.get("value").get("id").asInteger() ,json.get("value").get("x_begin").asDouble(),json.get("value").get("y_begin").asDouble());
                return fishes;
            }
        });


 I can't seem to apply aggregations on the Point class without extracting the x coordinates in a separate stream so here are the 2 methods I have applied:

Method 1: Simple reduce

KeyedStream<Fish, Integer> keyed = fishes.keyBy(s->s.fish_id);
        keyed.reduce(new ReduceFunction<Fish>() {
            @Override
            public Fish reduce(Fish t, Fish t1) throws Exception {
                if (t.X > t1.X) {
                    return t;
                } else
                    return t1;
            }
        }).map(new MapFunction<Fish, Double>() {
                    @Override
                    public Double map(Fish t) throws Exception {
                        return t.point.getX();
                    }
                }).print();


Result: 1> -73.8517632
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.851446       //smaller than previous value!
1> -73.851446
1> -73.851446
1> -73.8505642
1> -73.8517632
1> -73.851446
3> -73.85012          
3> -73.850212
3> -73.851979     //smaller than previous value
3> -73.850212
3> -73.8512969
3> -73.8512969  
1) I'm trying to compute the max so why do I see smaller values after the latest maximum value. I think this is because order of the outputs is not preserved as same as inputs?
If this is so how do I ensure that the order is preserved and I only see the latest maximum value?

2) Another thing I have noticed is that if instance 1 produces the max say -73.8505642 but then instance 2 would produce  -73.9064 which is again smaller than the value produced by instance 1. I'm assuming its because there is no communication between parallel instances hence they produce two value. If this is so how do I get ONE maximum value from all the parallel instances combined? 

Method 2: Using States

public class GetMaximum extends RichMapFunction<Fish, Fish> {

    private transient ValueState<Fish> max;

    @Override
    public Fish map(Fish input) throws Exception {

        // access the state value
       Fish currentMaximum = max.value();

        if (input.point.getX() > currentMaximum.point.getX()) {
            currentMaximum.objid = input.objid;
            currentMaximum.point = (org.locationtech.jts.geom.Point) input.point.clone();
            max.update(currentMaximum);
        }
        return currentMaximum;
    }

    @Override
    public void open(Configuration config) {
        ValueStateDescriptor<Fish> descriptor =
                new ValueStateDescriptor<>(
                        "average", // the state name
                        TypeInformation.of(new TypeHint<Fish>() {}), // type information
                        new Fish(-100,0)); // default value of the state, if nothing was set
        max = getRuntimeContext().getState(descriptor);
    }
}

3) I'm getting the same results as method 1. isn't ValueState shared between all instances of the same operator? 
4) Out of the two methodologies which is a better choice?

Would really appreciate your help!

Best Regards,
Komal
Komal Mariam
Reply | Threaded
Open this post in threaded view
♦
♦
|

Re: Finding the Maximum Value Received so far in a Stream

Komal Mariam
Thank you for your help all. I understand now and made the changes.

 Since I needed return the entire object that contained the max value of X, I used reduce instead of max.
Free forum by Nabble Edit this page