cds-data-pipeline

cds-data-pipeline

Documentation · npm

A CAP plugin for declarative, scheduled data pipelines between CAP services. Each pipeline moves data from one source to one target in a linear READ → MAP → WRITE flow — with tracking, retry, delta support, and a management API out of the box.

Beyond CAP

CAP already gives you the moving parts — cds.connect.to, consumption views, cds.ql, UPSERT, cds.spawn, the Event Queues scheduling API (CDS 10), and the standard before / on / after hook API on services. Reference samples assemble the read–map–write loop by hand: paging, delta watermarks, error handling, and scheduling copied into every project.

This plugin reuses those primitives and adds the orchestration layer on top:

You call addPipeline({ source, target, … }) once; the engine owns the loop. Hooks are the CAP escape hatch when a run needs custom logic — not a requirement for every pipeline.

Install

npm add cds-data-pipeline

Connect: cds.connect.to('data-pipeline'). Full feature list: Features.

Features

Pipeline intents

Intent is inferred from the config shape — no kind flag.

Source adapters

OData V2/V4, REST (cursor/offset/page), CQN, server-driven paging, custom. See Sources.

Target adapters

Local DB (default), remote OData, custom. See Targets.

Delta, scheduling, hooks

Scheduling engines

schedule: 600_000,                              // spawn — implicit engine
schedule: { every: '10m', engine: 'queued' }, // Event Queues: .schedule().every()
schedule: { cron: '0 2 * * *', engine: 'queued' },

Details: Internal scheduling (queued) · External scheduling · Recipes · Event hooks

Management API and observability

OData at /pipeline: pipeline list, run history, statistics, execute, flush, inspectData, configuration overrides. See Management service.

Resilience

Retry with exponential backoff on remote I/O, concurrency guard (no parallel runs), transactional batches.

Quick example

const cds = require('@sap/cds');

cds.on('served', async () => {
    const pipelines = await cds.connect.to('data-pipeline');

    await pipelines.addPipeline({
        name: 'BusinessPartners',
        source: { service: 'API_BUSINESS_PARTNER', entity: 'A_BusinessPartner' },
        target: { entity: 'db.BusinessPartners' },
        delta: { field: 'modifiedAt', mode: 'timestamp' },
        schedule: 600_000,
    });
});

module.exports = cds.server;

Step-by-step: Get started

Used by cds-data-federation

cds-data-federation composes this engine for annotation-driven sync — you install both packages, but rarely call addPipeline yourself:

• @federation.replicate — federation scans consumption views at boot and registers entity-shape pipelines via pipeline-binding.js (source entity → local table, schedule, delta from annotation options).
• cache.strategy: 'entity' on @federation.delegate — on TTL miss, a pipeline fills a SQLite snapshot of the projected entity.

Federation-bound pipelines show up in /pipeline and the Pipeline Console alongside programmatic ones. For custom transforms on a federation pipeline, hook the pipeline by name (defaults to the entity name). See the federation README and first replication.

Event hooks (optional)

DataPipelineService is a normal cds.Service. The default path needs no handler code — register before / on / after on PIPELINE.* only when a run needs filtering, enrichment, side effects, or observability beyond the built-in MAP/WRITE. That is the usual CAP pattern: declarative wiring by default, intercept when processing gets non-trivial.

const pipelines = await cds.connect.to('data-pipeline');

// Filter source rows before rename mapping (federation replicate or addPipeline)
pipelines.before('PIPELINE.MAP', 'ReplicatedPartners', (req) => {
    req.data.sourceRecords = req.data.sourceRecords.filter(r => !r.blocked);
});

// Per-batch side effect after the default WRITE commits its statistics
pipelines.after('PIPELINE.WRITE', 'Shipments', async (_results, req) => {
    const { runId, batchIndex, statistics } = req.data;
    await cds.tx(req).run(INSERT.into('BatchMetrics').entries({
        runId, batchIndex, created: statistics?.created ?? 0,
    }));
});

Full event reference: Event hooks · Management service

Pipeline Console

Pre-built UI for /pipeline — enable via config reuse or cds add pipeline-console:

{
  "cds": {
    "requires": {
      "data-pipeline": {
        "impl": "cds-data-pipeline",
        "management": {
          "reuse": {
            "api": true,
            "console": true
          }
        }
      }
    }
  }
}

The console loads pinned SAPUI5 1.150.0 from ui5.sap.com. Set management.ui5Url to an HTTP(S) URL or same-origin absolute path when you host the runtime yourself.

Details: Pipeline Console · Feature activation

Documentation

• Pipeline guide — concepts, recipes, adapters
• Feature catalog
• Programmatic API
• Management service

AI assistants

Coming soon. Agent guidance (AGENTS.md) and task skills for coding assistants will ship in a future release.

SAP data extraction. @sap/cds ships under the SAP Developer License Agreement (3.2 CAP). Section 1 limits mass extraction from an SAP product to a non-SAP product unless required for interoperability with an SAP product. When pointing a pipeline at an SAP source, stay within that carve-out.