What is the best way to keep a HashMap into the state in Flink

user_6456107 0
Viewed 1060

I have a Flink job, and one of the statefull operators needs to keep into the state a class that contains a HashMap as attribute, because this hasMap keeps different affinities for a user, example:

public class Affinity {
public String id;
public String colorTriggered;
public Map<String,Integer> affinities;
/*this object keeps the affinity for a user to a different colors for example: 
affinities.put(green, 5);
affinities.put(blue, 9);
affinities.put(white, 2);

to calculate then what is the color's affinity of this user, in this case the answer will be blue
*/
}

This hashMap is used to track those affinities and in certain moment ask for the color's affinity of a user and get the key of the highest affinity value which will be blue that the value 9.

As hashMaps are not part of Flink serialization I will need to include implement Serializable to my class.

Is that a bad idea or there is a better way to do this and keep the object into the states?

In a full example more or less what I need to do but not sure if using HashMap into a Flink operator and into states is a good idea:

public class AffinityFlatMapFunction extends RichFlatMapFunction<Event, Affinity> implements MapOperations {

  @Override
  public void flatMap(Event event, Collector<Affinity> collector) throws Exception {
   Affinity previous = state.value();
    if(previous.hashMap.contains(event.color)){
        previous.hashMap.replace(event.color, value + 1);
    }else previous.hashMap.put(event.color, 1);
   /*something like this*/
  String match = previous.hashMap.stream.filter(x -> 
              x.getKey().contains(event.color)).max(Map.Entry.comparingByValue())
                .map(Map.Entry::getKey).orElse("empty");
   if(!match.equals(previous.colorTriggered){
       previous.colorTriggered = match;
       state.update(previous);
       collector.collect(previous);
   }
 }
}

Kind regards!

flink-cep flink-streaming apache-flink java
1 Answers

According to the documentation, there is a state construct called MapState<UK, UV>, which does following:

MapState<UK, UV>: This keeps a list of mappings. You can put key-value pairs into the state and retrieve an Iterable over all currently stored mappings. Mappings are added using put(UK, UV) or putAll(Map<UK, UV>). The value associated with a user key can be retrieved using get(UK). The iterable views for mappings, keys and values can be retrieved using entries(), keys() and values() respectively. You can also use isEmpty() to check whether this map contains any key-value mappings.

I once read in a Flink thread a few days back that the provided StateDescriptors are optimized and almost always the preferred choice instead of implementing an own mechanism.

If you don't shard your stream in an unwanted way (using keyBy(color) should be fine), you should always have the most current state of your Map. I don't know if your concern about RocksDB latency is valid, since Flink state is kept on heap and only checkpointed to a RocksDB, so all current values are available on the fly; but I may have misunderstood that. In retrospect I even doubt you need a map, but a simple ValueState to hold your integer, as the "key" part of the map is taken care of by keyBy() in that case.

user_60975600
Related