我如何从 JavaPairRDD 中过滤记录答案

【问题标题】：How can i filter records from JavaPairRDD我如何从 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;
        }
    });
}

}

【问题讨论】：

见stackoverflow.com/questions/26886275/…

标签： java apache-spark

【解决方案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);
   }
});

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

【讨论】：