@lde/sparql-anything

Convert tabular and other non-RDF sources to RDF with the SPARQL Anything CLI.

Installation

npm install @lde/sparql-anything

SparqlAnythingConverter

The converter runs the SPARQL Anything jar once per chunk to bound memory use, then concatenates the resulting N-Triples into a single file. A job is a query and the chunks to run it over, so what those processes share is stated once. Processes are spawned through a @lde/task-runner, so the same converter works on the host, in Docker, or anywhere else a TaskRunner is implemented.

import { SparqlAnythingConverter } from '@lde/sparql-anything';
import { NativeTaskRunner } from '@lde/task-runner-native';

const converter = new SparqlAnythingConverter({
  jarPath: 'bin/sparql-anything.jar',
  workDir: 'data', // the task runner's working directory
  taskRunner: new NativeTaskRunner({ cwd: 'data' }),
});

await converter.convert(
  [
    // One query over many chunks, with reference data loaded alongside it.
    {
      queryFile: 'config/places.rq',
      chunks: ['data/places_aa.csv', 'data/places_ab.csv'],
      load: 'data/reference.ttl',
    },
    // A shorter job of a different shape, in the same call.
    { queryFile: 'config/names.rq', chunks: ['data/names_aa.csv'] },
    // A query that names its own input, so it takes no chunks.
    { queryFile: 'config/ontology.rq', load: 'data/ontology.rdf' },
  ],
  'output/places.nt',
);

Options

Option	Type	Description
jarPath	string	Path to the SPARQL Anything CLI jar, as the task runner sees it
workDir	string	The task runner's working directory; see Where files are written
heap	string	Maximum JVM heap per chunk process, as -Xmx takes it (default '2g'); see Memory
cliArgs	string[]	Further arguments for the SPARQL Anything CLI; see Memory
concurrency	number	How many chunks to convert at once (default 1); see Converting several chunks at once
onChunkConverted	(progress) => void	Called as each chunk finishes; see Following a conversion
taskRunner	TaskRunner<Task>	Runs the SPARQL Anything process for each chunk

Jobs

Each entry passed to convert() is one query and the chunks to run it over – one process per chunk, so what those processes share is stated once.

Field	Type	Description
queryFile	string	Path to the SPARQL CONSTRUCT query to run
chunks	string[]	Paths substituted for the literal {SOURCE} in the query, one process each; omit when the query names its own input
load	string	Optional path passed to --load; see Loading existing RDF

A query and its chunks have to agree: a query naming {SOURCE} without chunks, or chunks whose query never names {SOURCE}, is rejected rather than run – SPARQL Anything would report the first as a parse error and the second not at all. An empty chunks array is rejected too: a step that produced none has already failed.

Jobs of different shapes belong in one call. They are run by one converter, so a long job's chunks and a short one's pack together instead of draining in phases – and their outputs land in one file, in the order given, without the caller stitching anything together afterwards.

Where files are written

workDir is the task runner's working directory – cwd for a NativeTaskRunner, mountDir for a DockerTaskRunner. The converter writes its generated query files and per-process outputs into a fresh subdirectory there and removes it when the conversion ends, then refers to them by a path relative to workDir, so the identical command works on the host and inside a container.

Per-run directories matter for more than tidiness: an output left over from an earlier run would satisfy the non-empty check below with stale triples.

jarPath, and each job's load and chunks paths, are passed through as given, because only the caller knows how the runner sees them – in a container the jar usually lives in the image, while the chunks have to be under the mount.

Loading existing RDF

load is optional. Pass it to combine the converted data with RDF you already have – a lookup table the query joins against, for instance. SPARQL Anything reads a file into the default graph, and a directory as one named graph per RDF file it holds, so the two are not interchangeable. Leave load unset and no --load is passed at all.

Memory

One process per chunk bounds how much has to be held at once, but only together with a heap cap: SPARQL Anything materialises a chunk's whole result graph before writing it, and a JVM with no -Xmx helps itself to a quarter of host memory. There is therefore always a cap, 2g unless you raise it:

heap: '4g', // -Xmx4g

Size it with the chunk size. A chunk that outgrows the heap fails loudly – the JVM's OutOfMemoryError arrives in the output of a non-zero exit, which aborts the conversion – where an uncapped JVM instead grows until the OOM killer takes the whole container.

Converting several chunks at once

concurrency is how many chunks are converted at the same time. Each one is a JVM of its own, so it multiplies against heap: a run needs concurrency × heap, on the machine the task runner uses – which is not this process's machine when the runner is Docker or remote.

That is why the default is 1 rather than something derived from the CPU count or a memory limit: the converter cannot see the machine its processes run on, so the number is the caller's to choose. map.sh sizes its pool from nproc capped by the cgroup limit, budgeting ~3 GB a worker; a caller who knows their deployment can do the same arithmetic and pass the result.

Chunks of every job are converted through one pool, in the order the jobs and their chunks were given – a long job and a short one pack together rather than draining in phases. The output is concatenated in that same order, however the processes happened to finish.

The first failure aborts the run: no further chunk is started, and the processes still going are stopped rather than left writing into a directory the converter is about to delete. A process that cannot be stopped – one that has just exited, say – does not change what is reported: the conversion failure is the one worth reading.

NOTE

A DockerTaskRunner configured with a containerName runs one task at a time – the name is how other containers address it – so it rejects a second chunk rather than taking the name from the first. Leave containerName unset for a converter that runs chunks in parallel.

cliArgs is the escape hatch for CLI flags the converter does not model itself, appended to the arguments it sets. It cannot repeat those: -q, -f, -o and -l are rejected, in their long and --flag=value forms too, because the converter reads back the --output it named, in the --format it asked for – overriding either leaves it reporting an empty conversion, or concatenating fragments that are not N-Triples. SPARQL Anything documents repetition only for -v and -c, so a repeated flag has no defined winner to rely on.

Following a conversion

A conversion says nothing for as long as it takes – a quarter of an hour, over eighteen chunks, for the GeoNames dumps. onChunkConverted is called as each one finishes:

onChunkConverted: ({ index, total, chunk }) =>
  console.log(`Converted ${index}/${total}${chunk === undefined ? '' : `: ${chunk}`}`),

It is called once per chunk, in the order they finish rather than the order they were given, and not at all for a chunk that failed – a failure arrives as the rejection instead. A callback that throws aborts the run, like any other failure.

It is wiring rather than configuration: a runtime that already has somewhere to report progress passes a function that forwards to it, which is why this is a plain callback rather than a reporter interface of its own.

Chunking

convert() takes chunks it does not create, because a caller often has work to do first – filtering rows out, say. chunk() produces them:

import { chunk } from '@lde/sparql-anything';

const chunks = await chunk('data/allCountries.txt', {
  rows: 1_000_000,
  into: 'data/chunks',
  header: 'geonameid\tname\tlatitude\tlongitude',
  extension: '.csv',
});
// → ['data/chunks/allCountries-0000.csv', 'data/chunks/allCountries-0001.csv', …]

It streams, so the file never has to fit in memory, and returns the chunk paths in order – which is what a job's chunks takes.

Option	Type	Description
rows	number	Data rows per chunk, chosen together with heap: a chunk is what one process has to hold
into	string	Directory the chunks are written to, created if it does not exist
header	string	Line repeated at the top of every chunk; leave out for a format without a header
extension	string	Extension for the chunk files (default: the input's own); see below

Set extension for a tool that reads the format from the file name – SPARQL Anything does, so a .txt export of a CSV has to be chunked as .csv to be read as one.

Splitting is by line, so every record must be one line. A delimited format that wraps a field in quotes to carry a newline inside it would be cut in two; tab-separated exports, N-Triples and NDJSON are one record per line by definition. Line endings are normalised to \n.

Chunks of the same input left by an earlier call are removed first, so a re-run cannot leave a longer run's tail behind for something to pick up. Only those: everything else in the directory is the caller's. Take the paths chunk() returns as the list of chunks to convert, rather than rediscovering them by name in the directory – a listing would also pick up whatever else is there.

An input with no rows is an error rather than an empty set of chunks: a step that produced an empty file has already failed.

How a conversion runs

For each chunk – or once, for a job that has none – the converter:

1. Replaces the literal {SOURCE} in the job’s query with the chunk’s path and writes the result to a temporary .rq file. The query is read once per job, and interpolated per chunk.
2. Runs java -Xmx<heap> -jar <jar> -q <query> [--load <load>] --format NT --output <chunk>.nt [cliArgs], with every path quoted, so a space or a shell metacharacter in a filename can neither break the command nor inject into it.
3. Waits for the process; a non-zero exit aborts the whole conversion so a crashed chunk can never be silently dropped from the output.
4. Checks that the output is not empty. SPARQL Anything exits successfully when it cannot read or parse an input – it logs the problem and writes nothing – so an empty or missing output aborts the conversion too.

Converting an empty list of jobs is an error rather than an empty output: a step that produced none has already failed.

Finally, the .nt files are concatenated, in the order the jobs and their chunks were given, into the output path. The concatenation streams, so multi-gigabyte outputs do not have to fit in memory, and it is byte for byte what cat would give: a newline is inserted only after a file that does not end in one, so two triples never share a line and no blank line is added between files. N-Triples has no prefixes or document structure, so concatenating per-chunk files always yields a single valid document.