@Test public void readAffiliations() throws Exception { String inputPath = "/data/affiliations"; when(sparkAvroLoader.loadJavaRDD(sparkContext, inputPath, ExtractedDocumentMetadata.class)).thenReturn(inputDocuments); doReturn(affMatchAffiliations).when(inputDocuments).flatMap(any()); JavaRDD<AffMatchAffiliation> retAffMatchAffiliations = reader.readAffiliations(sparkContext, inputPath); assertTrue(affMatchAffiliations == retAffMatchAffiliations); verify(inputDocuments).flatMap(convertFunction.capture()); assertConvertFunction(convertFunction.getValue()); }
@Override public JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath) { Preconditions.checkNotNull(sc); Preconditions.checkArgument(StringUtils.isNotBlank(inputPath)); JavaRDD<ExtractedDocumentMetadata> sourceAffiliations = sparkAvroLoader .loadJavaRDD(sc, inputPath, ExtractedDocumentMetadata.class); return sourceAffiliations.flatMap(srcAff -> affiliationConverter.convert(srcAff).iterator()); }
IisAffiliationReader implements Serializable, AffiliationReader { @Override public JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath) { Preconditions.checkNotNull(sc); Preconditions.checkArgument(StringUtils.isNotBlank(inputPath)); JavaRDD<ExtractedDocumentMetadata> sourceAffiliations = sparkAvroLoader .loadJavaRDD(sc, inputPath, ExtractedDocumentMetadata.class); return sourceAffiliations.flatMap(srcAff -> affiliationConverter.convert(srcAff).iterator()); } }
IisAffiliationReader implements Serializable, AffiliationReader { @Override public JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath) { Preconditions.checkNotNull(sc); Preconditions.checkArgument(StringUtils.isNotBlank(inputPath)); JavaRDD<ExtractedDocumentMetadata> sourceAffiliations = sparkAvroLoader .loadJavaRDD(sc, inputPath, ExtractedDocumentMetadata.class); return sourceAffiliations.flatMap(srcAff -> affiliationConverter.convert(srcAff).iterator()); }  }
IisAffiliationReader implements Serializable, AffiliationReader { @Override public JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath) { Preconditions.checkNotNull(sc); Preconditions.checkArgument(StringUtils.isNotBlank(inputPath)); JavaRDD<ExtractedDocumentMetadata> sourceAffiliations = sparkAvroLoader .loadJavaRDD(sc, inputPath, ExtractedDocumentMetadata.class); return sourceAffiliations.flatMap(srcAff -> affiliationConverter.convert(srcAff).iterator()); }  @Override JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath); }
IisAffiliationReader implements Serializable, AffiliationReader { @Override public JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath) { Preconditions.checkNotNull(sc); Preconditions.checkArgument(StringUtils.isNotBlank(inputPath)); JavaRDD<ExtractedDocumentMetadata> sourceAffiliations = sparkAvroLoader .loadJavaRDD(sc, inputPath, ExtractedDocumentMetadata.class); return sourceAffiliations.flatMap(srcAff -> affiliationConverter.convert(srcAff).iterator()); }  @Override JavaRDD<AffMatchAffiliation> readAffiliations(JavaSparkContext sc, String inputPath);  }