在 flatMapToPair 中访问 HashMap

Posted 2023-03-31

【中文标题】在 flatMapToPair 中访问 HashMap

【英文标题】：Accessing HashMap inside flatMapToPair

【发布时间】：2020-04-10 04:08:28

【问题描述】：

编辑：已经使用RDD.collectAsMap()解决了

我正在尝试从http://on-demand.gputechconf.com/gtc/2016/presentation/S6424-michela-taufer-apache-spark.pdf 的第 28-30 页复制该问题的解决方案

我有一个在 map 函数之外实例化的 HashMap。 HashMap 包含以下数据：

1:2, 2:3, 3:2, 4:2, 5:3

先前定义的 RDD previousRDD 具有以下类型：

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

有数据：

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)]

我尝试用 flatMapToPair 创建一个新的 RDD：

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();
    
);

但在这种情况下，for 循环内的hashMap.get(t._2) 大部分时间都为 NULL。我检查了正确的值在 HashMap 中。

有没有办法在 Spark 函数中正确获取 HashMap 的值？

【问题讨论】：

hashMap 是在方法中声明还是作为类变量声明？

它被声明为一个类变量。

其实我错了。它在 main 方法中声明。

【参考方案1】：

它应该工作。 Spark 应该捕获您的变量，将其序列化并通过每个任务发送给每个工作人员。您可以尝试广播此地图

sc.broadcast(hashMap)

并使用结果而不是hashMap。它在内存方面也更有效（每个执行程序共享存储）。

【讨论】：

我将上述结果保存到广播变量中。但是，当我在 Spark 函数中执行 broadcast.getValue().get(key) 时，每个键（即使是以前工作的键）都返回 null。

【参考方案2】：

我对类变量也有类似的问题。您可以尝试将变量设为本地变量或再声明一个，如下所示：

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

我认为这是由于 spark 序列化机制。你可以阅读更多关于它的信息here。

【讨论】：

这就是我所做的。我像你一样声明 HashMap，在 RDD 上执行forEach，然后填充 HashMap。然后，我用previousRDD 执行flatMapToPair，得到每个键的值。