net.sansa_stack.rdf.spark.io

Class RddRdfSaver<T>

java.lang.Object

net.sansa_stack.rdf.spark.io.RddRdfSaver<T>

Type Parameters:

T -

public class RddRdfSaver<T>
extends Object

A fluent API for configuration of how to save an RDD of RDF data to disk. This class uniformly handles Triples, Quads, Model, Datasets, etc using a set of lambdas for relevant conversion. Instances of this class should be created using the appropriate createFor[Type] methods. TODO Implementation for Model is currently missing

Field Summary

Fields

Modifier and Type	Field and Description
protected boolean	allowOverwriteFiles
protected Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad>>	convertToQuad
protected Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple>>	convertToTriple
protected boolean	deletePartitionFolderAfterMerge
protected org.apache.jena.shared.PrefixMapping	globalPrefixMapping
protected org.apache.hadoop.conf.Configuration	hadoopConfiguration
protected boolean	mapQuadsToTriplesForTripleLangs Whether to convert quads to triples if a triple-based output format is requested
protected org.apache.jena.riot.RDFFormat	outputFormat
protected org.apache.hadoop.fs.Path	partitionFolder
protected boolean	partitionsAsIndependentFiles
protected org.apache.spark.api.java.JavaRDD<T>	rdd
protected BiConsumer<T,org.apache.jena.riot.system.StreamRDF>	sendRecordToStreamRDF
protected org.apache.spark.api.java.JavaSparkContext	sparkContext
protected org.apache.hadoop.fs.Path	targetFile
protected boolean	useCoalesceOne
protected boolean	useElephas

Constructor Summary

Constructors

Constructor and Description
RddRdfSaver(org.apache.spark.api.java.JavaRDD<T> rdd, BiConsumer<T,org.apache.jena.riot.system.StreamRDF> sendRecordToStreamRDF, Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple>> convertToTriple, Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad>> convertToQuad)

Method Summary

Modifier and Type	Method and Description
static <T> RddRdfSaver	create(org.apache.spark.api.java.JavaRDD<T> rdd, org.aksw.commons.lambda.serializable.SerializableBiConsumer<T,org.apache.jena.riot.system.StreamRDF> sendRecordToStreamRDF, Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple>> convertToTriple, Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad>> convertToQuad) Create method.
static RddRdfSaver<org.apache.jena.query.Dataset>	createForDataset(org.apache.spark.api.java.JavaRDD<org.apache.jena.query.Dataset> rdd)
static RddRdfSaver<org.apache.jena.sparql.core.Quad>	createForQuad(org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad> rdd)
static RddRdfSaver<org.apache.jena.graph.Triple>	createForTriple(org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple> rdd)
static Function<OutputStream,org.apache.jena.riot.system.StreamRDF>	createStreamRDFFactory(org.apache.jena.riot.RDFFormat rdfFormat, boolean mapQuadsToTriplesForTripleLangs, org.apache.jena.shared.PrefixMapping prefixMapping) Create a function that can create a StreamRDF instance that is backed by the given OutputStream.
org.apache.jena.shared.PrefixMapping	getGlobalPrefixMapping()
org.apache.jena.riot.RDFFormat	getOutputFormat()
org.apache.hadoop.fs.Path	getPartitionFolder()
org.apache.hadoop.fs.Path	getTargetFile()
boolean	isAllowOverwriteFiles()
boolean	isDeletePartitionFolderAfterMerge()
boolean	isMapQuadsToTriplesForTripleLangs()
boolean	isPartitionsAsIndependentFiles()
boolean	isUseCoalesceOne()
boolean	isUseElephas()
static void	mergeFolder(Path outFile, Path srcFolder, String pattern, Comparator<? super Path> pathComparator)
RddRdfSaver<T>	mutate(Consumer<RddRdfSaver<T>> action) Pass this object to a consumer.
static Iterator<String>	partitionMapperNQuads(Iterator<org.apache.jena.sparql.core.Quad> it)
static Iterator<String>	partitionMapperNTriples(Iterator<org.apache.jena.graph.Triple> it) Save the RDD to a single file.
static <T> org.aksw.commons.lambda.throwing.ThrowingFunction<Iterator<T>,Iterator<String>>	partitionMapperRDFStream(Function<OutputStream,org.apache.jena.riot.system.StreamRDF> streamRDFFactory, BiConsumer<? super T,org.apache.jena.riot.system.StreamRDF> sendRecordToWriter)
void	run() Run the save action according to configuration
static <T> void	saveToFolder(org.apache.spark.api.java.JavaRDD<T> javaRdd, String path, org.apache.jena.riot.RDFFormat rdfFormat, boolean mapQuadsToTriplesForTripleLangs, org.apache.jena.shared.PrefixMapping globalPrefixMapping, BiConsumer<T,org.apache.jena.riot.system.StreamRDF> sendRecordToStreamRDF) Save the data in Trig/Turtle or its sub-formats (n-quads/n-triples) format.
static <T> void	saveUsingElephas(org.apache.spark.api.java.JavaRDD<T> rdd, org.apache.hadoop.fs.Path path, org.apache.jena.riot.Lang lang, org.aksw.commons.lambda.serializable.SerializableFunction<? super T,?> recordToWritable)
RddRdfSaver<T>	setAllowOverwriteFiles(boolean allowOverwriteFiles)
RddRdfSaver<T>	setDeletePartitionFolderAfterMerge(boolean deletePartitionFolderAfterMerge)
RddRdfSaver<T>	setGlobalPrefixMapping(org.apache.jena.shared.PrefixMapping globalPrefixMapping) Set a prefix mapping to be used "globally" across all partitions.
RddRdfSaver	setMapQuadsToTriplesForTripleLangs(boolean mapQuadsToTriplesForTripleLangs) Whether to convert quads to triples if a triple-based output format is requested Jena by default discards any quad outside of the default graph when writing to a triple format.
RddRdfSaver<T>	setOutputFormat(org.apache.jena.riot.RDFFormat outputFormat)
RddRdfSaver<T>	setPartitionFolder(org.apache.hadoop.fs.Path partitionFolder)
RddRdfSaver<T>	setPartitionFolder(String partitionFolder)
RddRdfSaver<T>	setPartitionsAsIndependentFiles(boolean partitionsAsIndependentFiles)
RddRdfSaver<T>	setTargetFile(org.apache.hadoop.fs.Path targetFile)
RddRdfSaver<T>	setTargetFile(String targetFile)
void	setUseCoalesceOne(boolean useCoalesceOne)
RddRdfSaver<T>	setUseElephas(boolean useElephas)
static String	toString(org.apache.jena.shared.PrefixMapping prefixMapping, org.apache.jena.riot.RDFFormat rdfFormat) Convert a prefix mapping to a string
static void	validateOutFolder(org.apache.hadoop.fs.Path path, org.apache.hadoop.conf.Configuration conf, boolean deleteIfExists)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

rdd

protected org.apache.spark.api.java.JavaRDD<T> rdd

partitionFolder

protected org.apache.hadoop.fs.Path partitionFolder

targetFile

protected org.apache.hadoop.fs.Path targetFile

useCoalesceOne

protected boolean useCoalesceOne

deletePartitionFolderAfterMerge

protected boolean deletePartitionFolderAfterMerge

globalPrefixMapping

protected org.apache.jena.shared.PrefixMapping globalPrefixMapping

outputFormat

protected org.apache.jena.riot.RDFFormat outputFormat

allowOverwriteFiles

protected boolean allowOverwriteFiles

useElephas

protected boolean useElephas

partitionsAsIndependentFiles

protected boolean partitionsAsIndependentFiles

mapQuadsToTriplesForTripleLangs

protected boolean mapQuadsToTriplesForTripleLangs

Whether to convert quads to triples if a triple-based output format is requested

sendRecordToStreamRDF

protected BiConsumer<T,org.apache.jena.riot.system.StreamRDF> sendRecordToStreamRDF

convertToTriple

protected Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple>> convertToTriple

convertToQuad

protected Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad>> convertToQuad

sparkContext

protected org.apache.spark.api.java.JavaSparkContext sparkContext

hadoopConfiguration

protected org.apache.hadoop.conf.Configuration hadoopConfiguration

Constructor Detail

RddRdfSaver

public RddRdfSaver(org.apache.spark.api.java.JavaRDD<T> rdd,
                   BiConsumer<T,org.apache.jena.riot.system.StreamRDF> sendRecordToStreamRDF,
                   Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple>> convertToTriple,
                   Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad>> convertToQuad)

Method Detail

isMapQuadsToTriplesForTripleLangs

public boolean isMapQuadsToTriplesForTripleLangs()

setMapQuadsToTriplesForTripleLangs

public RddRdfSaver setMapQuadsToTriplesForTripleLangs(boolean mapQuadsToTriplesForTripleLangs)

Whether to convert quads to triples if a triple-based output format is requested Jena by default discards any quad outside of the default graph when writing to a triple format. Setting this flag to true will map each quad in a named graph to the default graph.

isUseCoalesceOne

public boolean isUseCoalesceOne()

setUseCoalesceOne

public void setUseCoalesceOne(boolean useCoalesceOne)

isDeletePartitionFolderAfterMerge

public boolean isDeletePartitionFolderAfterMerge()

setDeletePartitionFolderAfterMerge

public RddRdfSaver<T> setDeletePartitionFolderAfterMerge(boolean deletePartitionFolderAfterMerge)

getGlobalPrefixMapping

public org.apache.jena.shared.PrefixMapping getGlobalPrefixMapping()

getPartitionFolder

public org.apache.hadoop.fs.Path getPartitionFolder()

setPartitionFolder

public RddRdfSaver<T> setPartitionFolder(org.apache.hadoop.fs.Path partitionFolder)

setPartitionFolder

public RddRdfSaver<T> setPartitionFolder(String partitionFolder)

getTargetFile

public org.apache.hadoop.fs.Path getTargetFile()

setTargetFile

public RddRdfSaver<T> setTargetFile(org.apache.hadoop.fs.Path targetFile)

setTargetFile

public RddRdfSaver<T> setTargetFile(String targetFile)

setGlobalPrefixMapping

public RddRdfSaver<T> setGlobalPrefixMapping(org.apache.jena.shared.PrefixMapping globalPrefixMapping)

Set a prefix mapping to be used "globally" across all partitions.

Parameters:

globalPrefixMapping -

Returns:

getOutputFormat

public org.apache.jena.riot.RDFFormat getOutputFormat()

setOutputFormat

public RddRdfSaver<T> setOutputFormat(org.apache.jena.riot.RDFFormat outputFormat)

isAllowOverwriteFiles

public boolean isAllowOverwriteFiles()

setAllowOverwriteFiles

public RddRdfSaver<T> setAllowOverwriteFiles(boolean allowOverwriteFiles)

isUseElephas

public boolean isUseElephas()

setUseElephas

public RddRdfSaver<T> setUseElephas(boolean useElephas)

isPartitionsAsIndependentFiles

public boolean isPartitionsAsIndependentFiles()

setPartitionsAsIndependentFiles

public RddRdfSaver<T> setPartitionsAsIndependentFiles(boolean partitionsAsIndependentFiles)

mutate

public RddRdfSaver<T> mutate(Consumer<RddRdfSaver<T>> action)

Pass this object to a consumer. Useful to conditionally configure this object without breaking the fluent chain:

    rdd.configureSave().mutate(self -> { if (condition) { self.setX(); }}).run();
 

Parameters:

action -

Returns:

run

public void run()
         throws IOException

Run the save action according to configuration

Throws:

IOException

validateOutFolder

public static void validateOutFolder(org.apache.hadoop.fs.Path path,
                                     org.apache.hadoop.conf.Configuration conf,
                                     boolean deleteIfExists)
                              throws IOException

Throws:

IOException

toString

public static String toString(org.apache.jena.shared.PrefixMapping prefixMapping,
                              org.apache.jena.riot.RDFFormat rdfFormat)

Convert a prefix mapping to a string

mergeFolder

public static void mergeFolder(Path outFile,
                               Path srcFolder,
                               String pattern,
                               Comparator<? super Path> pathComparator)
                        throws IOException

Throws:

IOException

partitionMapperNTriples

public static Iterator<String> partitionMapperNTriples(Iterator<org.apache.jena.graph.Triple> it)

Save the RDD to a single file. Underneath invokes [[JenaDatasetWriter#saveToFolder]] and merges the set of files created by it. See [[JenaDatasetWriter#saveToFolder]] for supported formats.

Parameters:

mode -

exitOnError - / def saveToFile(outFile: String, prefixMapping: PrefixMapping, rdfFormat: RDFFormat, outFolder: String, mode: io.SaveMode.Value = SaveMode.ErrorIfExists, exitOnError: Boolean = false): Unit = {

val outFilePath = Paths.get(outFile).toAbsolutePath val outFileFileName = outFilePath.getFileName.toString val outFolderPath = if (outFolder == null) outFilePath.resolveSibling(outFileFileName + "-parts") else Paths.get(outFolder).toAbsolutePath

saveToFolder(outFolderPath.toString, prefixMapping, rdfFormat, mode, exitOnError) mergeFolder(outFilePath, outFolderPath, "part*") }

partitionMapperNQuads

public static Iterator<String> partitionMapperNQuads(Iterator<org.apache.jena.sparql.core.Quad> it)

createStreamRDFFactory

public static Function<OutputStream,org.apache.jena.riot.system.StreamRDF> createStreamRDFFactory(org.apache.jena.riot.RDFFormat rdfFormat,
                                                                                                  boolean mapQuadsToTriplesForTripleLangs,
                                                                                                  org.apache.jena.shared.PrefixMapping prefixMapping)

Create a function that can create a StreamRDF instance that is backed by the given OutputStream.

Parameters:

rdfFormat -

prefixMapping -

Returns:

partitionMapperRDFStream

public static <T> org.aksw.commons.lambda.throwing.ThrowingFunction<Iterator<T>,Iterator<String>> partitionMapperRDFStream(Function<OutputStream,org.apache.jena.riot.system.StreamRDF> streamRDFFactory,
                                                                                                                           BiConsumer<? super T,org.apache.jena.riot.system.StreamRDF> sendRecordToWriter)

saveToFolder

public static <T> void saveToFolder(org.apache.spark.api.java.JavaRDD<T> javaRdd,
                                    String path,
                                    org.apache.jena.riot.RDFFormat rdfFormat,
                                    boolean mapQuadsToTriplesForTripleLangs,
                                    org.apache.jena.shared.PrefixMapping globalPrefixMapping,
                                    BiConsumer<T,org.apache.jena.riot.system.StreamRDF> sendRecordToStreamRDF)
                             throws IOException

Save the data in Trig/Turtle or its sub-formats (n-quads/n-triples) format. If prefixes should be written out then they have to provided as an argument to the prefixMapping parameter. Prefix mappings are broadcasted to and processed in a .mapPartition operation. If the prefixMapping is non-empty then the first part file written out contains them. No other partition will write out prefixes.

Parameters:

path - the folder into which the file(s) will be written to

mode - the expected behavior of saving the data to a data source

Throws:

IOException

saveUsingElephas

public static <T> void saveUsingElephas(org.apache.spark.api.java.JavaRDD<T> rdd,
                                        org.apache.hadoop.fs.Path path,
                                        org.apache.jena.riot.Lang lang,
                                        org.aksw.commons.lambda.serializable.SerializableFunction<? super T,?> recordToWritable)

create

public static <T> RddRdfSaver create(org.apache.spark.api.java.JavaRDD<T> rdd,
                                     org.aksw.commons.lambda.serializable.SerializableBiConsumer<T,org.apache.jena.riot.system.StreamRDF> sendRecordToStreamRDF,
                                     Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple>> convertToTriple,
                                     Function<org.apache.spark.api.java.JavaRDD<T>,org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad>> convertToQuad)

Create method. Note that the 'sendRecordToSTreamRDF' parameter is serializable

Type Parameters:

T -

Parameters:

rdd -

sendRecordToStreamRDF -

convertToTriple -

convertToQuad -

Returns:

createForTriple

public static RddRdfSaver<org.apache.jena.graph.Triple> createForTriple(org.apache.spark.api.java.JavaRDD<org.apache.jena.graph.Triple> rdd)

createForQuad

public static RddRdfSaver<org.apache.jena.sparql.core.Quad> createForQuad(org.apache.spark.api.java.JavaRDD<org.apache.jena.sparql.core.Quad> rdd)

createForDataset

public static RddRdfSaver<org.apache.jena.query.Dataset> createForDataset(org.apache.spark.api.java.JavaRDD<org.apache.jena.query.Dataset> rdd)