Accessing HashMap inside flatMapToPair

Edit: Already solved using RDD.collectAsMap()

I am trying to replicate the solution to the problem from pages 28-30 of http://on-demand.gputechconf.com/gtc/2016/presentation/S6424-michela-taufer-apache-spark.pdf

I have a HashMap<Integer, Integer> that I instantiate outside of the map function. The HashMap contains the following data:

{1:2, 2:3, 3:2, 4:2, 5:3}

A previously defined RDD previousRDD was has the type:

JavaPairRDD<Integer, Iterable<Tuple2<Integer, Integer>>>

has the data:

1: [(1,2), (1,5)]
2: [(2,1), (2,3), (2,5)]
3: [(3,2), (3,4)]
4: [(4,3), (4,5)]
5: [(5,1), (5,2), (5,4)]

I try to create a new RDD with a flatMapToPair:

JavaPairRDD<Integer, Integer> newRDD = previousRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<Integer, Iterable<Tuple2<Integer, Integer>>>, Integer, Integer>() {
    @Override
    public Iterator<Tuple2<Integer, Integer>> call(Tuple2<Integer, Iterable<Tuple2<Integer, Integer>>> integerIterableTuple2) throws Exception {
        Integer count;
        ArrayList<Tuple2<Integer, Integer>> list = new ArrayList<>();
        count = hashMap.get(integerIterableTuple2._1);
        for (Tuple2<Integer, Integer> t : integerIterableTuple2._2) {
            Integer tcount = hashMap.get(t._2);
            if (count < tcount || (count.equals(tcount) && integerIterableTuple2._1 < t._2)) {
                list.add(t);
            }
        }
        return list.iterator();
    }
});

But in this, the hashMap.get(t._2) inside the for loop gets NULLs most of the time. I have checked that the proper values are inside the HashMap.

Is there a way to properly get the values of a HashMap inside a Spark function?

It should work. Spark should capture your variable, serialize it and send to each worker with each task. You might try broadcasting this map

sc.broadcast(hashMap)

and use the result instead of hashMap. It is more efficient memory-wise too (shared storage per executor).

I had similar problem with class variables. You can try make your variable local or declare one more, like this:

 Map localMap = hashMap;
 JavaPairRDD<Integer, Integer> newRDD = previousRDD.flatMapToPair(
   ...
      Integer tcount = localMap.get(t._2);
   ...
 );

I think this is due to spark serialization mechanism. You can read more about it here.