我如何从 JavaPairRDD 中过滤记录

Posted 2023-04-15

【中文标题】我如何从 JavaPairRDD 中过滤记录

【英文标题】：How can i filter records from JavaPairRDD

【发布时间】：2016-02-09 16:14:10

【问题描述】：

我正在 Apache spark 中做 WordCount 的一个简单示例，现在我终于得到了字数计数我只想从中过滤唯一字。

public class SparkClass 
    public static void main(String[] args) 

    String file = "/home/bhaumik/Documents/my";
    JavaSparkContext sc = new JavaSparkContext("local", "SimpleApp");
    JavaRDD<String> lines = sc.textFile("/home/bhaumik/Documents/myText", 5)
            .flatMap(new FlatMapFunction<String, String>() 

                @Override
                public Iterable<String> call(String t) throws Exception 
                    // TODO Auto-generated method stub
                    return Arrays.asList(t.split(" "));
                
            );

    JavaPairRDD<String, Integer> pairs = lines.mapToPair(new PairFunction<String, String, Integer>() 

        @Override
        public Tuple2<String, Integer> call(String t) throws Exception 
            // TODO Auto-generated method stub
            return new Tuple2<String, Integer>(t, 1);
        
    );

    JavaPairRDD<String, Integer> counts = pairs.reduceByKey(new Function2<Integer, Integer, Integer>() 

        @Override
        public Integer call(Integer v1, Integer v2) throws Exception 
            // TODO Auto-generated method stub
            return v1 + v2;
        
    );

【问题讨论】：

见***.com/questions/26886275/…

【参考方案1】：

在 counts 中，您有一个带有键及其出现次数的 RDD。你现在不能得到最小值，所以你应该减少

Tuple2<String, Integer> minApp = counts.reduce((a, b) -> (a._2 > b._2)? b : a);

【讨论】：

无法将其放入 JavaPairRDD 或者有什么方法可以将 Tuple2 转换为 JavaPairRDD ??

在这种情况下它没有多大意义。其他选项可能是对整个 RDD 进行排序并获取第一个元素

实际上我在我的日志文件上做同样的事情，我从中获取 IP，然后现在我想要唯一的 IP，为什么我问你？ :(

感谢您的帮助和努力。

【参考方案2】：

JavaPairRDD<String,Integer> uniqueIP = counts.filter(newFunction<Tuple2<String,Integer>,Boolean()>
   @Override
   public Boolean call(Tuple<String, Integer> v1) throws Exception 
   return v1._2.equals(1);
   
);

这就是我解决问题的方法...