InvalidHashExtractor { public JavaPairRDD<String, Long> extractInvalidHashes(JavaPairRDD<String, String> hashIdPairs1, JavaPairRDD<String, String> hashIdPairs2, long maxHashBucketSize) { JavaPairRDD<String, Long> hash1Count = hashIdPairs1.mapValues(v-> 1l).reduceByKey((x,y)->x+y); JavaPairRDD<String, Long> hash2Count = hashIdPairs2.mapValues(v-> 1l).reduceByKey((x,y)->x+y); JavaPairRDD<String, Long> hashesSmallerThanBucketSize = hash1Count.cogroup(hash2Count).mapValues(v-> { Long hash1IdCount = v._1().iterator().hasNext() ? v._1().iterator().next(): 0l; Long hash2IdCount = v._2().iterator().hasNext() ? v._2().iterator().next(): 0l; return hash1IdCount*hash2IdCount; }).filter(x->(x._2()>maxHashBucketSize || x._2()==0)); return hashesSmallerThanBucketSize; }  JavaPairRDD<String, Long> extractInvalidHashes(JavaPairRDD<String, String> hashIdPairs1, JavaPairRDD<String, String> hashIdPairs2, long maxHashBucketSize);  }
@Test public void extractInvalidHashes() { SparkConf conf = new SparkConf().setMaster("local[1]").setAppName("HashBucketLimiterTest"); try (JavaSparkContext sc = new JavaSparkContext(conf)) { @SuppressWarnings("unchecked") List<Tuple2<String, String>> hashes1 = Lists.newArrayList(tuple("1", "x"), tuple("1", "y"), tuple("3", "z"), tuple("1", "x"), tuple("2", "x"), tuple("5", "z")); @SuppressWarnings("unchecked") List<Tuple2<String, String>> hashes2 = Lists.newArrayList(tuple("1", "x"), tuple("1", "a"), tuple("3", "z"), tuple("3", "aa"), tuple("4", "a"), tuple("5", "z")); JavaPairRDD<String, String> hash1Rdd = sc.parallelizePairs(hashes1); JavaPairRDD<String, String> hash2Rdd = sc.parallelizePairs(hashes2); JavaPairRDD<String, Long> invalidHashes = invalidHashExtractor.extractInvalidHashes(hash1Rdd, hash2Rdd, 2); assertEquals(3, invalidHashes.count()); checkContains(invalidHashes, "1", 6); checkContains(invalidHashes, "2", 0); checkContains(invalidHashes, "4", 0); } }