@Test public void report() throws Exception { 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(matchedCitations); 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(JavaRDD<Citation> matchedCitations) { checkState(); ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(matchedCitations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(matchedCitations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, reportPath); }
CitationMatchingCounterReporter { public void report(JavaRDD<Citation> matchedCitations) { checkState(); ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(matchedCitations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(matchedCitations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, reportPath); } }
CitationMatchingCounterReporter { public void report(JavaRDD<Citation> matchedCitations) { checkState(); ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(matchedCitations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(matchedCitations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, reportPath); }  }
CitationMatchingCounterReporter { public void report(JavaRDD<Citation> matchedCitations) { checkState(); ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(matchedCitations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(matchedCitations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, reportPath); }  void report(JavaRDD<Citation> matchedCitations); void setReportPath(String reportPath); void setSparkContext(JavaSparkContext sparkContext); }
CitationMatchingCounterReporter { public void report(JavaRDD<Citation> matchedCitations) { checkState(); ReportEntry matchedCitationsCounter = generateMatchedCitationsCounter(matchedCitations); ReportEntry docsWithMatchedCitationsCounter = generateDocsWithCitationsCounter(matchedCitations); JavaRDD<ReportEntry> report = sparkContext .parallelize(Lists.newArrayList(matchedCitationsCounter, docsWithMatchedCitationsCounter), 1); avroSaver.saveJavaRDD(report, ReportEntry.SCHEMA$, reportPath); }  void report(JavaRDD<Citation> matchedCitations); void setReportPath(String reportPath); void setSparkContext(JavaSparkContext sparkContext);  }