@Test public void buildDocumentMetadataShouldRunProperly() { AvroDataFrameSupport avroDataFrameSupport = new AvroDataFrameSupport(spark); Dataset<Row> documentTextDF = avroDataFrameSupport.createDataFrame( Arrays.asList( createDocumentText("docId-1", "text-1"), createDocumentText("docId-2", "text-2") ), DocumentText.SCHEMA$); Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF = avroDataFrameSupport.createDataFrame( Arrays.asList( createExtractedDocumentMetadataMergedWithOriginal("docId-1"), createExtractedDocumentMetadataMergedWithOriginal("docId-a") ), ExtractedDocumentMetadataMergedWithOriginal.SCHEMA$); Dataset<Row> resultDF = buildDocumentMetadata(documentTextDF, extractedDocumentMetadataMergedWithOriginalDF); assertEquals(TaraReferenceExtractionJob.DOCUMENT_METADATA_SCHEMA, resultDF.schema()); List<Row> results = resultDF.collectAsList().stream() .sorted(Comparator.comparing(o -> o.getAs("id"))) .collect(Collectors.toList()); assertEquals(2, results.size()); assertForDocumentMetadataRow(results.get(0), "docId-1", null, null, "text-1"); assertForDocumentMetadataRow(results.get(1), "docId-2", null, null, "text-2"); }
public static Dataset<Row> buildDocumentMetadata(Dataset<Row> documentTextDF, Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF) { logger.info("Building document metadata for input data."); Column joinExprs = documentTextDF.col("id").equalTo( extractedDocumentMetadataMergedWithOriginalDF.col("id")); return documentTextDF .join(extractedDocumentMetadataMergedWithOriginalDF, joinExprs, "left_outer") .select( documentTextDF.col("id"), extractedDocumentMetadataMergedWithOriginalDF.col("title"), extractedDocumentMetadataMergedWithOriginalDF.col("abstract"), documentTextDF.col("text") ); }
TaraReferenceExtractionUtils { public static Dataset<Row> buildDocumentMetadata(Dataset<Row> documentTextDF, Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF) { logger.info("Building document metadata for input data."); Column joinExprs = documentTextDF.col("id").equalTo( extractedDocumentMetadataMergedWithOriginalDF.col("id")); return documentTextDF .join(extractedDocumentMetadataMergedWithOriginalDF, joinExprs, "left_outer") .select( documentTextDF.col("id"), extractedDocumentMetadataMergedWithOriginalDF.col("title"), extractedDocumentMetadataMergedWithOriginalDF.col("abstract"), documentTextDF.col("text") ); } }
TaraReferenceExtractionUtils { public static Dataset<Row> buildDocumentMetadata(Dataset<Row> documentTextDF, Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF) { logger.info("Building document metadata for input data."); Column joinExprs = documentTextDF.col("id").equalTo( extractedDocumentMetadataMergedWithOriginalDF.col("id")); return documentTextDF .join(extractedDocumentMetadataMergedWithOriginalDF, joinExprs, "left_outer") .select( documentTextDF.col("id"), extractedDocumentMetadataMergedWithOriginalDF.col("title"), extractedDocumentMetadataMergedWithOriginalDF.col("abstract"), documentTextDF.col("text") ); }  }
TaraReferenceExtractionUtils { public static Dataset<Row> buildDocumentMetadata(Dataset<Row> documentTextDF, Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF) { logger.info("Building document metadata for input data."); Column joinExprs = documentTextDF.col("id").equalTo( extractedDocumentMetadataMergedWithOriginalDF.col("id")); return documentTextDF .join(extractedDocumentMetadataMergedWithOriginalDF, joinExprs, "left_outer") .select( documentTextDF.col("id"), extractedDocumentMetadataMergedWithOriginalDF.col("title"), extractedDocumentMetadataMergedWithOriginalDF.col("abstract"), documentTextDF.col("text") ); }  static Dataset<Row> buildDocumentMetadata(Dataset<Row> documentTextDF,
                                                     Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF); static Dataset<Row> runReferenceExtraction(SparkSession spark,
                                                      Dataset<Row> documentMetadataDF,
                                                      PipeExecutionEnvironment environment); }
TaraReferenceExtractionUtils { public static Dataset<Row> buildDocumentMetadata(Dataset<Row> documentTextDF, Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF) { logger.info("Building document metadata for input data."); Column joinExprs = documentTextDF.col("id").equalTo( extractedDocumentMetadataMergedWithOriginalDF.col("id")); return documentTextDF .join(extractedDocumentMetadataMergedWithOriginalDF, joinExprs, "left_outer") .select( documentTextDF.col("id"), extractedDocumentMetadataMergedWithOriginalDF.col("title"), extractedDocumentMetadataMergedWithOriginalDF.col("abstract"), documentTextDF.col("text") ); }  static Dataset<Row> buildDocumentMetadata(Dataset<Row> documentTextDF,
                                                     Dataset<Row> extractedDocumentMetadataMergedWithOriginalDF); static Dataset<Row> runReferenceExtraction(SparkSession spark,
                                                      Dataset<Row> documentMetadataDF,
                                                      PipeExecutionEnvironment environment);  }