DocOrgRelationAffOrgJoiner implements AffOrgJoiner { @Override public JavaRDD<Tuple2<AffMatchAffiliation, AffMatchOrganization>> join(JavaRDD<AffMatchAffiliation> affiliations, JavaRDD<AffMatchOrganization> organizations) { JavaPairRDD<String, AffMatchAffiliation> affiliationsDocIdKey = affiliations.keyBy(aff -> aff.getDocumentId()); JavaPairRDD<String, AffMatchOrganization> organizationsOrgIdKey = organizations.keyBy(org -> org.getId()); JavaRDD<AffMatchDocumentOrganization> documentOrganizations = documentOrganizationFetcher.fetchDocumentOrganizations(); JavaPairRDD<String, AffMatchDocumentOrganization> documentOrganizationDocIdKey = documentOrganizations.keyBy(docOrg -> docOrg.getDocumentId()); JavaRDD<Tuple2<AffMatchAffiliation, AffMatchOrganization>> affOrgBucketPairs = affiliationsDocIdKey .join(documentOrganizationDocIdKey) .mapToPair(x -> new Tuple2<String, AffMatchAffiliation>(x._2._2.getOrganizationId(), x._2._1)) .join(organizationsOrgIdKey) .values(); return affOrgBucketPairs; }  @Override JavaRDD<Tuple2<AffMatchAffiliation, AffMatchOrganization>> join(JavaRDD<AffMatchAffiliation> affiliations, JavaRDD<AffMatchOrganization> organizations); void setDocumentOrganizationFetcher(DocumentOrganizationFetcher documentOrganizationFetcher);  }
@Test public void join() { JavaRDD<AffMatchAffiliation> affiliations = sparkContext.parallelize(ImmutableList.of( new AffMatchAffiliation("DOC1", 1), new AffMatchAffiliation("DOC1", 2), new AffMatchAffiliation("DOC2", 1), new AffMatchAffiliation("DOC3", 1), new AffMatchAffiliation("DOC3", 2), new AffMatchAffiliation("DOC3", 3))); JavaRDD<AffMatchOrganization> organizations = sparkContext.parallelize(ImmutableList.of( new AffMatchOrganization("ORG1"), new AffMatchOrganization("ORG2"), new AffMatchOrganization("ORG3"), new AffMatchOrganization("ORG4"))); JavaRDD<AffMatchDocumentOrganization> documentOrganizations = sparkContext.parallelize(ImmutableList.of( new AffMatchDocumentOrganization("DOC1", "ORG2"), new AffMatchDocumentOrganization("DOC3", "ORG3"), new AffMatchDocumentOrganization("DOC3", "ORG4"), new AffMatchDocumentOrganization("DOC3", "ORG5"), new AffMatchDocumentOrganization("DOC4", "ORG2"))); when(documentOrganizationFetcher.fetchDocumentOrganizations()).thenReturn(documentOrganizations); JavaRDD<Tuple2<AffMatchAffiliation, AffMatchOrganization>> affOrgPairsRDD = docOrgRelationAffOrgJoiner.join(affiliations, organizations); List<Tuple2<AffMatchAffiliation, AffMatchOrganization>> affOrgPairs = affOrgPairsRDD.collect(); List<Tuple2<AffMatchAffiliation, AffMatchOrganization>> expectedAffOrgPairs = ImmutableList.of( new Tuple2<>(new AffMatchAffiliation("DOC1", 1), new AffMatchOrganization("ORG2")), new Tuple2<>(new AffMatchAffiliation("DOC1", 2), new AffMatchOrganization("ORG2")), new Tuple2<>(new AffMatchAffiliation("DOC3", 1), new AffMatchOrganization("ORG3")), new Tuple2<>(new AffMatchAffiliation("DOC3", 2), new AffMatchOrganization("ORG3")), new Tuple2<>(new AffMatchAffiliation("DOC3", 3), new AffMatchOrganization("ORG3")), new Tuple2<>(new AffMatchAffiliation("DOC3", 1), new AffMatchOrganization("ORG4")), new Tuple2<>(new AffMatchAffiliation("DOC3", 2), new AffMatchOrganization("ORG4")), new Tuple2<>(new AffMatchAffiliation("DOC3", 3), new AffMatchOrganization("ORG4"))); assertThat(affOrgPairs, containsInAnyOrder(expectedAffOrgPairs.toArray())); }