Import Jobs

Bulk-ingest models, twins and relationships from cloud storage with resumable background jobs.

Import jobs bulk-ingest graph data from a blob in cloud storage. The job runs in the background, reports progress, and resumes from a checkpoint if the API restarts — so a million-line import does not depend on a single HTTP request staying open.

This page covers what you need to use jobs. The configuration keys live in the configuration reference.

When to use an import job

Loading an existing dataset (an ADT export, a dump, a generated ontology).
Anything larger than a single batch request: POST /digitaltwins and POST /relationships are capped by Parameters:MaxBatchSize (default 2000) and reject larger payloads with 400.
Work that must survive a pod restart — the job, not your shell, owns the progress.

Supported storage

Provider	URI form	Credentials
Azure Blob Storage	`https://<account>.blob.core.windows.net/<container>/<blob>?sv=…&sig=…`	SAS token in the query string, or managed identity
AWS S3	`s3://bucket/key` or `https://bucket.s3.amazonaws.com/key`	Default AWS credential chain, or a pre-signed URL
Google Cloud Storage	`gs://bucket/key` or `https://storage.googleapis.com/bucket/key`	Default GCP credentials, or a signed URL

Input needs read; output needs write. For Azure SAS that means sp=r and sp=w; for S3/GCS it means your IAM/service-account policy allows the calls.

When the URI carries a SAS token or a pre-signed URL, the service needs no cloud credential at all.

Starting a job

curl -X PUT http://localhost:8080/jobs/imports/import-2026-10-20 \
  -H 'Content-Type: application/json' \
  -d '{
        "inputBlobUri":  "https://acct.blob.core.windows.net/data/export.ndjson?sv=...&sp=r&sig=...",
        "outputBlobUri": "https://acct.blob.core.windows.net/data/export.result.ndjson?sv=...&sp=w&sig=..."
      }'

The call returns 201 immediately; execution is in the background. The job id is yours — reuse it and you overwrite that job's record rather than creating a second one.

Method	Route	Purpose
`PUT`	`/jobs/imports/{id}`	Create and start (body: `inputBlobUri`, `outputBlobUri`)
`GET`	`/jobs/imports/{id}`	Status, counters and errors
`GET`	`/jobs/imports`	List jobs
`POST`	`/jobs/imports/{id}/cancel`	Request cancellation (valid while `Notstarted`/`Running`)
`POST`	`/jobs/imports/{id}/resume`	Resume from the last checkpoint (valid while `Running`/`Failed`, and not locked elsewhere)
`DELETE`	`/jobs/imports/{id}`	Delete the job record

The input format

NDJSON only — one JSON object per line, no embedded newlines, sections in a fixed order:

{"Section": "Header"}
{"fileVersion": "1.0.0"}
{"Section": "Models"}
<DTDL model JSON, one per line>
{"Section": "Twins"}
<twin JSON, one per line>
{"Section": "Relationships"}
<relationship JSON, one per line>

Rules the validator actually enforces:

The Header section is required and must be first.
fileVersion must be exactly "1.0.0".
Models, Twins and Relationships are optional, but appear in that order.
Every object must be valid JSON on a single line.

Models are processed as one unit (they can reference each other through inheritance), while twins and relationships are batched.

JSON-LD and other formats are not supported.

Generating the file

ndjson_generator.py

import json

def write_ndjson(models, twins, relationships, path, header=None):
    header_obj = header or {
        "fileVersion": "1.0.0",
        "author": "authorName",
        "organization": "organization",
    }
    with open(path, "w", encoding="utf-8") as f:
        # Header is mandatory and must come first
        f.write(json.dumps({"Section": "Header"}) + "\n")
        f.write(json.dumps(header_obj) + "\n")

        f.write(json.dumps({"Section": "Models"}) + "\n")
        for model in models:
            # models are often handed over as already-serialized JSON
            line = model.strip().replace("\n", "") if isinstance(model, str) else json.dumps(model)
            f.write(line + "\n")

        f.write(json.dumps({"Section": "Twins"}) + "\n")
        for twin in twins:
            f.write(json.dumps(twin) + "\n")

        f.write(json.dumps({"Section": "Relationships"}) + "\n")
        for rel in relationships:
            f.write(json.dumps(rel) + "\n")

Statuses

Status	Meaning
`Notstarted`	Created, not picked up yet
`Running`	Reading the input stream
`Succeeded`	Everything imported
`PartiallySucceeded`	Finished, with items that failed — see the errors
`Failed`	Nothing imported, or the job gave up
`Cancelling`	Cancellation requested; the worker has not stopped yet
`Cancelled`	Stopped

Two behaviours worth designing around:

Cancellation is cooperative. cancel sets Cancelling; the worker checks it on a heartbeat, so expect a short delay before Cancelled.
A lost database connection does not fail the job. It stays Running with no finish time and is picked up again later — from its last checkpoint.

Resumption

Progress is checkpointed at section boundaries and every Parameters:DefaultCheckpointInterval items (default 50). On resume, the job seeks back to the recorded line and continues; sections that already completed are skipped outright.

An interrupted job is recovered automatically: the service polls for Running jobs whose lock is free or expired (every two minutes) and resumes them. You only need resume when you want to drive that decision yourself — for example after fixing a bad input line, where the job is Failed.

For S3 and GCS, resuming appends to the existing output object rather than replacing it.

Permissions

Blob	Needs
Input	read
Output	write

Azure: sp=r for input, sp=w for output. AWS/GCP: the corresponding IAM or service-account permissions. All URIs must be reachable from the Konnektr Graph instance — a URI that works on your laptop but not inside the cluster is the most common cause of an immediately failed job.