@Test public void report() throws Exception { JavaSparkContext sparkContext = mock(JavaSparkContext.class); String reportPath = "/report/path"; when(matchedCitations.count()).thenReturn(14L); doReturn(matchedCitationsDocumentIds).when(matchedCitations).map(any()); when(matchedCitationsDocumentIds.distinct()).thenReturn(matchedCitationsDistinctDocumentIds); when(matchedCitationsDistinctDocumentIds.count()).thenReturn(3L); doReturn(reportCounters).when(sparkContext).parallelize(any(), eq(1)); counterReporter.report(sparkContext, matchedCitations, reportPath); verify(matchedCitations).map(extractDocIdFunction.capture()); assertExtractDocIdFunction(extractDocIdFunction.getValue()); verify(sparkContext).parallelize(reportEntriesCaptor.capture(), eq(1)); assertReportEntries(reportEntriesCaptor.getValue()); verify(avroSaver).saveJavaRDD(reportCounters, ReportEntry.SCHEMA$, reportPath); }
public void report(JavaSparkContext sparkContext, JavaRDD<Citation> citations, String outputReportPath) { ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(citations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(citations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, outputReportPath); }
CitationMatchingDirectCounterReporter { public void report(JavaSparkContext sparkContext, JavaRDD<Citation> citations, String outputReportPath) { ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(citations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(citations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, outputReportPath); } }
CitationMatchingDirectCounterReporter { public void report(JavaSparkContext sparkContext, JavaRDD<Citation> citations, String outputReportPath) { ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(citations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(citations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, outputReportPath); }  }
CitationMatchingDirectCounterReporter { public void report(JavaSparkContext sparkContext, JavaRDD<Citation> citations, String outputReportPath) { ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(citations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(citations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, outputReportPath); }  void report(JavaSparkContext sparkContext, JavaRDD<Citation> citations, String outputReportPath); }
CitationMatchingDirectCounterReporter { public void report(JavaSparkContext sparkContext, JavaRDD<Citation> citations, String outputReportPath) { ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(citations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(citations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, outputReportPath); }  void report(JavaSparkContext sparkContext, JavaRDD<Citation> citations, String outputReportPath);  }