Pipelines and jobs
Pipelines and jobs turn source content into searchable VectorAmp data. They handle parsing, chunking, embedding, indexing, retries, and operational status.
Users judge search by the answer they see. Operators debug search by checking ingestion jobs: what ran, what failed, what changed, and whether the dataset is fresh.
What a pipeline does
A pipeline usually performs these steps:
- Read source content from file upload, object storage, web content, Drive, or application data.
- Parse files into text and structured metadata.
- Chunk content into retrievable units.
- Generate embeddings with the dataset’s embedding configuration.
- Attach metadata for filtering and citations.
- Index into SABLE so search and cited answers can use the content.
- Record job state for monitoring, retries, and audit.
Jobs vs pipelines
| Term | Meaning |
|---|---|
| Pipeline | The configured workflow: source, target dataset, parsing/chunking behavior, model, and schedule. |
| Job | One execution of that workflow. A job has status, timestamps, counts, errors, and retry state. |
| Schedule | A recurring trigger for a pipeline, such as hourly, daily, or weekly sync. |
Job status
| Status | Meaning |
|---|---|
queued | Waiting for workers or source access. |
running | Actively reading, parsing, embedding, or indexing. |
completed | Finished successfully. |
failed | Hit an error that requires review or retry. |
paused | Stopped by an operator or schedule setting. |
Status names can vary by API version and connector, but every production workflow should expose the same operator questions: what ran, what succeeded, what failed, and what should be retried?
Monitoring checklist
Track these fields for production ingestion:
- source name and type,
- target dataset,
- job status,
- started/completed timestamps,
- records or files discovered,
- records or files processed,
- skipped files and reasons,
- errors and retry count,
- embedding/indexing completion,
- last successful sync time.
Failure handling
Common ingestion failures include:
| Failure | What to check |
|---|---|
| Permission denied | Source credentials, bucket/folder access, expired OAuth, revoked service account. |
| Unsupported file | File type, corruption, password protection, binary-only content. |
| Empty output | Parser configuration, OCR needs, source file quality. |
| Metadata missing | Source mapping, custom fields, transform logic, API payload shape. |
| Search stale after sync | Job completion, indexing completion, cache/window timing, dataset ID mismatch. |
Scheduling guidance
Choose sync frequency based on user risk:
- Minutes: support, incident, customer-facing, or operational data.
- Hourly: active team knowledge and frequently updated docs.
- Daily: policies, handbooks, stable docs, reporting material.
- Manual approval: legal, compliance, security, or high-risk content where review matters.
Example: job-oriented workflow
source = client.sources.create(
dataset_id="ds_123",
type="s3",
config={"bucket": "company-knowledge", "prefix": "policies/"},
)
job = client.jobs.create(source_id=source.id)
print(job.id, job.status)
Use the API or Jobs dashboard to monitor the job until it completes, then run representative search and citation checks.