@SuppressWarnings("unchecked") @Test public void readDocuments() throws Exception { doReturn(inputRecords).when(sparkContext).newAPIHadoopFile(any(), any(), any(), any(), any()); doReturn(inputDocuments).when(inputRecords).map(any()); doReturn(documents).when(inputDocuments).mapToPair(any()); JavaPairRDD<String, DocumentMetadata> retDocuments = documentMetadataInputReader.readDocuments(sparkContext, "/some/path"); assertTrue(retDocuments == documents); verify(sparkContext).newAPIHadoopFile( eq("/some/path"), eq(AvroKeyInputFormat.class), eq(DocumentMetadata.class), eq(NullWritable.class), isA(Configuration.class)); verify(inputDocuments).mapToPair(attachIdFunction.capture()); assertAttachIdFunction(attachIdFunction.getValue()); }
@Override public JavaPairRDD<String, DocumentMetadata> readDocuments(JavaSparkContext sparkContext, String inputDocumentsPath) { JavaRDD<DocumentMetadata> documents = avroLoader.loadJavaRDD(sparkContext, inputDocumentsPath, DocumentMetadata.class); return documents.mapToPair(doc -> new Tuple2<>(buildDocumentId(doc), doc)); }
DocumentMetadataInputReader implements InputDocumentReader<String, DocumentMetadata>, Serializable { @Override public JavaPairRDD<String, DocumentMetadata> readDocuments(JavaSparkContext sparkContext, String inputDocumentsPath) { JavaRDD<DocumentMetadata> documents = avroLoader.loadJavaRDD(sparkContext, inputDocumentsPath, DocumentMetadata.class); return documents.mapToPair(doc -> new Tuple2<>(buildDocumentId(doc), doc)); } }
DocumentMetadataInputReader implements InputDocumentReader<String, DocumentMetadata>, Serializable { @Override public JavaPairRDD<String, DocumentMetadata> readDocuments(JavaSparkContext sparkContext, String inputDocumentsPath) { JavaRDD<DocumentMetadata> documents = avroLoader.loadJavaRDD(sparkContext, inputDocumentsPath, DocumentMetadata.class); return documents.mapToPair(doc -> new Tuple2<>(buildDocumentId(doc), doc)); }  }
DocumentMetadataInputReader implements InputDocumentReader<String, DocumentMetadata>, Serializable { @Override public JavaPairRDD<String, DocumentMetadata> readDocuments(JavaSparkContext sparkContext, String inputDocumentsPath) { JavaRDD<DocumentMetadata> documents = avroLoader.loadJavaRDD(sparkContext, inputDocumentsPath, DocumentMetadata.class); return documents.mapToPair(doc -> new Tuple2<>(buildDocumentId(doc), doc)); }  @Override JavaPairRDD<String, DocumentMetadata> readDocuments(JavaSparkContext sparkContext, String inputDocumentsPath); }
DocumentMetadataInputReader implements InputDocumentReader<String, DocumentMetadata>, Serializable { @Override public JavaPairRDD<String, DocumentMetadata> readDocuments(JavaSparkContext sparkContext, String inputDocumentsPath) { JavaRDD<DocumentMetadata> documents = avroLoader.loadJavaRDD(sparkContext, inputDocumentsPath, DocumentMetadata.class); return documents.mapToPair(doc -> new Tuple2<>(buildDocumentId(doc), doc)); }  @Override JavaPairRDD<String, DocumentMetadata> readDocuments(JavaSparkContext sparkContext, String inputDocumentsPath);  }