DocumentAttacher { public JavaPairRDD<String, MatchableEntity> attachDocuments(JavaPairRDD<String, String> matchedCitations, JavaPairRDD<String, MatchableEntity> documents) { return matchedCitations .mapToPair(x -> x.swap()) .join(documents) .mapToPair(x -> new Tuple2<String, MatchableEntity>(x._2._1, x._2._2)); }  JavaPairRDD<String, MatchableEntity> attachDocuments(JavaPairRDD<String, String> matchedCitations, JavaPairRDD<String, MatchableEntity> documents);  }
@Test public void attachDocuments() { JavaPairRDD<String, String> matchedCitations = sparkContext.parallelizePairs(ImmutableList.of( new Tuple2<>(citation1.id(), document1.id()), new Tuple2<>(citation1.id(), document2.id()), new Tuple2<>(citation4.id(), document2.id()), new Tuple2<>(citation4.id(), document5.id()))); JavaPairRDD<String, MatchableEntity> documents = sparkContext.parallelizePairs(generateIdWithEntityTuples( Lists.newArrayList(document1, document2, document3, document4, document5))); JavaPairRDD<String, MatchableEntity> actualCitIdDocPairs = documentAttacher.attachDocuments(matchedCitations, documents); List<Tuple2<String, MatchableEntity>> expectedCitIdDocPairs = ImmutableList.of( new Tuple2<>(citation1.id(), document1), new Tuple2<>(citation1.id(), document2), new Tuple2<>(citation4.id(), document2), new Tuple2<>(citation4.id(), document5)); assertDocAttachedMatchedCitations(actualCitIdDocPairs.collect(), expectedCitIdDocPairs); }
@Test public void attachDocuments_EMPTY_DOCS() { JavaPairRDD<String, String> matchedCitations = sparkContext.parallelizePairs(ImmutableList.of( new Tuple2<>(citation1.id(), document1.id()), new Tuple2<>(citation1.id(), document2.id()), new Tuple2<>(citation4.id(), document5.id()))); JavaPairRDD<String, MatchableEntity> documents = sparkContext.parallelizePairs(Lists.newArrayList()); JavaPairRDD<String, MatchableEntity> actualCitIdDocPairs = documentAttacher.attachDocuments(matchedCitations, documents); assertEquals(actualCitIdDocPairs.count(), 0); }