@Test public void merge() { JavaRDD<AffMatchDocumentProject> firstDocumentProjects = sparkContext.parallelize(of( new AffMatchDocumentProject("DOC1", "PROJ1", 1f), new AffMatchDocumentProject("DOC1", "PROJ2", 0.6f), new AffMatchDocumentProject("DOC1", "PROJ3", 0.4f), new AffMatchDocumentProject("DOC2", "PROJ4", 0.8f))); JavaRDD<AffMatchDocumentProject> secondDocumentProjects = sparkContext.parallelize(of( new AffMatchDocumentProject("DOC1", "PROJ1", 0.3f), new AffMatchDocumentProject("DOC1", "PROJ2", 1f), new AffMatchDocumentProject("DOC1", "PROJ4", 0.7f))); JavaRDD<AffMatchDocumentProject> retDocumentProjects = documentProjectMerger .merge(firstDocumentProjects, secondDocumentProjects); List<AffMatchDocumentProject> expectedDocumentProjects = of( new AffMatchDocumentProject("DOC1", "PROJ1", 1f), new AffMatchDocumentProject("DOC1", "PROJ2", 1f), new AffMatchDocumentProject("DOC1", "PROJ3", 0.4f), new AffMatchDocumentProject("DOC1", "PROJ4", 0.7f), new AffMatchDocumentProject("DOC2", "PROJ4", 0.8f)); assertThat(retDocumentProjects.collect(), containsInAnyOrder(expectedDocumentProjects.toArray())); }
public JavaRDD<AffMatchDocumentProject> merge(JavaRDD<AffMatchDocumentProject> firstDocumentProjects, JavaRDD<AffMatchDocumentProject> secondDocumentProjects) { JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> firstDocProjWithKey = firstDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> secondDocProjWithKey = secondDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); return firstDocProjWithKey .union(secondDocProjWithKey) .reduceByKey((x, y) -> x.getConfidenceLevel() >= y.getConfidenceLevel() ? x : y) .values(); }
DocumentProjectMerger implements Serializable { public JavaRDD<AffMatchDocumentProject> merge(JavaRDD<AffMatchDocumentProject> firstDocumentProjects, JavaRDD<AffMatchDocumentProject> secondDocumentProjects) { JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> firstDocProjWithKey = firstDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> secondDocProjWithKey = secondDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); return firstDocProjWithKey .union(secondDocProjWithKey) .reduceByKey((x, y) -> x.getConfidenceLevel() >= y.getConfidenceLevel() ? x : y) .values(); } }
DocumentProjectMerger implements Serializable { public JavaRDD<AffMatchDocumentProject> merge(JavaRDD<AffMatchDocumentProject> firstDocumentProjects, JavaRDD<AffMatchDocumentProject> secondDocumentProjects) { JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> firstDocProjWithKey = firstDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> secondDocProjWithKey = secondDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); return firstDocProjWithKey .union(secondDocProjWithKey) .reduceByKey((x, y) -> x.getConfidenceLevel() >= y.getConfidenceLevel() ? x : y) .values(); }  }
DocumentProjectMerger implements Serializable { public JavaRDD<AffMatchDocumentProject> merge(JavaRDD<AffMatchDocumentProject> firstDocumentProjects, JavaRDD<AffMatchDocumentProject> secondDocumentProjects) { JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> firstDocProjWithKey = firstDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> secondDocProjWithKey = secondDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); return firstDocProjWithKey .union(secondDocProjWithKey) .reduceByKey((x, y) -> x.getConfidenceLevel() >= y.getConfidenceLevel() ? x : y) .values(); }  JavaRDD<AffMatchDocumentProject> merge(JavaRDD<AffMatchDocumentProject> firstDocumentProjects, JavaRDD<AffMatchDocumentProject> secondDocumentProjects); }
DocumentProjectMerger implements Serializable { public JavaRDD<AffMatchDocumentProject> merge(JavaRDD<AffMatchDocumentProject> firstDocumentProjects, JavaRDD<AffMatchDocumentProject> secondDocumentProjects) { JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> firstDocProjWithKey = firstDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); JavaPairRDD<Tuple2<String, String>, AffMatchDocumentProject> secondDocProjWithKey = secondDocumentProjects .keyBy(x -> new Tuple2<>(x.getDocumentId(), x.getProjectId())); return firstDocProjWithKey .union(secondDocProjWithKey) .reduceByKey((x, y) -> x.getConfidenceLevel() >= y.getConfidenceLevel() ? x : y) .values(); }  JavaRDD<AffMatchDocumentProject> merge(JavaRDD<AffMatchDocumentProject> firstDocumentProjects, JavaRDD<AffMatchDocumentProject> secondDocumentProjects);  }