IisAffMatchResultWriter implements AffMatchResultWriter { @Override public void write(JavaSparkContext sc, JavaRDD<AffMatchResult> matchedAffOrgs, String outputPath, String outputReportPath, int numberOfEmittedFiles) { Preconditions.checkNotNull(sc); Preconditions.checkNotNull(matchedAffOrgs); Preconditions.checkArgument(StringUtils.isNotBlank(outputPath)); Preconditions.checkArgument(StringUtils.isNotBlank(outputReportPath)); JavaRDD<MatchedOrganization> matchedOrganizations = matchedAffOrgs.map(affOrgMatch -> affMatchResultConverter.convert(affOrgMatch)); JavaRDD<MatchedOrganization> distinctMatchedOrganizations = matchedOrganizations .keyBy(match -> new Tuple2<>(match.getDocumentId(), match.getOrganizationId())) .reduceByKey((match1, match2) -> matchedOrgStrengthRecalculation.recalculateStrength(match1, match2)) .values(); distinctMatchedOrganizations.cache(); sparkAvroSaver.saveJavaRDD(distinctMatchedOrganizations.repartition(numberOfEmittedFiles), MatchedOrganization.SCHEMA$, outputPath); List<ReportEntry> reportEntries = reportGenerator.generateReport(distinctMatchedOrganizations); sparkAvroSaver.saveJavaRDD(sc.parallelize(reportEntries, 1), ReportEntry.SCHEMA$, outputReportPath); }  @Override void write(JavaSparkContext sc, JavaRDD<AffMatchResult> matchedAffOrgs, String outputPath, String outputReportPath, int numberOfEmittedFiles);  }
@Test(expected = NullPointerException.class) public void write_sc_null() { writer.write(null, affMatchResults, "/output", "/report", 1); }
@Test(expected = NullPointerException.class) public void write_matchedAffOrgs_null() { writer.write(sc, null, "/output", "/report", 1); }
@Test(expected = IllegalArgumentException.class) public void write_outputPath_blank() { writer.write(sc, affMatchResults, " ", "/report", 1); }
@Test(expected = IllegalArgumentException.class) public void write_outputReportPath_blank() { writer.write(sc, affMatchResults, "/output", " ", 1); }
@Test public void write() throws Exception { String outputPath = "/data/matchedAffiliations/output"; String outputReportPath = "/data/matchedAffiliations/report"; doReturn(matchedOrganizations).when(affMatchResults).map(any()); doReturn(matchedOrganizationsDocOrgIdKey).when(matchedOrganizations).keyBy(any()); when(matchedOrganizationsDocOrgIdKey.reduceByKey(any())).thenReturn(distinctMatchedOrganizations); when(distinctMatchedOrganizations.values()).thenReturn(distinctMatchedOrganizationsValues); when(distinctMatchedOrganizationsValues.repartition(2)).thenReturn(distinctMatchedOrganizationsValuesRepartition); when(reportGenerator.generateReport(distinctMatchedOrganizationsValues)).thenReturn(reportEntries); when(sc.parallelize(reportEntries, 1)).thenReturn(rddReportEntries); writer.write(sc, affMatchResults, outputPath, outputReportPath, 2); verify(sparkAvroSaver).saveJavaRDD(distinctMatchedOrganizationsValuesRepartition, MatchedOrganization.SCHEMA$, outputPath); verify(sparkAvroSaver).saveJavaRDD(rddReportEntries, ReportEntry.SCHEMA$, outputReportPath); verify(affMatchResults).map(convertFunction.capture()); assertConvertFunction(convertFunction.getValue()); verify(matchedOrganizations).keyBy(extractDocOrgIdFunction.capture()); assertExtractDocOrgIdFunction(extractDocOrgIdFunction.getValue()); verify(matchedOrganizationsDocOrgIdKey).reduceByKey(duplicateMatchedOrgsReduceFunction.capture()); assertDuplicateMatchedOrgsReduceFunction(duplicateMatchedOrgsReduceFunction.getValue()); verify(distinctMatchedOrganizations).values(); }