@Test(expected = NullPointerException.class) public void write_matchedAffOrgs_null() { writer.write(sc, null, "/output", "/report", 1); }
@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); }
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); } }
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); }  }
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); }
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);  }