PatentExportCounterReporter { public void report(JavaSparkContext sc, JavaRDD<DocumentToPatentWithIdsToExport> documentToPatentsToExportWithIds, String outputReportPath) { Preconditions.checkNotNull(sc, "sparkContext has not been set"); Preconditions.checkNotNull(outputReportPath, "reportPath has not been set"); ReportEntry totalRelationsCounter = ReportEntryFactory.createCounterReportEntry(PATENT_REFERENCES_COUNTER, totalRelationsCount(documentToPatentsToExportWithIds)); ReportEntry totalEntitiesCounter = ReportEntryFactory.createCounterReportEntry(EXPORTED_PATENT_ENTITIES_COUNTER, totalEntitiesCount(documentToPatentsToExportWithIds)); ReportEntry distinctPublicationsCounter = ReportEntryFactory.createCounterReportEntry(DISTINCT_PUBLICATIONS_WITH_PATENT_REFERENCES_COUNTER, distinctPublicationsCount(documentToPatentsToExportWithIds)); JavaRDD<ReportEntry> report = sc.parallelize(Arrays.asList( totalRelationsCounter, totalEntitiesCounter, distinctPublicationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, outputReportPath); }  void report(JavaSparkContext sc,
                       JavaRDD<DocumentToPatentWithIdsToExport> documentToPatentsToExportWithIds,
                       String outputReportPath); static final String PATENT_REFERENCES_COUNTER; static final String EXPORTED_PATENT_ENTITIES_COUNTER; static final String DISTINCT_PUBLICATIONS_WITH_PATENT_REFERENCES_COUNTER; }
@Test(expected = NullPointerException.class) public void reportShouldThrowExceptionWhenSparkContextIsNull() { JavaRDD<DocumentToPatentWithIdsToExport> documentToPatentsToExportWithIds = sc.emptyRDD(); reporter.report(null, documentToPatentsToExportWithIds, outputReportPath); }
@Test(expected = NullPointerException.class) public void reportShouldThrowExceptionWhenOutputReportPathIsNull() { JavaRDD<DocumentToPatentWithIdsToExport> documentToPatentsToExportWithIds = sc.emptyRDD(); reporter.report(sc, documentToPatentsToExportWithIds, null); }
@Test public void reportShouldCreateAndSaveReportAsAvroDatastoreOfReportEntries() { List<DocumentToPatent> documentToPatents = Arrays.asList( DocumentToPatent.newBuilder().setDocumentId("d1").setApplnNr("p1").setConfidenceLevel(0.9f).build(), DocumentToPatent.newBuilder().setDocumentId("d1").setApplnNr("p2").setConfidenceLevel(0.9f).build(), DocumentToPatent.newBuilder().setDocumentId("d2").setApplnNr("p2").setConfidenceLevel(0.9f).build() ); JavaRDD<DocumentToPatentWithIdsToExport> documentToPatentsToExportWithIds = sc.parallelize(documentToPatents) .map(x -> new DocumentToPatentWithIdsToExport(x, String.format("export_%s", x.getDocumentId()), String.format("export_%s", x.getApplnNr()))); reporter.report(sc, documentToPatentsToExportWithIds, outputReportPath); verify(avroSaver, times(1)).saveJavaRDD(report.capture(), eq(ReportEntry.SCHEMA$), eq(outputReportPath)); List<String> actualReportEntriesJson = report.getValue() .map(SpecificRecordBase::toString).collect() .stream().sorted().collect(Collectors.toList()); List<ReportEntry> expectedReportEntries = Arrays .asList( ReportEntryFactory.createCounterReportEntry(PatentExportCounterReporter.PATENT_REFERENCES_COUNTER, 3), ReportEntryFactory.createCounterReportEntry(PatentExportCounterReporter.EXPORTED_PATENT_ENTITIES_COUNTER, 2), ReportEntryFactory.createCounterReportEntry(PatentExportCounterReporter.DISTINCT_PUBLICATIONS_WITH_PATENT_REFERENCES_COUNTER, 2) ); List<String> expectedReportEntriesJson = expectedReportEntries.stream() .map(SpecificRecordBase::toString) .sorted() .collect(Collectors.toList()); ListTestUtils .compareLists(actualReportEntriesJson, expectedReportEntriesJson); }