curl --request POST \
--url http://localhost:8080/api/v1/pipeline \
--header 'Content-Type: application/json' \
--header 'x-api-key: <api-key>' \
--data '
{
"name": "<string>",
"source": {
"schemaProperties": {
"fields": [
{
"name": "<string>",
"type": "<string>"
}
]
}
},
"protection": {
"purgeSource": true,
"preset": "hipaa-safe-harbor",
"presetExempt": [
"<string>"
]
},
"preprocessor": {},
"dataQuality": {},
"transformation": {},
"destination": {},
"unityCatalog": {
"enabled": false,
"comments": true,
"tags": true,
"properties": true,
"lineage": true,
"credentialsSecret": "<string>",
"catalog": "<string>",
"schema": "default",
"register": true
}
}
'import requests
url = "http://localhost:8080/api/v1/pipeline"
payload = {
"name": "<string>",
"source": { "schemaProperties": { "fields": [
{
"name": "<string>",
"type": "<string>"
}
] } },
"protection": {
"purgeSource": True,
"preset": "hipaa-safe-harbor",
"presetExempt": ["<string>"]
},
"preprocessor": {},
"dataQuality": {},
"transformation": {},
"destination": {},
"unityCatalog": {
"enabled": False,
"comments": True,
"tags": True,
"properties": True,
"lineage": True,
"credentialsSecret": "<string>",
"catalog": "<string>",
"schema": "default",
"register": True
}
}
headers = {
"x-api-key": "<api-key>",
"Content-Type": "application/json"
}
response = requests.post(url, json=payload, headers=headers)
print(response.text)const options = {
method: 'POST',
headers: {'x-api-key': '<api-key>', 'Content-Type': 'application/json'},
body: JSON.stringify({
name: '<string>',
source: {schemaProperties: {fields: [{name: '<string>', type: '<string>'}]}},
protection: {purgeSource: true, preset: 'hipaa-safe-harbor', presetExempt: ['<string>']},
preprocessor: {},
dataQuality: {},
transformation: {},
destination: {},
unityCatalog: {
enabled: false,
comments: true,
tags: true,
properties: true,
lineage: true,
credentialsSecret: '<string>',
catalog: '<string>',
schema: 'default',
register: true
}
})
};
fetch('http://localhost:8080/api/v1/pipeline', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));<?php
$curl = curl_init();
curl_setopt_array($curl, [
CURLOPT_PORT => "8080",
CURLOPT_URL => "http://localhost:8080/api/v1/pipeline",
CURLOPT_RETURNTRANSFER => true,
CURLOPT_ENCODING => "",
CURLOPT_MAXREDIRS => 10,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
CURLOPT_CUSTOMREQUEST => "POST",
CURLOPT_POSTFIELDS => json_encode([
'name' => '<string>',
'source' => [
'schemaProperties' => [
'fields' => [
[
'name' => '<string>',
'type' => '<string>'
]
]
]
],
'protection' => [
'purgeSource' => true,
'preset' => 'hipaa-safe-harbor',
'presetExempt' => [
'<string>'
]
],
'preprocessor' => [
],
'dataQuality' => [
],
'transformation' => [
],
'destination' => [
],
'unityCatalog' => [
'enabled' => false,
'comments' => true,
'tags' => true,
'properties' => true,
'lineage' => true,
'credentialsSecret' => '<string>',
'catalog' => '<string>',
'schema' => 'default',
'register' => true
]
]),
CURLOPT_HTTPHEADER => [
"Content-Type: application/json",
"x-api-key: <api-key>"
],
]);
$response = curl_exec($curl);
$err = curl_error($curl);
curl_close($curl);
if ($err) {
echo "cURL Error #:" . $err;
} else {
echo $response;
}package main
import (
"fmt"
"strings"
"net/http"
"io"
)
func main() {
url := "http://localhost:8080/api/v1/pipeline"
payload := strings.NewReader("{\n \"name\": \"<string>\",\n \"source\": {\n \"schemaProperties\": {\n \"fields\": [\n {\n \"name\": \"<string>\",\n \"type\": \"<string>\"\n }\n ]\n }\n },\n \"protection\": {\n \"purgeSource\": true,\n \"preset\": \"hipaa-safe-harbor\",\n \"presetExempt\": [\n \"<string>\"\n ]\n },\n \"preprocessor\": {},\n \"dataQuality\": {},\n \"transformation\": {},\n \"destination\": {},\n \"unityCatalog\": {\n \"enabled\": false,\n \"comments\": true,\n \"tags\": true,\n \"properties\": true,\n \"lineage\": true,\n \"credentialsSecret\": \"<string>\",\n \"catalog\": \"<string>\",\n \"schema\": \"default\",\n \"register\": true\n }\n}")
req, _ := http.NewRequest("POST", url, payload)
req.Header.Add("x-api-key", "<api-key>")
req.Header.Add("Content-Type", "application/json")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.post("http://localhost:8080/api/v1/pipeline")
.header("x-api-key", "<api-key>")
.header("Content-Type", "application/json")
.body("{\n \"name\": \"<string>\",\n \"source\": {\n \"schemaProperties\": {\n \"fields\": [\n {\n \"name\": \"<string>\",\n \"type\": \"<string>\"\n }\n ]\n }\n },\n \"protection\": {\n \"purgeSource\": true,\n \"preset\": \"hipaa-safe-harbor\",\n \"presetExempt\": [\n \"<string>\"\n ]\n },\n \"preprocessor\": {},\n \"dataQuality\": {},\n \"transformation\": {},\n \"destination\": {},\n \"unityCatalog\": {\n \"enabled\": false,\n \"comments\": true,\n \"tags\": true,\n \"properties\": true,\n \"lineage\": true,\n \"credentialsSecret\": \"<string>\",\n \"catalog\": \"<string>\",\n \"schema\": \"default\",\n \"register\": true\n }\n}")
.asString();require 'uri'
require 'net/http'
url = URI("http://localhost:8080/api/v1/pipeline")
http = Net::HTTP.new(url.host, url.port)
request = Net::HTTP::Post.new(url)
request["x-api-key"] = '<api-key>'
request["Content-Type"] = 'application/json'
request.body = "{\n \"name\": \"<string>\",\n \"source\": {\n \"schemaProperties\": {\n \"fields\": [\n {\n \"name\": \"<string>\",\n \"type\": \"<string>\"\n }\n ]\n }\n },\n \"protection\": {\n \"purgeSource\": true,\n \"preset\": \"hipaa-safe-harbor\",\n \"presetExempt\": [\n \"<string>\"\n ]\n },\n \"preprocessor\": {},\n \"dataQuality\": {},\n \"transformation\": {},\n \"destination\": {},\n \"unityCatalog\": {\n \"enabled\": false,\n \"comments\": true,\n \"tags\": true,\n \"properties\": true,\n \"lineage\": true,\n \"credentialsSecret\": \"<string>\",\n \"catalog\": \"<string>\",\n \"schema\": \"default\",\n \"register\": true\n }\n}"
response = http.request(request)
puts response.read_body{
"warnings": [
"<string>"
],
"codegenScripts": [
{
"kind": "dataQuality",
"status": "ready",
"pendingReason": "<string>",
"generatedAt": "2023-11-07T05:31:56Z",
"model": "<string>",
"warning": "<string>"
}
]
}{
"error": "<string>"
}Create or update a pipeline configuration
Saves (upserts) a pipeline configuration. Invalid field protection is rejected at save time (unknown or not-yet-supported protect.method, preserve outside mask, a string-producing method on a non-string field, any method but hmac on a keyFields column, or a source that is not delimited or JSON). Like every other validator error, the rejection returns HTTP 400 with {"error": "..."} naming the field.
curl --request POST \
--url http://localhost:8080/api/v1/pipeline \
--header 'Content-Type: application/json' \
--header 'x-api-key: <api-key>' \
--data '
{
"name": "<string>",
"source": {
"schemaProperties": {
"fields": [
{
"name": "<string>",
"type": "<string>"
}
]
}
},
"protection": {
"purgeSource": true,
"preset": "hipaa-safe-harbor",
"presetExempt": [
"<string>"
]
},
"preprocessor": {},
"dataQuality": {},
"transformation": {},
"destination": {},
"unityCatalog": {
"enabled": false,
"comments": true,
"tags": true,
"properties": true,
"lineage": true,
"credentialsSecret": "<string>",
"catalog": "<string>",
"schema": "default",
"register": true
}
}
'import requests
url = "http://localhost:8080/api/v1/pipeline"
payload = {
"name": "<string>",
"source": { "schemaProperties": { "fields": [
{
"name": "<string>",
"type": "<string>"
}
] } },
"protection": {
"purgeSource": True,
"preset": "hipaa-safe-harbor",
"presetExempt": ["<string>"]
},
"preprocessor": {},
"dataQuality": {},
"transformation": {},
"destination": {},
"unityCatalog": {
"enabled": False,
"comments": True,
"tags": True,
"properties": True,
"lineage": True,
"credentialsSecret": "<string>",
"catalog": "<string>",
"schema": "default",
"register": True
}
}
headers = {
"x-api-key": "<api-key>",
"Content-Type": "application/json"
}
response = requests.post(url, json=payload, headers=headers)
print(response.text)const options = {
method: 'POST',
headers: {'x-api-key': '<api-key>', 'Content-Type': 'application/json'},
body: JSON.stringify({
name: '<string>',
source: {schemaProperties: {fields: [{name: '<string>', type: '<string>'}]}},
protection: {purgeSource: true, preset: 'hipaa-safe-harbor', presetExempt: ['<string>']},
preprocessor: {},
dataQuality: {},
transformation: {},
destination: {},
unityCatalog: {
enabled: false,
comments: true,
tags: true,
properties: true,
lineage: true,
credentialsSecret: '<string>',
catalog: '<string>',
schema: 'default',
register: true
}
})
};
fetch('http://localhost:8080/api/v1/pipeline', options)
.then(res => res.json())
.then(res => console.log(res))
.catch(err => console.error(err));<?php
$curl = curl_init();
curl_setopt_array($curl, [
CURLOPT_PORT => "8080",
CURLOPT_URL => "http://localhost:8080/api/v1/pipeline",
CURLOPT_RETURNTRANSFER => true,
CURLOPT_ENCODING => "",
CURLOPT_MAXREDIRS => 10,
CURLOPT_TIMEOUT => 30,
CURLOPT_HTTP_VERSION => CURL_HTTP_VERSION_1_1,
CURLOPT_CUSTOMREQUEST => "POST",
CURLOPT_POSTFIELDS => json_encode([
'name' => '<string>',
'source' => [
'schemaProperties' => [
'fields' => [
[
'name' => '<string>',
'type' => '<string>'
]
]
]
],
'protection' => [
'purgeSource' => true,
'preset' => 'hipaa-safe-harbor',
'presetExempt' => [
'<string>'
]
],
'preprocessor' => [
],
'dataQuality' => [
],
'transformation' => [
],
'destination' => [
],
'unityCatalog' => [
'enabled' => false,
'comments' => true,
'tags' => true,
'properties' => true,
'lineage' => true,
'credentialsSecret' => '<string>',
'catalog' => '<string>',
'schema' => 'default',
'register' => true
]
]),
CURLOPT_HTTPHEADER => [
"Content-Type: application/json",
"x-api-key: <api-key>"
],
]);
$response = curl_exec($curl);
$err = curl_error($curl);
curl_close($curl);
if ($err) {
echo "cURL Error #:" . $err;
} else {
echo $response;
}package main
import (
"fmt"
"strings"
"net/http"
"io"
)
func main() {
url := "http://localhost:8080/api/v1/pipeline"
payload := strings.NewReader("{\n \"name\": \"<string>\",\n \"source\": {\n \"schemaProperties\": {\n \"fields\": [\n {\n \"name\": \"<string>\",\n \"type\": \"<string>\"\n }\n ]\n }\n },\n \"protection\": {\n \"purgeSource\": true,\n \"preset\": \"hipaa-safe-harbor\",\n \"presetExempt\": [\n \"<string>\"\n ]\n },\n \"preprocessor\": {},\n \"dataQuality\": {},\n \"transformation\": {},\n \"destination\": {},\n \"unityCatalog\": {\n \"enabled\": false,\n \"comments\": true,\n \"tags\": true,\n \"properties\": true,\n \"lineage\": true,\n \"credentialsSecret\": \"<string>\",\n \"catalog\": \"<string>\",\n \"schema\": \"default\",\n \"register\": true\n }\n}")
req, _ := http.NewRequest("POST", url, payload)
req.Header.Add("x-api-key", "<api-key>")
req.Header.Add("Content-Type", "application/json")
res, _ := http.DefaultClient.Do(req)
defer res.Body.Close()
body, _ := io.ReadAll(res.Body)
fmt.Println(string(body))
}HttpResponse<String> response = Unirest.post("http://localhost:8080/api/v1/pipeline")
.header("x-api-key", "<api-key>")
.header("Content-Type", "application/json")
.body("{\n \"name\": \"<string>\",\n \"source\": {\n \"schemaProperties\": {\n \"fields\": [\n {\n \"name\": \"<string>\",\n \"type\": \"<string>\"\n }\n ]\n }\n },\n \"protection\": {\n \"purgeSource\": true,\n \"preset\": \"hipaa-safe-harbor\",\n \"presetExempt\": [\n \"<string>\"\n ]\n },\n \"preprocessor\": {},\n \"dataQuality\": {},\n \"transformation\": {},\n \"destination\": {},\n \"unityCatalog\": {\n \"enabled\": false,\n \"comments\": true,\n \"tags\": true,\n \"properties\": true,\n \"lineage\": true,\n \"credentialsSecret\": \"<string>\",\n \"catalog\": \"<string>\",\n \"schema\": \"default\",\n \"register\": true\n }\n}")
.asString();require 'uri'
require 'net/http'
url = URI("http://localhost:8080/api/v1/pipeline")
http = Net::HTTP.new(url.host, url.port)
request = Net::HTTP::Post.new(url)
request["x-api-key"] = '<api-key>'
request["Content-Type"] = 'application/json'
request.body = "{\n \"name\": \"<string>\",\n \"source\": {\n \"schemaProperties\": {\n \"fields\": [\n {\n \"name\": \"<string>\",\n \"type\": \"<string>\"\n }\n ]\n }\n },\n \"protection\": {\n \"purgeSource\": true,\n \"preset\": \"hipaa-safe-harbor\",\n \"presetExempt\": [\n \"<string>\"\n ]\n },\n \"preprocessor\": {},\n \"dataQuality\": {},\n \"transformation\": {},\n \"destination\": {},\n \"unityCatalog\": {\n \"enabled\": false,\n \"comments\": true,\n \"tags\": true,\n \"properties\": true,\n \"lineage\": true,\n \"credentialsSecret\": \"<string>\",\n \"catalog\": \"<string>\",\n \"schema\": \"default\",\n \"register\": true\n }\n}"
response = http.request(request)
puts response.read_body{
"warnings": [
"<string>"
],
"codegenScripts": [
{
"kind": "dataQuality",
"status": "ready",
"pendingReason": "<string>",
"generatedAt": "2023-11-07T05:31:56Z",
"model": "<string>",
"warning": "<string>"
}
]
}{
"error": "<string>"
}Authorizations
Optional API key for authentication (enabled via application.yaml)
Query Parameters
Optional note stored as changeNote on the version this save creates (see the versions endpoints).
Body
Pipeline name
Source configuration (fileAttributes, schemaProperties)
Show child attributes
Show child attributes
Options for field protection. purgeSource is only read when a source field carries protect; preset applies on its own.
Show child attributes
Show child attributes
Optional REST endpoint preprocessor (endpoint, async, bearerToken, apiKey, timeoutMs, batchSize). batchSize defaults to 0 (one call carrying every row, whose response data replaces the payload). With batchSize > 0, row-based payloads are POSTed in batches of that many rows — each request also carrying top-level batch (1-based) and ofBatches — and every response's data.rows are stitched, in call order, into the payload passed downstream.
Data quality rules (AI rules, header validation, schema validation)
Transformation rules (AI transformation, deduplication, row functions).
rowFunctions is an array of { "function", "parameters" }. For "function": "restEndpoint" the positional parameters are [url, mode, timeoutMs, bearerToken, apiKey, batchSize]: mode is "row" (default; one call per row carrying {pipelineName, pipelineToken, row}, a null row in the response removes it) or "batch" (calls carry {pipelineName, pipelineToken, rows}). In batch mode the optional sixth parameter batchSize (default 0) sets the rows per call: 0 makes one call carrying every row; > 0 sends ceil(rows / batchSize) calls of that many rows, in source order, and stitches the responses in call order. Only url is required.
Destination configuration (database, objectStore, kafka, activeMQ, restEndpoint, qdrant, weaviate, milvus, chroma, pgvector).
The database destination selects its engine with exactly one of usePostgres, useMongoDB, useSnowflake, useDatabricks, or useSqlServer (boolean, default false). Snowflake and Databricks additionally take credentialsSecret (a Platform secret) and warehouse (Snowflake warehouse name / Databricks SQL warehouse ID); for Databricks, dbName names the Unity Catalog catalog. For Databricks, warehouse may be omitted when the credentials secret has a warehouse field (aliases httpPath, http_path, DATABRICKS_WAREHOUSE). useSqlServer loads a Microsoft SQL Server or Azure SQL database and requires credentialsSecret (a Platform secret with host, optional port, and authMethod — sql (default): user, password; entraServicePrincipal: clientId, clientSecret, tenantId; entraPassword: user, password; entraManagedIdentity: optional managedIdentityClientId; kerberos: principal plus either password or jaasConfigurationName (keytab)); dbName is the database on that server and schema is usually dbo. It accepts CSV, JSON (single _json field) and XML (single _xml field) sources.
objectStore supports two providers via the provider field: minio (default, uses the built-in MinIO credentials) and s3 (AWS S3). For provider: s3, set destinationBucketOverride (required), credentialsSecret (Vault secret with accessKey/secretKey/region and optional sessionToken), and optionally endpoint (must be https://). Region lives in the credentials secret, not on the objectStore config.
fileFormat accepts parquet (default), orc, or iceberg; with iceberg, writeMode may also be merge, which upserts on the keyFields array (each a destination schema column).
restEndpoint takes endpoint, async, bearerToken, apiKey, timeoutMs and an optional batchSize (default 0 = one call carrying every row). With batchSize > 0, row-based payloads are POSTed in batches of that many rows, each request also carrying top-level batch (1-based) and ofBatches.
Opt-in Unity Catalog sync. On a Databricks destination: metadata push (comments, tags, table properties) and lineage publish. On an object-store destination with fileFormat: iceberg: the table is registered in Unity Catalog as <catalog>.<schema>.<pipeline> (requires credentialsSecret and catalog). Rejected on any other destination when enabled is true. Omitted booleans are treated as true.
Show child attributes
Show child attributes
Response
Pipeline created/updated. warnings is advisory; the pipeline is saved either way (empty array when there is nothing to say). It warns when an object-store Iceberg pipeline's Unity Catalog secret points at a Databricks workspace with catalogMode register or rest, which Databricks cannot serve (set catalogMode managed or use the databricks destination), when a CodeGen script could not be generated at save, and when the save dropped an AI instruction whose script file was kept in the code repository (edited there since its recorded commit, or the repository could not be reached; the file is named so it can be removed by hand). codegenScripts reports, per AI data-quality rule and AI transformation, whether the script generated at save is ready or pending (empty array when the pipeline has neither). Saving a pipeline with an AI instruction can wait for up to two model calls; re-saving with an unchanged instruction and source schema makes none.
Advisory messages about the saved configuration; the pipeline is saved regardless.
One entry per AI kind the pipeline has. pending means no script was generated at save (no AI key, model unreachable, AI disabled, or a JSON/XML source with no column schema); pendingReason says why and the first run generates and stores the script.
Show child attributes
Show child attributes