Task Reference

The tasks that read, transform and write streaming records between pipeline stages, with every parameter and worked examples.

Aggregate

Perform aggregate operations (sum, avg, min, max, count) across multiple fields on streaming pipeline records. Results are emitted as a single record after all input records have been processed.

Parameters

Simple form (single field, single operation)

Parameter	Type	Required	Description
field	string	Yes	Field name to aggregate
op	string	Yes	Operation: sum, avg, min, max

Full form (multiple fields, multiple operations)

Parameter	Type	Required	Description
fields	list	Yes	List of field specifications
fields[].field	string	Yes	Field name
fields[].ops	list	Yes	Operations: sum, avg, min, max, count

Output Keys

Results are emitted as a single record with keys named {field}_{op}:

Key Pattern	Example	Description
{field}_sum	amount_sum	Sum of all values
{field}_avg	amount_avg	Average of all values
{field}_min	amount_min	Minimum value
{field}_max	amount_max	Maximum value
{field}_count	amount_count	Count of values for the field
count	count	Total records processed (always included)

Examples

Simple sum

tasks:
  - name: total
    aggregate:
      field: amount
      op: sum
    outputs:
      - result

  - name: result
    print: "Total: {{amount_sum}}, Records: {{count}}"

Multiple fields with multiple operations

tasks:
  - name: stats
    aggregate:
      fields:
        - field: amount
          ops: [sum, avg, min, max]
        - field: quantity
          ops: [sum, min, max]
    outputs:
      - report

  - name: report
    writecsv:
      path: /data/summary.csv

Aggregate CSV to database

tasks:
  - name: read-data
    read-csv:
      path: /data/daily-metrics.csv
    outputs:
      - aggregate

  - name: aggregate
    aggregate:
      fields:
        - field: cpu_usage
          ops: [avg, max]
        - field: memory_usage
          ops: [avg, max]
        - field: request_count
          ops: [sum]
    outputs:
      - write-db

  - name: write-db
    writedb:
      table: daily_summary

Namespace: aggregate

Perform multi-field aggregation computations from Script scripts.

Functions

Function	Description
aggregate.compute()	Compute aggregate operations on a set of records

aggregate.compute() Parameters

Parameter	Type	Default	Description
fields	list	—	List of {field: string, ops: []string} (required)
records	list	—	List of maps representing records (required)

Supported operations: sum, avg, min, max, count

aggregate.compute() Returns

Field	Type	Description
success	bool	Whether the computation succeeded
result	map	Aggregated values keyed as {field}_{op}, plus count
error	string	Error message if failed

Examples

// Compute sum and average of amounts
var result = aggregate.compute({
    fields: [
        { field: "amount", ops: ["sum", "avg", "min", "max"] },
        { field: "quantity", ops: ["sum"] }
    ],
    records: [
        { amount: 100, quantity: 5 },
        { amount: 200, quantity: 3 },
        { amount: 150, quantity: 8 }
    ]
});
// result.result = { amount_sum: 450, amount_avg: 150, amount_min: 100,
//                   amount_max: 200, quantity_sum: 16, count: 3 }

// Read CSV and aggregate in one script
var data = csv.read({ path: "/data/sales.csv" });
if (data.success) {
    var stats = aggregate.compute({
        fields: [
            { field: "revenue", ops: ["sum", "avg"] },
            { field: "units", ops: ["sum", "min", "max"] }
        ],
        records: data.records
    });
    if (stats.success) {
        console.log("Total revenue: " + stats.result.revenue_sum);
        console.log("Total units: " + stats.result.units_sum);
    }
}

Morph

Execute multi-language scripts for each pipeline record using the Script scripting engine. Similar to Plugin but supports JavaScript, Lua, Starlark, and V8. The user’s script must define a function named morph.

Parameters

Parameter	Type	Required	Default	Description
language	string	No	javascript	Script language: javascript, js, lua, starlark, javascript-v8, v8
execute	string	Yes*	—	Script code executed per record
rows	string	Yes*	—	Per-record script (alias for execute)
end	string	No	—	Script executed after all records are processed

* One of execute or rows is required.

Script Function Contract

The script must define a function named morph that receives an input argument:

function morph(input) {
    // input.data  - the current record (map)
    // input.env   - the pipeline environment variables (map)
}

Available Utility Functions

These functions are injected as globals and available in all supported languages:

Function	Description
write(key, value)	Sink a value to a named output pipe
setmemory(key, value)	Store a value in pipeline memory
getmemory(key)	Retrieve a value from pipeline memory
deletememory(key)	Delete a value from pipeline memory
hasmemory(key)	Check if a key exists in pipeline memory
memorykeys()	Get all memory keys
memorykeyslike(pattern)	Get memory keys matching a pattern
clearmemory()	Clear all pipeline memory

Examples

JavaScript transform

tasks:
  - name: transform
    morph:
      language: javascript
      execute: |
        function morph(input) {
          write("output", {
            full_name: input.data.first_name + " " + input.data.last_name,
            email: input.data.email.toLowerCase()
          });
        }
    outputs:
      - output

Lua transform

tasks:
  - name: transform
    morph:
      language: lua
      execute: |
        function morph(input)
          local price = tonumber(input.data.price)
          write("output", {
            product = input.data.name,
            price_with_tax = price * 1.1
          })
        end
    outputs:
      - output

Aggregation with memory and end code

tasks:
  - name: aggregate
    morph:
      language: javascript
      rows: |
        function morph(input) {
          var category = input.data.category;
          var current = hasmemory(category) ? getmemory(category) : 0;
          setmemory(category, current + parseFloat(input.data.amount));
        }
      end: |
        function morph(input) {
          var keys = memorykeys();
          for (var idx in keys) {
            var cat = keys[idx];
            write("summary", { category: cat, total: getmemory(cat) });
          }
        }
    outputs:
      - summary

Starlark transform

tasks:
  - name: transform
    morph:
      language: starlark
      execute: |
        def morph(input):
          data = input["data"]
          write("output", {
            "event": data["type"].upper(),
            "timestamp": data["ts"]
          })
    outputs:
      - output

Namespace: morph

Execute inline morph scripts from within another Script script. Enables nested script execution for dynamically generating transformation code or running user-provided scripts.

Functions

Function	Description
morph.execute()	Run a morph script inline with specified language and input

morph.execute() Parameters

Parameter	Type	Default	Description
language	string	javascript	Script language: javascript, js, lua, starlark, javascript-v8, v8
code	string	—	Script source defining a morph(input) function (required)
input	any	—	Input data passed to the morph function

morph.execute() Returns

Field	Type	Description
success	bool	Whether the execution succeeded
result	any	Return value from the morph function
error	string	Error message if failed

Examples

// Execute a JavaScript morph inline
var result = morph.execute({
    language: "javascript",
    code: 'function morph(input) { return input.x * 2; }',
    input: { x: 21 }
});
// result = { success: true, result: 42 }

// Execute a Lua morph from JavaScript
var result = morph.execute({
    language: "lua",
    code: 'function morph(input) return input.name .. " processed" end',
    input: { name: "Alice" }
});
// result = { success: true, result: "Alice processed" }

// Combine csv.read + morph.execute for dynamic processing
var data = csv.read({ path: "/data/records.csv" });
if (data.success) {
    var transformCode = 'function morph(input) { return { upper: input.name.toUpperCase(), len: input.name.length }; }';
    for (var i = 0; i < data.records.length; i++) {
        var transformed = morph.execute({
            code: transformCode,
            input: data.records[i]
        });
        console.log(JSON.stringify(transformed.result));
    }
}

Noop

No-op passthrough that forwards each record unchanged to all output sinks. Useful as a placeholder or for testing pipelines.

Parameters

No parameters required.

Examples

Passthrough to next stage

tasks:
  - name: passthrough
    noop:
    outputs:
      - next-stage

Plugin

Execute JavaScript code for each pipeline record. Provides built-in functions for writing to sinks and reading/writing pipeline memory for aggregation patterns.

Parameters

Parameter	Type	Required	Default	Description
execute	string	Yes*	—	JavaScript code executed once per pipeline (one-time mode)
rows	string	Yes*	—	JavaScript code executed per record (row mode)
end	string	No	—	JavaScript code executed after all records are processed

* One of execute or rows is required. If execute is empty, rows is used in per-record mode.

Available JavaScript Functions

Function	Description
write(key, value)	Sink a value to a named output pipe
setmemory(key, value)	Store a value in pipeline memory
getmemory(key)	Retrieve a value from pipeline memory
deletememory(key)	Delete a value from pipeline memory
hasmemory(key)	Check if a key exists in pipeline memory
memorykeys()	Get all memory keys
memorykeyslike(pattern)	Get memory keys matching a pattern
clearmemory()	Clear all pipeline memory

Record data is available as env.data and environment variables as env.env in the JavaScript context.

Examples

Per-record transformation

tasks:
  - name: transform
    plugin:
      rows: |
        var record = env.data;
        var total = parseFloat(record.price) * parseInt(record.quantity);
        write("output", {product: record.name, total: total});
    outputs:
      - output

Aggregation with memory

tasks:
  - name: aggregate
    plugin:
      rows: |
        var record = env.data;
        var category = record.category;
        var current = hasmemory(category) ? getmemory(category) : 0;
        setmemory(category, current + parseFloat(record.amount));
      end: |
        var keys = memorykeys();
        for (var k in keys) {
          write("output", {category: keys[k], total: getmemory(keys[k])});
        }
    outputs:
      - output

Print

Print each pipeline record to stdout using a Liquid template string. Useful for debugging pipelines.

Parameters

Parameter	Type	Required	Default	Description
(direct value)	string	Yes	—	Liquid template string for printing. Record fields available as template variables

Examples

Print record fields

tasks:
  - name: debug-print
    print: "Record: id={{id}}, name={{name}}, amount={{amount}}"

ReadCSV

Read CSV files and stream each row as a key-value record into the data pipeline. The first row is used as the header to name each field.

Parameters

Parameter	Type	Required	Default	Description
(direct value)	string	Yes*	—	Shorthand: file path as the direct parameter value
path	string	Yes*	—	Single CSV file path. Supports Liquid templates and ~ expansion
paths	list	Yes*	—	List of path specs for reading multiple files
paths[].src	string	Yes	—	Source directory. Supports Liquid templates and ~ expansion
paths[].include	list	No	—	Glob patterns to include. Supports Liquid templates
paths[].exclude	list	No	—	Glob patterns to exclude. Supports Liquid templates
wait	duration	No	—	Delay between reading each record (e.g. 100ms, 1s)
entity	map/URL	No	—	Entity definition for record validation
validateentity	string	No	—	Entity name to validate against (when multiple entities defined)

* One of direct value, path, or paths is required.

Examples

Read a single file (shorthand)

tasks:
  - name: ingest
    read-csv: /data/users.csv
    outputs:
      - process

Read a single file

tasks:
  - name: ingest
    read-csv:
      path: /data/users.csv
    outputs:
      - process

Tilde and template path

vars:
  month: "2025-01"
tasks:
  - name: ingest
    read-csv:
      path: "~/exports/{{month}}/report.csv"
    outputs:
      - process

Read with throttle

tasks:
  - name: ingest
    read-csv:
      path: /data/large.csv
      wait: 100ms
    outputs:
      - process

Glob patterns with include/exclude

tasks:
  - name: ingest
    read-csv:
      paths:
        - src: /data/reports/
          include:
            - "sales_*.csv"
            - "revenue_*.csv"
          exclude:
            - "*_draft.csv"
    outputs:
      - process

Multiple source directories

tasks:
  - name: ingest
    read-csv:
      paths:
        - src: /data/region-us/
          include: ["*.csv"]
        - src: /data/region-eu/
          include: ["*.csv"]
    outputs:
      - merge

With entity validation

tasks:
  - name: ingest
    read-csv:
      path: /data/products.csv
      entity:
        name: product
        fields:
          - name: sku
            type: string
            required: true
          - name: price
            type: decimal
            required: true
      validateentity: product
    outputs:
      - store

Pipeline: CSV to database

tasks:
  - name: ingest
    read-csv:
      path: /data/orders.csv
    outputs:
      - write

  - name: write
    writedb:
      table: orders

Namespace: csv

Read and write CSV files from Script scripts.

Functions

Function	Description
csv.read()	Read a CSV file into an array of records
csv.write()	Write records to a CSV file

csv.read() Parameters

Parameter	Type	Default	Description
path	string	—	CSV file path (required)

csv.read() Returns

Field	Type	Description
success	bool	Whether the read succeeded
records	array	Array of key-value maps (one per row)
error	string	Error message if failed

Examples

// Read a CSV file
var result = csv.read({
    path: "/data/users.csv"
});
if (result.success) {
    for (var i = 0; i < result.records.length; i++) {
        console.log(result.records[i].name + ": " + result.records[i].email);
    }
}

ReadDB

Read records from a PostgreSQL database query and stream them into the pipeline. Supports SSL/TLS and vault-based credential retrieval.

Parameters

Parameter	Type	Required	Default	Description
host	string	Yes	—	Database host. Supports Liquid templates
username	string	Yes	—	Database user. Supports Liquid templates
password	string	Yes	—	Vault path for password retrieval. Supports Liquid templates
query	string	Yes	—	SQL query to execute. Supports Liquid templates
database	string	No	username	Database name
port	int	No	5432	Database port
certificate	string	No	—	SSL client certificate (PEM content)
key	string	No	—	SSL client key (PEM content)
issuer	string	No	—	SSL CA certificate (PEM content)
sslmode	string	No	verify-ca	SSL mode
wait	duration	No	—	Delay between reading each record
entity	map/URL	No	—	Entity definition for record validation

Examples

Basic query

tasks:
  - name: read-users
    read-db:
      host: db.example.com
      username: reader
      password: "{{vault.db_password}}"
      query: "SELECT id, name, email FROM users WHERE active = true"
    outputs:
      - process

With SSL

tasks:
  - name: read-secure
    read-db:
      host: db.example.com
      username: reader
      password: "{{vault.db_password}}"
      database: analytics
      port: 5433
      sslmode: verify-full
      certificate: "{{vault.db_cert}}"
      key: "{{vault.db_key}}"
      issuer: "{{vault.db_ca}}"
      query: "SELECT * FROM events"
    outputs:
      - process

Namespace: db

Query and execute statements against PostgreSQL databases from Script scripts.

Functions

Function	Description
db.query()	Execute a SELECT query and return records
db.execute()	Execute an INSERT/UPDATE/DELETE statement

db.query() Parameters

Parameter	Type	Default	Description
connection	string	—	PostgreSQL connection string (required)
query	string	—	SQL query to execute (required)

db.query() Returns

Field	Type	Description
success	bool	Whether the query succeeded
records	array	Array of key-value maps (one per row)
error	string	Error message if failed

Examples

// Query a PostgreSQL database
var result = db.query({
    connection: "postgres://user:pass@host:5432/db?sslmode=disable",
    query: "SELECT * FROM users"
});
if (result.success) {
    for (var i = 0; i < result.records.length; i++) {
        console.log(result.records[i].name + ": " + result.records[i].email);
    }
}

ReadEntity

Read records from a Data API entity store using a bulk query. Requires entity definition files and a database pool configuration.

Parameters

Parameter	Type	Required	Default	Description
files	list	Yes	—	Entity definition file paths
dbpool	map	Yes	—	Database pool configuration
query	string	Yes	—	Entity query to execute
entity	map/URL	No	—	Entity definition for record validation

Examples

Read from entity store

tasks:
  - name: read-entities
    readentity:
      files:
        - /schemas/user.yaml
      dbpool:
        host: db.example.com
        port: 5432
        username: reader
        password: "{{vault.db_password}}"
        database: app
      query: "SELECT * FROM user WHERE active = true"
    outputs:
      - process

Namespace: entity

Query and mutate Data API entity stores from Script scripts.

Functions

Function	Description
entity.query()	Query records from an entity store
entity.mutate()	Write records to an entity store

Note: Entity operations require a pipeline environment with Data API EntityManager. The namespace functions are stubs in Script standalone mode and return errors directing to the YAML pipeline interface.

entity.query() Parameters

Parameter	Type	Default	Description
files	list	—	Entity definition file paths
dbpool	map	—	Database pool configuration
query	string	—	Entity query to execute

entity.query() Returns

Field	Type	Description
success	bool	Whether the query succeeded
records	array	Array of entity records
error	string	Error message (stub always returns error)

Examples

// Entity queries are not available in Script standalone mode.
// Use the YAML pipeline interface (readentity) instead.
var result = entity.query({
    files: ["/schemas/user.yaml"],
    dbpool: {},
    query: "SELECT * FROM user"
});
// result = { success: false, error: "entity operations require pipeline environment" }

ReadEntityDatastore

Read records from a datastore entity using metadata service configuration. Lazily initializes the entity manager and caches it for reuse.

Parameters

Parameter	Type	Required	Default	Description
datastore	string	Yes	—	Datastore name
query	string	Yes	—	Entity query to execute
metaurl	string	No	tenantmeta.url	Metadata service URL override
entity	map/URL	No	—	Entity definition for record validation

Examples

Read from datastore

tasks:
  - name: read-from-datastore
    readentity-in-datastore:
      datastore: customer-db
      query: "SELECT * FROM orders WHERE status = 'pending'"
    outputs:
      - process

Namespace: entity

See readentity for the entity namespace documentation. The entity.query() and entity.mutate() functions are shared across ReadEntity, ReadEntityDatastore, and WriteEntityDatastore tasks.

Examples

// Entity queries are not available in Script standalone mode.
// Use the YAML pipeline interface (readentity-in-datastore) instead.
var result = entity.query({
    datastore: "customer-db",
    query: "SELECT * FROM orders"
});
// result = { success: false, error: "entity operations require pipeline environment" }

ReadExcel

Read Excel spreadsheet files and stream each row as a pipeline record. The first row of each sheet is used as the header.

Parameters

Parameter	Type	Required	Default	Description
(direct value)	string	Yes*	—	Shorthand: file path
path	string	Yes*	—	Excel file path. Supports Liquid templates and ~ expansion
paths	list	Yes*	—	List of path specs with src, include, exclude
wait	duration	No	—	Delay between records
entity	map/URL	No	—	Entity definition for record validation

* One of direct value, path, or paths is required.

Examples

Read an Excel file

tasks:
  - name: read-report
    read-excel:
      path: /data/report.xlsx
    outputs:
      - process

Read multiple Excel files by pattern

tasks:
  - name: read-all
    read-excel:
      paths:
        - src: /data/imports/
          include: ["*.xlsx"]
          exclude: ["~$*"]
    outputs:
      - process

Namespace: excel

Read and write Excel files from Script scripts.

Functions

Function	Description
excel.read()	Read an Excel file into an array of records
excel.write()	Write records to an Excel file

excel.read() Parameters

Parameter	Type	Default	Description
path	string	—	Excel file path (required)
sheet	string	—	Sheet name (reads all sheets if omitted)

excel.read() Returns

Field	Type	Description
success	bool	Whether the read succeeded
records	array	Array of key-value maps (one per row)
error	string	Error message if failed

Examples

// Read records from an Excel file
var result = excel.read({ path: "/data/report.xlsx", sheet: "Sales" });
if (result.success) {
    for (var i = 0; i < result.records.length; i++) {
        console.log(result.records[i].product + ": " + result.records[i].amount);
    }
}

ReadParquet

Read Apache Parquet files and stream each row as a pipeline record. All values are converted to strings.

Parameters

Parameter	Type	Required	Default	Description
(direct value)	string	Yes*	—	Shorthand: file path
path	string	Yes*	—	Parquet file path. Supports Liquid templates and ~ expansion
paths	list	Yes*	—	List of path specs with src, include, exclude
wait	duration	No	—	Delay between records
entity	map/URL	No	—	Entity definition for record validation

* One of direct value, path, or paths is required.

Examples

Read a Parquet file

tasks:
  - name: read-data
    read-parquet:
      path: /data/events.parquet
    outputs:
      - process

Namespace: parquet

Read and write Parquet files from Script scripts.

Functions

Function	Description
parquet.read()	Read a Parquet file into an array of records
parquet.write()	Write records to a Parquet file

parquet.read() Parameters

Parameter	Type	Default	Description
path	string	—	Parquet file path (required)

parquet.read() Returns

Field	Type	Description
success	bool	Whether the read succeeded
records	array	Array of key-value maps (one per row)
error	string	Error message if failed

Examples

// Read records from a Parquet file
var result = parquet.read({ path: "/data/events.parquet" });
if (result.success) {
    for (var i = 0; i < result.records.length; i++) {
        console.log(JSON.stringify(result.records[i]));
    }
}

ReadREST

Make an HTTP request and parse the CSV response into pipeline records. The first row of the response is used as the header.

Parameters

Parameter	Type	Required	Default	Description
verb	string	Yes	—	HTTP method (GET, POST, etc.). Supports Liquid templates
url	string	Yes	—	REST endpoint URL. Supports Liquid templates
headers	map	No	—	HTTP request headers. Supports Liquid templates
timeout	string	No	10s	Connection timeout
wait	duration	No	—	Delay between records
entity	map/URL	No	—	Entity definition for record validation

Examples

Fetch CSV from a REST endpoint

tasks:
  - name: fetch-data
    read-rest:
      verb: GET
      url: "https://api.example.com/export.csv"
      headers:
        Authorization: "Bearer {{token}}"
      timeout: 30s
    outputs:
      - process

Namespace: rest

Make HTTP requests from Script scripts.

Functions

Function	Description
rest.request()	Make an HTTP request and return the response

rest.request() Parameters

Parameter	Type	Default	Description
url	string	—	Request URL (required)
verb	string	GET	HTTP method
headers	map	—	HTTP request headers
timeout	string	10s	Connection timeout

rest.request() Returns

Field	Type	Description
success	bool	Whether the request succeeded
status	int	HTTP status code
result	any	Response body
error	string	Error message if failed

Note: In pipeline mode, the ReadREST response is parsed as CSV into records. In Script standalone mode, rest.request returns the raw response body.

Examples

// Fetch data from a REST endpoint
var result = rest.request({
    url: "https://api.example.com/data.csv"
});
if (result.success) {
    console.log("Status: " + result.status);
    console.log("Body: " + result.result);
}

ReadRESTJSON

Make an HTTP request with an optional JSON payload and sink the full response as a single record.

Parameters

Parameter	Type	Required	Default	Description
verb	string	Yes	—	HTTP method. Supports Liquid templates
url	string	Yes	—	REST endpoint URL. Supports Liquid templates
payload	string	No	""	Request body template. Supports Liquid templates
headers	map	No	—	HTTP request headers. Supports Liquid templates
timeout	string	No	10s	Connection timeout
wait	duration	No	—	Delay between records

Examples

POST with JSON payload

tasks:
  - name: fetch-json
    read-rest-json:
      verb: POST
      url: "https://api.example.com/query"
      payload: '{"filter": "{{filter_value}}"}'
      headers:
        Content-Type: application/json
    outputs:
      - process

Namespace: rest

See readrest for the rest namespace documentation. The rest.request() function is shared across ReadREST, ReadRESTJSON, and ReadRESTJSONPipe tasks.

Examples

// POST a JSON payload and read the response
var result = rest.request({
    verb: "POST",
    url: "https://api.example.com/data",
    body: { key: "value" },
    headers: { "Authorization": "Bearer token" }
});
if (result.success) {
    console.log("Status: " + result.status);
    console.log("Response: " + JSON.stringify(result.result));
}

ReadStdout

Execute a shell command and parse its CSV stdout output into pipeline records. The first line is used as the header.

Parameters

Parameter	Type	Required	Default	Description
command	string	Yes	—	Shell command to execute
wait	duration	No	—	Delay between records
entity	map/URL	No	—	Entity definition for record validation

Examples

Read from command output

tasks:
  - name: read-process-output
    read-stdout:
      command: "cat /var/log/metrics.csv"
    outputs:
      - process

Namespace: stdout

Execute shell commands and parse CSV output from Script scripts.

Functions

Function	Description
stdout.exec()	Execute a command and parse its CSV output into records

stdout.exec() Parameters

Parameter	Type	Default	Description
command	string	—	Shell command to execute (required)

stdout.exec() Returns

Field	Type	Description
success	bool	Whether the execution succeeded
records	array	Array of key-value maps parsed from CSV output
output	string	Raw command output
error	string	Error message if failed

Examples

// Execute a command and parse CSV output
var result = stdout.exec({
    command: "echo 'name,age\nAlice,30\nBob,25'"
});
if (result.success) {
    for (var i = 0; i < result.records.length; i++) {
        console.log(result.records[i].name + " is " + result.records[i].age);
    }
}

Replicate

Replicate each incoming record unchanged to all output sinks. Useful for fan-out patterns where the same data feeds into multiple downstream tasks.

Parameters

No parameters required.

Examples

Fan-out to multiple sinks

tasks:
  - name: fan-out
    replicate:
    outputs:
      - write-csv
      - write-db
      - write-analytics

Namespace: compute

See sum for the full compute namespace documentation.

compute.replicate() Parameters

Parameter	Type	Default	Description
record	map	—	Record to replicate (required)

compute.replicate() Returns

Field	Type	Description
success	bool	Whether the replication succeeded
result	map	Deep copy of the input record
error	string	Error message if failed

Examples

// Create a deep copy of a record
var result = compute.replicate({
    record: { name: "Alice", age: 30, role: "admin" }
});
if (result.success) {
    var copy = result.result;
    console.log(copy.name); // "Alice"
}

ReadRESTJSONPipe

Full-featured REST API client for data pipelines. Supports service authentication, retry logic, polling with wait-for conditions, and post-request JavaScript processing. Parses JSON array or object responses into individual pipeline records.

Parameters

Parameter	Type	Required	Default	Description
type	string	Yes	—	HTTP method
url	string	Yes	—	REST endpoint URL. Supports Liquid templates
payload	string	No	—	Request body template
payloadkey	string	No	—	Key to extract from payload
datakey	string	No	—	Key to extract from response JSON
setbody	string	No	—	Key name to store response body
setstatuscode	string	No	—	Key name to store HTTP status code
headers	map	No	—	HTTP request headers
sendheaders	map	No	—	Headers to include in output
timeout	string	No	10s	Connection timeout
serviceauth	bool	No	false	Use service-to-service JWT auth
mlresponse	bool	No	false	Multi-line response mode
failonerror	bool	No	true	Fail pipeline on HTTP error
doretry	bool	No	false	Enable retry on failure
retrycount	int	No	-1	Number of retries (-1 = unlimited)
waitfor	map	No	—	Polling config with callback expression
postrequest	map	No	—	Post-processing JavaScript expression

Examples

Basic REST pipeline

tasks:
  - name: fetch-records
    read-rest-json-pipe:
      type: GET
      url: "https://api.example.com/records"
      datakey: data
    outputs:
      - process

With retry and service auth

tasks:
  - name: fetch-secure
    read-rest-json-pipe:
      type: POST
      url: "https://internal-api/bulk"
      payload: '{"ids": {{ids}}}'
      serviceauth: true
      doretry: true
      retrycount: 3
      timeout: 30s
    outputs:
      - process

Namespace: rest

See readrest for the rest namespace documentation. The rest.request() function is shared across ReadREST, ReadRESTJSON, and ReadRESTJSONPipe tasks.

Note: Pipeline-only features (Vault service authentication, JavaScript post-processing, retry logic, and wait-for polling) are not available in Script standalone mode. Use rest.request for direct HTTP calls.

Examples

// POST with JSON body and auth header
var result = rest.request({
    verb: "POST",
    url: "https://api.example.com/data",
    body: { key: "value" },
    headers: { "Authorization": "Bearer token" }
});
if (result.success) {
    console.log("Status: " + result.status);
    console.log("Response: " + JSON.stringify(result.result));
}

Sum

Add a fixed numeric value to a field in each pipeline record. The result is emitted per record.

Parameters

Parameter	Type	Required	Default	Description
header	string	Yes	—	Column name containing the numeric value
value	float	Yes	—	Value to add to each record’s field

Examples

Add a fixed value to a field

tasks:
  - name: add-tax
    sum:
      header: price
      value: 10.5
    outputs:
      - process

Namespace: compute

Numeric and string computation functions for Script scripts.

Functions

Function	Description
compute.sum()	Add two numeric values
compute.concat()	Concatenate two strings
compute.replicate()	Create a deep copy of a record

compute.sum() Parameters

Parameter	Type	Default	Description
value	number	—	Base value (required)
add	number	—	Value to add (required)

compute.sum() Returns

Field	Type	Description
success	bool	Whether the computation succeeded
result	number	Sum of the two values
error	string	Error message if failed

Examples

// Add two numeric values
var result = compute.sum({ value: 42, add: 8 });
if (result.success) {
    console.log(result.result); // 50
}

SumStr

Concatenate a fixed string value to a field in each pipeline record. The result is emitted per record.

Parameters

Parameter	Type	Required	Default	Description
header	string	Yes	—	Column name containing the string value
value	string	Yes	—	String to concatenate

Examples

Add a prefix to a field

tasks:
  - name: add-prefix
    sumstr:
      header: name
      value: "user_"
    outputs:
      - process

Namespace: compute

See sum for the full compute namespace documentation.

compute.concat() Parameters

Parameter	Type	Default	Description
value	string	—	Base string (required)
add	string	—	String to concatenate (required)

compute.concat() Returns

Field	Type	Description
success	bool	Whether the computation succeeded
result	string	Concatenated string
error	string	Error message if failed

Examples

// Concatenate two strings
var result = compute.concat({ value: "Hello", add: " World" });
if (result.success) {
    console.log(result.result); // "Hello World"
}

WriteCSV

Write pipeline records to a CSV file. The header row is inferred from the first record’s keys unless pre-defined.

Parameters

Parameter	Type	Required	Default	Description
(direct value)	string	Yes*	—	Shorthand: file path
path	string	Yes*	—	Output CSV file path
overwrite	bool	No	false	Truncate file before writing (vs append)
headers	list	No	—	Pre-defined column headers (inferred from first record if omitted)
entity	map/URL	No	—	Entity definition for record validation

* One of direct value or path is required.

Examples

Write to CSV

tasks:
  - name: write-output
    writecsv:
      path: /output/results.csv

Overwrite existing file

tasks:
  - name: write-fresh
    writecsv:
      path: /output/results.csv
      overwrite: true
      headers:
        - id
        - name
        - email

Namespace: csv

See readcsv for the full csv namespace documentation.

csv.write() Parameters

Parameter	Type	Default	Description
path	string	—	Output CSV file path (required)
records	array	—	Array of key-value maps to write (required)
headers	list	—	Column headers (inferred if omitted)
overwrite	bool	false	Truncate file before writing

csv.write() Returns

Field	Type	Description
success	bool	Whether the write succeeded
count	int	Number of records written
error	string	Error message if failed

Examples

// Write records to a CSV file
var result = csv.write({
    path: "/output/data.csv",
    records: [
        { name: "Alice", age: "30" },
        { name: "Bob", age: "25" }
    ],
    overwrite: true
});
if (result.success) {
    console.log("Wrote " + result.count + " records");
}

WriteDB

Write pipeline records to a PostgreSQL database by executing a templated SQL query per record. Supports SSL/TLS and vault-based credential retrieval.

Parameters

Parameter	Type	Required	Default	Description
host	string	Yes	—	Database host. Supports Liquid templates
username	string	Yes	—	Database user. Supports Liquid templates
password	string	Yes	—	Vault path for password retrieval
query	string	Yes	—	SQL INSERT/UPDATE template. Supports Liquid templates with record fields
database	string	No	username	Database name
port	int	No	5432	Database port
certificate	string	No	—	SSL client certificate
key	string	No	—	SSL client key
issuer	string	No	—	SSL CA certificate
sslmode	string	No	verify-full	SSL mode
maxconnections	string	No	3	Max connection pool size
connectionlifetime	string	No	1s	Connection max lifetime
connectimeout	string	No	1	Connect timeout in seconds
entity	map/URL	No	—	Entity definition for record validation

Examples

Write records to database

tasks:
  - name: write-to-db
    writedb:
      host: db.example.com
      username: writer
      password: "{{vault.db_password}}"
      database: analytics
      query: "INSERT INTO events (id, name, value) VALUES ('{{id}}', '{{name}}', '{{value}}')"

Namespace: db

See readdb for the full db namespace documentation.

db.execute() Parameters

Parameter	Type	Default	Description
connection	string	—	PostgreSQL connection string (required)
query	string	—	SQL statement to execute (required)

db.execute() Returns

Field	Type	Description
success	bool	Whether the execution succeeded
affected	int	Number of rows affected
error	string	Error message if failed

Examples

// Execute an INSERT statement
var result = db.execute({
    connection: "postgres://user:pass@host:5432/db?sslmode=disable",
    query: "INSERT INTO users (name, email) VALUES ('Alice', '[email protected]')"
});
if (result.success) {
    console.log("Affected rows: " + result.affected);
}

WriteEntityDatastore

Write pipeline records to a datastore entity using a JSON record template. Executes bulk mutations via the Data API entity manager.

Parameters

Parameter	Type	Required	Default	Description
datastore	string	Yes	—	Datastore name
record-template	string	Yes	—	JSON template for record. Supports Liquid templates with record fields
metaurl	string	No	—	Metadata service URL override

Examples

Write entities to datastore

tasks:
  - name: write-entities
    writeentity-in-datastore:
      datastore: customer-db
      record-template: |
        {
          "entity": "order",
          "data": {
            "id": "{{id}}",
            "customer": "{{customer}}",
            "amount": "{{amount}}"
          }
        }

Namespace: entity

See readentity for the full entity namespace documentation.

entity.mutate() Parameters

Parameter	Type	Default	Description
datastore	string	—	Datastore name (required)
record	map	—	Entity record data (required)

entity.mutate() Returns

Field	Type	Description
success	bool	Whether the mutation succeeded
error	string	Error message (stub always returns error)

Note: Entity operations require a pipeline environment with Data API EntityManager. The entity.mutate function is a stub in Script standalone mode.

Examples

// Entity mutations are not available in Script standalone mode.
// Use the YAML pipeline interface (writeentity-in-datastore) instead.
var result = entity.mutate({
    datastore: "customer-db",
    record: { id: "1" }
});
// result = { success: false, error: "entity operations require pipeline environment" }

WriteExcel

Write pipeline records to an Excel spreadsheet. Creates or appends to an Excel file. The sheet name is derived from the input pipeline name.

Parameters

Parameter	Type	Required	Default	Description
(direct value)	string	Yes*	—	Shorthand: file path
path	string	Yes*	—	Output Excel file path
overwrite	bool	No	false	Create new file (vs append to existing)
entity	map/URL	No	—	Entity definition for record validation

* One of direct value or path is required.

Examples

Write to Excel

tasks:
  - name: write-excel
    writeexcel:
      path: /output/report.xlsx
      overwrite: true

Namespace: excel

See readexcel for the full excel namespace documentation.

excel.write() Parameters

Parameter	Type	Default	Description
path	string	—	Output Excel file path (required)
sheet	string	—	Sheet name
records	array	—	Array of key-value maps to write (required)
overwrite	bool	false	Create new file (vs append)

excel.write() Returns

Field	Type	Description
success	bool	Whether the write succeeded
count	int	Number of records written
error	string	Error message if failed

Examples

// Write records to an Excel file
var result = excel.write({
    path: "/output/report.xlsx",
    sheet: "Results",
    records: [
        { name: "Alice", score: "95" },
        { name: "Bob", score: "87" }
    ],
    overwrite: true
});
if (result.success) {
    console.log("Wrote " + result.count + " records");
}

WriteParquet

Write pipeline records to an Apache Parquet file. Schema is inferred from the first record’s keys. All fields are written as strings.

Parameters

Parameter	Type	Required	Default	Description
(direct value)	string	Yes*	—	Shorthand: file path
path	string	Yes*	—	Output Parquet file path
overwrite	bool	No	false	Truncate file before writing
entity	map/URL	No	—	Entity definition for record validation

* One of direct value or path is required.

Examples

Write to Parquet

tasks:
  - name: write-parquet
    writeparquet:
      path: /output/events.parquet
      overwrite: true

Namespace: parquet

See readparquet for the full parquet namespace documentation.

parquet.write() Parameters

Parameter	Type	Default	Description
path	string	—	Output Parquet file path (required)
records	array	—	Array of key-value maps to write (required)

parquet.write() Returns

Field	Type	Description
success	bool	Whether the write succeeded
count	int	Number of records written
error	string	Error message if failed

Examples

// Write records to a Parquet file
var result = parquet.write({
    path: "/output/data.parquet",
    records: [
        { id: "1", name: "Alice" },
        { id: "2", name: "Bob" }
    ]
});
if (result.success) {
    console.log("Wrote " + result.count + " records");
}

WriteREST

Write each pipeline record to a REST API endpoint by making an HTTP request per record. Requests are executed asynchronously in goroutines.

Parameters

Parameter	Type	Required	Default	Description
url	string	Yes	—	REST endpoint URL. Supports Liquid templates with record fields
verb	string	No	post	HTTP method. Supports Liquid templates
payload	string	No	—	Request body template. Supports Liquid templates with record fields
headers	map	No	—	HTTP request headers. Supports Liquid templates
waitduration	string	No	10s	Delay between requests
connectiontimeout	string	No	10s	Connection timeout
entity	map/URL	No	—	Entity definition for record validation

Examples

POST records to an API

tasks:
  - name: post-records
    writerest:
      url: "https://api.example.com/records"
      verb: post
      payload: '{"id": "{{id}}", "name": "{{name}}"}'
      headers:
        Content-Type: application/json
        Authorization: "Bearer {{token}}"
      waitduration: 100ms

Namespace: rest

See readrest for the full rest namespace documentation. The rest.request() function is used for both reading and writing via REST endpoints.

Examples

// POST JSON data to a REST endpoint
var result = rest.request({
    verb: "POST",
    url: "https://api.example.com/users",
    body: { name: "Alice", email: "[email protected]" },
    headers: { "Content-Type": "application/json" }
});
if (result.success) {
    console.log("Status: " + result.status);
    console.log("Response: " + JSON.stringify(result.result));
}