General

This page provides an overview of methods for managing various pipeline behaviors.

Some of these methods take pipeline operator's metadata as an input which the Stream Processor exposes across many APIs, such as state and pushing data further in a pipeline within asynchronous nodes. A pipeline operator's metadata is a dictionary containing the following keys.

Metadata Keys

These keys will be extended as new capabilities are added to the Stream Processor engine.

key

An optional partition key for the events in this message. This is used to group related collections of data based on a common attribute. Events that do not share the same key are considered independent and may be processed independently of other keys. (any type)

window

Indicates the start time or index of a window of data (temporal)

offset

An event offset is a number or time that represents the current message offset to the stream reader. (int or temporal)

This value will be used to recover data if messages are lost downstream. If messages need to be replayed, the message offset will be used to indicate the last successfully processed message.

Pipeline Lifecycle

Functions for starting and tearing down pipelines, and for manually driving data into or through a running pipeline, typically from within custom operators or via external triggers such as the REST API.

Run

Runs a pipeline.

Multiple Workers

Depending on the deployment configuration, this function can run a pipeline in the current worker process or distribute work across several workers.

qPython

.qsp.run[pipe]

Parameters:

name	type	description
pipe	#.qsp.pipe	The pipeline to install and run in the current stream processor.

Examples:

Example 1: Running a pipeline that reads from an expression, then writes to the console.

.qsp.run
    .qsp.read.fromExpr["til 10"]
    .qsp.write.toConsole[]

0 1 2 3 4 5 6 7 8 9

Example 2: Running multiple pipelines.

pipelineA: .qsp.read.fromExpr["til 10"] .qsp.write.toConsole["A) ";.qsp.use ``timestamp!(::;`none)]
pipelineB: .qsp.read.fromExpr["til 20"] .qsp.write.toConsole["B) ";.qsp.use ``timestamp!(::;`none)]
pipelineC: .qsp.read.fromExpr["til 30"] .qsp.write.toConsole["C) ";.qsp.use ``timestamp!(::;`none)]

.qsp.run (pipelineA; pipelineB; pipelineC)

A) 0 1 2 3 4 5 6 7 8 9
B) 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
C) 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

sp.run(pipelines)

Parameters:

name	type	description
pipelines	Pipeline	The pipeline to install and run in the current stream processor.

Examples:

Example 1: Running a pipeline that reads from a nullary function, then writes to the console.

>>> from kxi import sp

>>> sp.run(sp.read.from_expr(lambda: range(10)) | sp.write.to_console())

0 1 2 3 4 5 6 7 8 9

Example 2: Running multiple units.

>>> from kxi import sp

>>> pipeline_a = sp.read.from_expr(lambda: range(10)) | sp.write.to_console('A) ', timestamp='none')
>>> pipeline_b = sp.read.from_expr(lambda: range(20)) | sp.write.to_console('B) ', timestamp='none')
>>> pipeline_c = sp.read.from_expr(lambda: range(30)) | sp.write.to_console('C) ', timestamp='none')
>>> sp.run(pipeline_a, pipeline_b, pipeline_c)

A) 0 1 2 3 4 5 6 7 8 9
B) 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19
C) 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29

Example 3: Argument unpacking can be used to run a list/tuple of units.

>>> from kxi import sp
>>> from random import sample

>>> source = sp.read.from_expr(lambda: sample(range(20, 50, 2), 10))
>>> pipelines = (
    source | sp.map(lambda x: x * 2) | sp.write.to_console('double  ) ', timestamp='none'),
    source | sp.write.to_console('original) ', timestamp='none'),
    source | sp.map(lambda x: x / 2) | sp.write.to_console('half    ) ', timestamp='none'),
    )
>>> sp.run(*pipelines)

double  ) 68 72 84 52 44 88 96 80 56 48
original) 34 36 42 26 22 44 48 40 28 24
half    ) 17 18 21 13 11 22 24 20 14 12

Teardown

Tears down a pipeline, removing all state and timers.

qPython

.qsp.teardown[]

sp.teardown()

On Teardown action

Any On Teardown handlers or subscribers to teardown will be called.

Push

Publishes data to all downstream operators in the pipeline.

qPython

.qsp.push[op;md;data]

Parameters:

name	type	description	default
op	.qsp.op	The current operator configuration.	Required
md	.qsp.message.metadata	Metadata for the message being emitted from the current asynchronous operator.	Required
data	.qsp.message.event or .qsp.message.event[]	The data to be published from this async operator.	Required

sp.push(operator, metadata, data)

Parameters:

name	type	description	default
operator	OperatorSpecifier	The operator, specified as a configuration dictionary or name.	Required
metadata	Metadata	Metadata for the message being emitted from the operator.	Required
data	Any	The data to be pushed to the downstream pipeline.	Required

Returns:

Publishes data to all downstream operators in the pipeline, and returns data.

Asynchronous Operators

Should be used from asynchronous operators in a pipeline to continue the flow of data in the pipeline. It is only required for sp.apply. Other operators are synchronous and any data returned will flow to the next operator.

Trigger Write

Triggers batch pipeline writer(s).

Write triggering and pipeline units

Write triggering is not synchronized with pipeline unit terms and rolling - roll the unit instead. See write triggering and units.

qPython

.qsp.triggerWrite[]

Parameters:

name	type	description
opIDs	symbol, symbol[] or null	List of operators (or operator IDs) to trigger write. If left empty all writers triggered.

For non-streaming writer operators, triggers a write. For streaming writer operators this will have no effect, e.g. for the direct write mode of Database Writer.

Returns:

Type	Description
null

Examples:

Example 1: Create a writer named dbWriter with directWrite enabled.

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.v2.write.toDatabase[`trade; `$":db-one"; .qsp.use `name`directWrite]!(`dbWriter;1b)]

Example 2: Publish some data, then trigger a write down of the trade table.

publish[([] sym:`KX`KX; timestamp: 2#.z.p; price:100.1 99.9; size: 100 200)]
.qsp.triggerWrite[]

Example 3: Explicitly trigger a write down for the writer named dbWriter.

.qsp.triggerWrite[enlist `dbWriter]

sp.trigger_write()

Parameters:

name	type	description	default
operatorIDs	List of operators (or operator IDs) to trigger write. If left empty all writers triggered.	Not required

Triggers a write for non streaming writer operators. For streaming writer operators this will have no effect.

Examples:

Example 1: Trigger a write-down using sp.write.to_database v2 in direct write mode.

The writer name can be explicitly set using the operator name parameter. To retrieve the name of an existing operator use the /pipeline/{id}/describe REST API.

>>> from kxi import sp
>>> import pandas as pd
>>> import pykx as kx

>>> sp.run(sp.read.from_callback('publish')
    | sp.write.to_database(table='tableName', database='dbName',
        directWrite=True, api_version=2, name='dbWriter'))

>>> data = pd.DataFrame({
        'name': ['a', 'b', 'c'],
        'id': [1, 2, 3],
        'quantity': [4, 5, 6]
    })

>>> kx.q('publish', data)

# trigger all writers (only db writer here anyway).
>>> sp.trigger_write()

# Explicitly trigger specific writer(s).
>>> sp.trigger_write(["dbWriter"])

Trigger Read

You can configure pull reader(s) for triggering. Configured readers have their read behavior changed from one shot reads, to reading on a user provided trigger. The following shows the q and python trigger APIs. A REST API is also available.

For pull reader operators (configured for triggering), triggers another read. For non pull readers or pull readers not configured for triggering this has no effect.

qPython

.qsp.triggerRead[]

Parameters:

name	q type	description
opIDs	symbol, symbol[] or null	A list of operators (or operator IDs) to trigger read. If left empty all readers are triggered.

Returns:

Type
null

Example:

This pipeline will not execute anything on startup. Once the trigger API is called, the expression reader will execute the expression and push data down the pipeline. If it is called again it will execute again and push.

 .qsp.run
     pipe: .qsp.read.fromExpr[{"sym,price\nKX,100\n"}; .qsp.use `trigger`name!(`api; `expr)]
     .qsp.decode.csv[]
     .qsp.write.toVariable[`.test.cache]

If you then want to execute the expression and push downstream, you can run

// No operators provided as an argument so all readers (configured for triggering) are triggered.
// In this case only a single reader configured for triggering.
 .qsp.triggerRead[]

or to explicitly trigger the expression reader call the API with operator ID/name you want triggered:

.qsp.triggerRead[`expr]

sp.trigger_read()

Parameters:

name	type	description	default
operatorIDs	A list of operators (or operator IDs) to trigger read. If left empty all pull readers are triggered.	Not required

Returns:

Type
null

Example:

Trigger a single read using an expression reader and sp.trigger_read.

The reader will not output any data until triggered. When more than one reader is present, specific readers can be triggered individually using the operator's name, which can be explicitly set using the operator's name parameter when it is defined. To retrieve the name of an existing operator use the /pipeline/{id}/describe REST API.

>>> import pykx as kx
>>> from kxi import sp

>>> expr = '"sym,price\nKX,100\n"'
>>> sp.run(sp.read.from_expr(expr, name='myExpr', trigger='api')
        | sp.decode.csv('')
        | sp.write.to_variable('.test.cache'))

# trigger all readers (only a single trigger configured reader anyway).
>>> sp.trigger_read()

# Explicitly trigger specific reader(s).
>>> sp.trigger_read(['myExpr'])

Units

Functions for monitoring and rolling pipeline units - independent sub-graphs of a pipeline that process data in a series of terms. A REST API is also available.

Get Units

Lists the units in the running pipeline.

qPython

.qsp.getUnits[]

Returns:

type	description
table	Keyed on id (unit ID); nodes holds the unit's operator IDs.

Example:

.qsp.run (
    .qsp.read.fromCallback[`pub1] .qsp.write.toVariable[`output1];
    .qsp.read.fromCallback[`pub2] .qsp.write.toVariable[`output2])

q).qsp.getUnits[]
id           | nodes
-------------| -------------------------------
callback_pub1| `callback_pub1`variable_output1
callback_pub2| `callback_pub2`variable_output2

sp.get_units()

Returns:

type	description
pykx.KeyedTable	Keyed on id (unit ID); nodes holds the unit's operator IDs.

Example:

>>> from kxi import sp

>>> sp.run(sp.read.from_callback('pub1') | sp.write.to_variable('output1'),
        sp.read.from_callback('pub2') | sp.write.to_variable('output2'))

>>> sp.get_units()
pykx.KeyedTable(pykx.q('
id           | nodes
-------------| ------------------------------
callback_pub1| callback_pub1 variable_output1
callback_pub2| callback_pub2 variable_output2
'))

Get Unit Terms

Summarises a unit: its state, whether it can roll, its current and previous term IDs, and a terms list (newest first) giving each term's role, state, timespan and per-operator nodes payloads. state and canRoll are the current term's.

qPython

.qsp.getUnitTerms[unitId]

Parameters:

name	q type	description
unitId	symbol	Unit ID, as returned by .qsp.getUnits.

Returns:

type	description
dictionary	Keys unitId, state, canRoll, currentTermId, previousTermId, terms.

Example:

q).qsp.getUnitTerms[`callback_pub1]
unitId        | `callback_pub1
state         | `running
canRoll       | 1b
currentTermId | 1
previousTermId| 0
terms         | +`termId`role`state`canRoll`startedAt`endedAt`nodes!(1 0;`current`previous;..)

sp.get_unit_terms(unit_id)

Parameters:

name	type	description
unit_id	str	Unit ID, as returned by sp.get_units.

Returns:

type	description
pykx.Dictionary	Keys unitId, state, canRoll, currentTermId, previousTermId, terms.

Example:

>>> sp.get_unit_terms('callback_pub1')
pykx.Dictionary(pykx.q('
unitId        | `callback_pub1
state         | `running
canRoll       | 1b
currentTermId | 1
previousTermId| 0
terms         | +`termId`role`state`canRoll`startedAt`endedAt`nodes!(1 0;`current`previous;..)
'))

Get Unit Term

Returns one term of a unit: the unit ID and the term's terms entry from Get Unit Terms.

qPython

.qsp.getUnitTerm[unitId; termId]

Parameters:

name	q type	description
unitId	symbol	Unit ID, as returned by .qsp.getUnits.
termId	long	Term ID of the unit's current or previous term.

Returns:

type	description
dictionary	Keys unitId, termId, role, state, canRoll, startedAt, endedAt, nodes.

Example:

q).qsp.getUnitTerm[`callback_pub1; 0]
unitId   | `callback_pub1
termId   | 0
role     | `current
state    | `running
canRoll  | 1b
startedAt| 2026.07.29D11:57:03.713654359
endedAt  | 0Np
nodes    | `callback_pub1`variable_output1!+(,`state)!,`running`running

sp.get_unit_term(unit_id, term_id)

Parameters:

name	type	description
unit_id	str	Unit ID, as returned by sp.get_units.
term_id	int	Term ID of the unit's current or previous term.

Returns:

type	description
pykx.Dictionary	Keys unitId, termId, role, state, canRoll, startedAt, endedAt, nodes.

Example:

>>> sp.get_unit_term('callback_pub1', 0)
pykx.Dictionary(pykx.q('
unitId   | `callback_pub1
termId   | 0
role     | `current
state    | `running
canRoll  | 1b
startedAt| 2026.07.29D11:57:03.713654359
endedAt  | 0Np
nodes    | `callback_pub1`variable_output1!+(,`state)!,`running`running
'))

Roll Unit

Rolls a unit from its current term to a new term. In most cases every node completes the rolled-from term immediately and the returned state is finished. A node may instead need to wait for an external event or condition to complete the term - for example a direct-write database writer waiting for the database to ingest its write session - in which case the state is accepted and the rolled-from term remains finishing until that work completes. rejected or failed mean the unit was not rolled - message gives the reason.

qPython

.qsp.rollUnit[unitId; termId]
.qsp.rollUnit[unitId; termId; force]

Parameters:

name	q type	description
unitId	symbol	The unit to roll.
termId	long	The term to roll from; must be the unit's current running term.
force	boolean	Optional, default 0b. Roll past a previous term stuck finishing, abandoning it and discarding its outstanding work. All other rejections still apply.

Returns:

type	description
table	Single-row table: unitId, fromTermId, toTermId, state, message.

Examples:

Example 1: every node completes the term synchronously.

q).qsp.rollUnit[`callback_pub1; 0]
unitId        fromTermId toTermId state    message
----------------------------------------------------------
callback_pub1 0          1        finished "Roll finished"

Example 2: a unit with a direct-write database writer drains asynchronously - the rolled-from term remains finishing until the database has ingested the write session.

q).qsp.rollUnit[`ingest; 3]
unitId fromTermId toTermId state    message
---------------------------------------------
ingest 3          4        accepted "Roll accepted"

sp.roll_unit(unit_id, term_id)
sp.roll_unit(unit_id, term_id, force)

Parameters:

name	type	description	default
unit_id	str	The unit to roll.	Required
term_id	int	The term to roll from; must be the unit's current running term.	Required
force	bool	Roll past a previous term stuck finishing, abandoning it and discarding its outstanding work. All other rejections still apply.	False

Returns:

type	description
pykx.Table	Single-row table: unitId, fromTermId, toTermId, state, message.

Examples:

Example 1: every node completes the term synchronously.

>>> sp.roll_unit('callback_pub1', 0)
pykx.Table(pykx.q('
unitId        fromTermId toTermId state    message
----------------------------------------------------------
callback_pub1 0          1        finished "Roll finished"
'))

Example 2: a unit with a direct-write database writer drains asynchronously - the rolled-from term remains finishing until the database has ingested the write session.

>>> sp.roll_unit('ingest', 3)
pykx.Table(pykx.q('
unitId fromTermId toTermId state    message
---------------------------------------------
ingest 3          4        accepted "Roll accepted"
'))

State Management

Functions for persisting in-memory q objects as part of pipeline state, so they survive checkpoints and restarts.

Track

Maintains state for a list of in-memory q objects.

qPython

.qsp.track[objects]

Parameters:

name	type	description	default
objects	symbol or symbol[]	The object name(s) to be tracked as state.	Required

Usage with Checkpoints

When tracking is enabled, at each checkpoint the Stream Processor will persist a copy of the tracked objects. On recovery these objects will be re-initialised before the pipeline begins again at the value corresponding to the last checkpoint.

Examples:

Example 1: Track a single variable.

.qsp.track[`iters]

iters:0

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[{iters+:1;x}]
    .qsp.write.toConsole[]

Example 2: Track multiple variables.

.qsp.track[`iters`last]

iters:0
last:()

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[{iters+:1;last::x;x}]
    .qsp.write.toConsole[]

Example 3: Track a namespace.

.qsp.track[`.info]  Track anything in the .info namespace

.info.iters:0
.info.last:()

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[{.info.iters+:1;.info.last::x;x}]
    .qsp.write.toConsole[]

sp.track(objects)

This persists their state across application restarts.

Parameters:

name	type	description	default
symbols	Union[List[Union[str, kx.SymbolAtom]], str, kx.SymbolAtom]	List of strings denoting the q object(s) to track in the q memory space, e.g. a single variable name, list of variables or a q namespace.	Required

Usage with Checkpoints

When tracking is enabled, at each checkpoint the Stream Processor will persist a copy of the tracked objects. On recovery these objects will be re-initialised before the pipeline begins again at the value corresponding to the last checkpoint.

Examples:

Example 1: Track a single variable.

>>> from kxi import sp
>>> import pykx as kx

>>> kx.q['iters'] = 0

>>> def transform(data):
    kx.q['iters'] += 1
    return data

>>> sp.track('iters')

>>> sp.run(sp.read.from_callback('publish')
        | sp.map(transform)
        | sp.write.to_console())

Example 2: Track multiple variables.

>>> from kxi import sp
>>> import pykx as kx

>>> kx.q['iters'] = 0
>>> kx.q['last'] = kx.q('()')

>>> def transform(data):
    kx.q['iters'] += 1
    kx.q['last'] = data
    return data

>>> sp.track(['iters', 'last'])

>>> sp.run(sp.read.from_callback('publish')
        | sp.map(transform)
        | sp.write.to_console())

Example 3: Track a namespace.

>>> from kxi import sp
>>> import pykx as kx

>>> kx.q['.info.iters'] = 0
>>> kx.q['.info.last'] = kx.q('()')

>>> def transform(data):
    kx.q['.info.iters'] += 1
    kx.q['.info.last'] = data
    return data

>>> sp.track('.info')  # Track anything in the .info namespace

>>> sp.run(sp.read.from_callback('publish')
        | sp.map(transform)
        | sp.write.to_console())

Configuration & Schema

Functions for reading mounted configuration, database schemas, and cached snapshot data made available by Insights Enterprise deployments.

Config Path

Gets the location of user mounted configurations.

qPython

.qsp.configPath[]
.qsp.configPath[object]

Parameters:

name	type	description	default
object	string	The configuration object to get the path of.	The KXI_SP_CONFIG_PATH environment variable

sp.config_path()
sp.config_path(obj)

Parameters:

name	type	description	default
obj	str	A string name of the configuration object to get the path of.	The KXI_SP_CONFIG_PATH environment variable

Returns:

The location of user mounted configurations.

Usage Guidelines

This is mostly useful in Insights Enterprise deployments to access the path of ConfigMaps and Secrets. Passing the name of the ConfigMap or Secret to this function will return the path where it was mounted.

Changing the mount path

The mount path can be changed by setting KXI_SP_CONFIG_PATH to a directory on the target deployment.

Get Schema

Loads a schema from a mounted database configuration in Insights Enterprise. Use Config Path to return the location of user mounted configurations in Insights Enterprise.

When more than one database config is mounted, specify which to read the table from. If no config is given, the first mounted config alphabetically is used.

qPython

A bare table name reads from the first mounted config. To select a specific config, pass it as a second argument.

.qsp.getSchema[name]
.qsp.getSchema[name;config]

Parameters:

name	type	description	default
name	symbol, string	The table name to read from the config.	Required
config	symbol, string	The config to read from. When omitted, the first mounted config alphabetically is used.	Optional

Returns:

.qsp.schema

Examples:

Get the trade schema from the first mounted config:

.qsp.getSchema[`trade]

name        datatype tokenize primary
-------------------------------------
sym         -11       0        0
timestamp   -12       0        0
price       -9        0        0
size        -7        0        0

Get the trade schema from the taq config when multiple config are mounted:

.qsp.getSchema[`trade;`taq]

sp.get_schema(name, config=None)

Parameters:

name	type	description	default
name	Union[str, bytes, kx.CharVector]	The table name to read from the database config.	Required
config	Union[str, bytes, kx.CharVector]	The config to read from. When omitted, the first mounted config alphabetically is used.	None

Returns:

sp.schema

Examples:

Get the trade schema from the first mounted config:

>>> sp.get_schema('trade')

name        datatype tokenize primary
-------------------------------------
sym         -11       0        0
timestamp   -12       0        0
price       -9        0        0
size        -7        0        0

Get Snapshot Cache

Reads the snapshot cache for pipeline operators.

Currently supported for Subscriber writer nodes.

qPython

.qsp.getSnapshotCache[opIDs; args]

Parameters:

name	q type	description
opIDs	symbol[] or null	List of operators (or operator IDs) to retrieve snapshots from. If null, all snapshots are returned.
args	dict	Dictionary of arguments to pass to the function.

Args:

name	q type	description
table	symbol	The table to retrieve snapshots from.

Returns data from the snapshot cache for subscribers.

Examples:

Read the snapshot cache for the 'data' table for the 'subscriber' operator

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.write.toSubscriber[`data; `sym`exch];

publish 2!enlist`sym`exch`time`pos!(`FD;`LSE;2023.06.25D15:00:00.000000000;100);
.qsp.getSnapshotCache[`subscriber_data;enlist[`table]!enlist[`data]]

sym exch| time                          pos
--------| ---------------------------------
FD  LSE | 2023.06.25D15:00:00.000000000 100

Read the snapshot cache for all operators

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.write.toSubscriber[`data; `sym`exch];

publish 2!enlist`sym`exch`time`pos!(`FD;`LSE;2023.06.25D15:00:00.000000000;100);
.qsp.getSnapshotCache[`;enlist[`table]!enlist[`data]]

()
(`s#+`sym`exch!(`p#,`FD;,`LSE))!+`time`pos!(,2023.06.25D15:00:00.000000000;,100)

get_snapshot_cache(operatorIDs, args)

name	type	description
operatorIDs	string[] or None	List of operators (or operator IDs) to retrieve snapshots from. If None, all snapshots are returned.
args	dict	Dictionary of arguments to pass to the function. Current supported argument is table

Examples:

Read the snapshot cache for the 'data' table

>>> from kxi import sp
>>> import pandas as pd
>>> import pykx as kx

>>> kx.q('upd:{show z}')
>>> sp.run(sp.read.from_callback('publish')
        | sp.write.to_subscriber('data', 'sym'))

>>> updData = pd.DataFrame({
        'sym': ['FDP', 'AAPL', 'MSFT'],
        'id': [1, 2, 3],
        'size': [100, 150, 200]
})

>>> kx.q('publish', updData)
>>> sp.get_snapshot_cache(None, {'table': 'data'})

Environment Variables

A function for resolving pipeline argument values from environment variables at runtime, rather than hardcoding them, most commonly to avoid embedding credentials in pipeline code.

Use Environment Variable

There are cases when arguments to an operator need to be extracted from an environment variable. This function allows users to input the name of the environment variable and the operator will receive the value of that environment variable converted to the correct type. One use of this can be to help hide credentials by only having references to them in the code, e.g. in a kubernetes deployment Environment Variables can be set using secrets and will not be displayed if a user describes the pod.

Resolving environment variable values

This function does not directly resolve environment variables to their values. Instead it uses a format that the pipeline will resolve internally and cast to the correct type. To get an environment variable value directly, you should use getenv in q or pykx.q.getenv in Python.

qPython

.qsp.useVar[envVar]

Parameters:

name	type	description
envVar	string or symbol	Name of the environment variable to be resolved.

Returns:

Type	Description
object	Environment variable object

Examples:

Example 1: This shows how we can hide credentials. The worker has environment variables $KAFKA_USER and$KAFKA_PASS` set. So we can successfully remove the credentials from the code.

.qsp.run
    .qsp.read.fromKafka[`spx; "kafka.insights-data.kx.com:443";
        .qsp.use enlist[`options]!enlist
            `sasl.username`sasl.password`sasl.mechanism`security.protocol!(.qsp.useVar["KAFKA_USER"]; .qsp.useVar["KAFKA_PASS"]; "SCRAM-SHA-512"; "SASL_SSL")]
    .qsp.write.toConsole[]

Example 2: Now additionally we have environment variable $WINDOW_SIZE set which we will convert to the correct type. .qsp.window.count expects a size argument of type long.The pipeline uses .qsp.useVar to extract the value of the $WINDOW_SIZE environment variable as a string and under the covers converts it to the correct type (long).

.qsp.run
    .qsp.read.fromKafka[`spx; "kafka.insights-data.kx.com:443";
        .qsp.use enlist[`options]!enlist
            `sasl.username`sasl.password`sasl.mechanism`security.protocol!(.qsp.useVar["KAFKA_USER"]; .qsp.useVar["KAFKA_PASS"]; "SCRAM-SHA-512"; "SASL_SSL")]
    .qsp.window.count[.qsp.useVar["WINDOW_SIZE"]]
    .qsp.write.toConsole[]

sp.use_var(env_var)

Parameters:

name	type	description	default
env_var	str	Name of the environment variable to be resolved.	Required

Returns:

Type	Description
object	Environment variable object

Examples:

Example 1: This shows how we can hide credentials. The worker has environment variables $KAFKA_USER and $KAFKA_PASS set. So we can successfully remove the credentials from the code.

>>> from kxi import sp
>>> import pykx as kx

>>> sp.run(sp.read.from_kafka('spx', 'kafka.insights-data.kx.com:443', options={
        'sasl.username': sp.use_var('KAFKA_USER'),
        'sasl.password': sp.use_var('KAFKA_PASS'),
        'sasl.mechanism': 'SCRAM-SHA-512',
        'security.protocol': 'SASL_SSL'})
    | sp.write.to_variable('out'))

Example 2: Now additionally we have environment variable $WINDOW_SIZE set which we will convert to the correct type. .qsp.window.count expects a size argument of type long.The pipeline uses .qsp.useVar to extract the value of the $WINDOW_SIZE environment variable as a string and under the covers converts it to the correct type (long).

>>> from kxi import sp
>>> import pykx as kx

>>> sp.run(sp.read.from_kafka('spx', 'kafka.insights-data.kx.com:443', options={
        'sasl.username': sp.use_var('KAFKA_USER'),
        'sasl.password': sp.use_var('KAFKA_PASS'),
        'sasl.mechanism': 'SCRAM-SHA-512',
        'security.protocol': 'SASL_SSL'})
    | sp.window.count(sp.use_var('WINDOW_SIZE'))
    | sp.write.to_variable('out'))

Timers

Timers let a pipeline schedule work to run periodically or once, independent of incoming data. Use these when a pipeline needs to poll a source, flush state, or otherwise perform an action on a schedule rather than in response to an event.

Add One Shot Timer

Adds or updates a once-off timer event.

qPython

.qsp.timer.add1shot[id; expression; offset]

Parameters:

name	type	description
id	symbol	Specifies the timer ID. If this ID exists, it is replaced.
expression	list	Specifies the expression to execute (typically a function name followed by its parameters).
offset	int, timespan	Specifies the offset from the current time to run the timer event in either an integer number of milliseconds or a timespan.

sp.timer_add1shot(id, expression, offset)

Parameters:

name	type	description
id	Union[str, kx.SymbolAtom]	Specifies the timer ID. If this ID exists, it is replaced.
expression	List[Any]	Specifies the expression to execute (typically a function name followed by its parameters).
offset	Union[int, datetime.timedelta]	Specifies the offset from the current time to run the timer event in either an integer number of milliseconds or a timedelta.

Add Timer

Adds or updates a frequent timer event.

qPython

.qsp.timer.add[id; expression; period; offset]

Parameters:

name	type	description
id	symbol	Specifies the timer ID. If this ID exists, it is replaced.
expression	list	Specifies the expression to execute. This is the name of a function in q followed by its parameters (None if no parameters exist)
period	int, int[2], timespan, timespan[2]	Specifies the timer period, as either a value in milliseconds or a timespan. If this value is a 2-element vector, an exponential backoff is applied to repeated invocations up to the maximum period specified by the second element
offset	int, timespan	Specifies the offset from the current time to run the timer event in either an integer number of milliseconds or a timespan.

Examples:

Call function every 20 seconds

counter:0
timerTick:{[] counter+1}
.qsp.timer.add[`timerTick; (`timerTick; ::); 20000; 0]

Call function every 1 second with exponential backoff up to 20 starting in 1 minutes time

counter:0
timerTick:{[] counter+:1 }
.qsp.timer.add[`timerTick; (`timerTick; ::); 0D00:00:01 0D00:00:20; 0D00:01:00]

sp.timer_add(id, expression, period, offset)

Parameters:

name	type	description
id	Union[str, kx.SymbolAtom]	Specifies the timer ID. If this ID exists, it is replaced
expression	List[Any]	Specifies the expression to execute. This is the name of a function in q followed by its parameters (None if no parameters exist)
period	Union[int, int[2], timedelta, timedelta[2]]	Specifies the timer period, as either a value in milliseconds or a timedelta. If this value is a 2-element vector, an exponential backoff is applied to repeated invocations up to the maximum period specified by the second element
offset	Union[int, Any]	Specifies the offset from the current time to run the timer event in either an integer number of milliseconds or a timedelta.

Examples:

Call function every 20 seconds

import pykx as kx
from kxi import sp

kx.q('counter: 0')
kx.q('timerTick:{[] counter+:1 }')

sp.timer_add('timerTick', ['timerTick', None], 20000, 0)

Call function every 1 second with exponential backoff up to 20 starting in 1 minutes time

import pykx as kx
from kxi import sp

kx.q('counter: 0')
kx.q('timerTick:{[] counter+:1 }')

sp.timer_add('timerTick', ['timerTick', None], [timedelta(seconds=1), timedelta(seconds=20)], timedelta(minutes=1))

Delete Timer

Deletes one or more timer events.

qPython

.qsp.timer.del[ids]

Parameters:

name	type	description
ids	symbol or symbol[]	Specifies the timer IDs to delete.

Examples:

Delete a single timer

.qsp.timer.del[`timerTick]

Delete multiple timers

.qsp.timer.del[`timerTick1`timerTick2]

sp.timer_del(ids)

Parameters:

name	type	description
ids	Union[str, str[], kx.SymbolAtom, kx.SymbolAtom[]]	Specifies the timer IDs to delete

Examples:

Delete a single timer

sp.timer_del('timerTick')

Delete multiple timers

sp.timer_del(['timerTick1', 'timerTick2'])

Get Timer

Retrieves the properties of one or more timer events.

qPython

.qsp.timer.get[ids]

Parameters:

name	type	description
ids	symbol or symbol[]	Specifies the timer IDs to retrieve. If this value is a generic null (::), all timer events are returned.

Returns:

type	description
table	Timer entries

Examples:

Get a single timer

.qsp.timer.get[`timerTick]

Get all timers

.qsp.timer.get[::]

sp.timer_get(ids)

Parameters:

name	type	description
ids	Union[str, str[], kx.SymbolAtom, kx.SymbolAtom[], None]	Specifies the timer IDs to retrieve. If this value is None, all timer events are returned.

Returns:

type	description
table	Timer entries

Examples:

Get a single timer

sp.timer_get('timerTick')

Get all timers

sp.timer_get(None)

Tracing & Debugging

These functions help debug pipeline behavior in two complementary ways. Both carry a performance overhead and are best avoided in production deployments.

Trace logging (Set Trace, Clear Trace) controls the verbosity of logging for events as they flow through the operators in a pipeline. When enabled, these events are printed to the pipeline logs, where they can be examined to track latency or understand data shape.

Data tracing (Enable Data Tracing, Disable Data Tracing, Get Data Trace, Reset Data Trace) caches the data at each operator node as it flows through the pipeline. The cached data can then be queried to investigate and debug problems.

Set Trace

Enables or disables trace logging. Sets the level of verbosity of trace logging.

qPython

.qsp.setTrace[level]

Parameters:

name	type	description	default
level	long	Level of trace logging to display.	Required

Examples:

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[{ select max price from x }]
    .qsp.write.toVariable[`output]

.qsp.setTrace[3];
publish ([] date: .z.d; sym: 10?3?`3; price:10?100f);

Log output:

{"time":"2026-07-29T10:01:24.785z","component":"SP","level":"TRACE","message":"[] Pushing data - id=callback_publish, offset=0N, key=, type=98, count=10, shape=[([]date:14h;sym:11h;price:9h)]"}
{"time":"2026-07-29T10:01:24.786z","component":"SP","level":"TRACE","message":"[] Received data - id=callback_publish, offset=0N, key=, type=98, count=10, shape=[([]date:14h;sym:11h;price:9h)]"}
{"time":"2026-07-29T10:01:24.786z","component":"SP","level":"TRACE","message":"[] Operator input - id=map, offset=0N, key=, type=98, count=10, shape=[([]date:14h;sym:11h;price:9h)]"}
{"time":"2026-07-29T10:01:24.786z","component":"SP","level":"TRACE","message":"[] Sending data - id=variable_output, offset=0N, key=, type=98, count=1, shape=[([]price:9h)]"}

sp.set_trace(level)

Parameters:

name	type	description	default
level	long	Level of trace logging to display.	Required

Examples:

from kxi import sp
sp.run(
    sp.read.from_callback('publish') |
    sp.map('{ select max price from x }') |
    sp.write.to_variable('output'))

sp.set_trace(3)
kx.q('{publish ([] date: .z.d; sym: 10?3?`3; price:10?100f)}', None)

Log output:

{"time":"2026-07-29T10:01:24.785z","component":"SP","level":"TRACE","message":"[] Pushing data - id=callback_publish, offset=0N, key=, type=98, count=10, shape=[([]date:14h;sym:11h;price:9h)]"}
{"time":"2026-07-29T10:01:24.786z","component":"SP","level":"TRACE","message":"[] Received data - id=callback_publish, offset=0N, key=, type=98, count=10, shape=[([]date:14h;sym:11h;price:9h)]"}
{"time":"2026-07-29T10:01:24.786z","component":"SP","level":"TRACE","message":"[] Operator input - id=map, offset=0N, key=, type=98, count=10, shape=[([]date:14h;sym:11h;price:9h)]"}
{"time":"2026-07-29T10:01:24.786z","component":"SP","level":"TRACE","message":"[] Sending data - id=variable_output, offset=0N, key=, type=98, count=1, shape=[([]price:9h)]"}

Available Levels

0: Disable trace logging (default).

1: Log data that is passed through readers and writers.

2: Log data pushed through buffers.

3: Log operator inputs.

4: Log state operations.

Clear Trace

Disables program tracing logs.

qPython

.qsp.clearTrace[]

Clears trace logging and sets the log level to its previous level.

sp.clear_trace()

Clears trace level logging and resets logging level.

Enable Data Tracing

Captures data outputs as they flow through a pipeline.

Data tracing captures data that is flowing in the streaming pipeline. This inserts probes that cache the last value emitted by each operator in the pipeline. Writer operators capture the input presented to the writer. If a given operator has an error, the error is also captured and where the input is synchronous, the data is the input to the operator.

Performance Implications

Adding data capture to a pipeline may have an impact on the pipeline performance. Data tracing should be reserved for debugging purposes and not used in production deployments where possible.

qPython

.qsp.enableDataTracing[]

Examples:

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[{ select max price from x }]
    .qsp.write.toVariable[`output]

.qsp.enableDataTracing[];
publish ([] date: .z.d; sym: 10?3?`3; price:10?100f);
.qsp.getDataTrace[]

                | error metadata data
----------------| ------------------------------------------------------------------..
callback_publish| ""    (,`)!,:: +`date`sym`price!(2022.06.06 2022.06.06 2022.06.06 ..
map             | ""    (,`)!,:: +(,`price)!,,95.97684
variable_output | ""    (,`)!,:: +(,`price)!,,95.97684

sp.enable_data_tracing()

Examples:

>>> from kxi import sp
>>> import pykx as kx

>>> sp.run(sp.read.from_callback('publish')
        | sp.map(lambda x: kx.q('{select max price from x}', x))
        | sp.write.to_variable('output'))

>>> sp.enable_data_tracing()
>>> kx.q('publish', kx.q('([] date: .z.d; sym: 10?3?`3; price:10?100f)'))
>>> sp.get_data_trace()

                | error metadata data
----------------| ------------------------------------------------------------------..
callback_publish| ""    (,`)!,:: +`date`sym`price!(2022.06.06 2022.06.06 2022.06.06 ..
map             | ""    (,`)!,:: +(,`price)!,,95.97684
variable_output | ""    (,`)!,:: +(,`price)!,,95.97684

Disable Data Tracing

Disables data tracing in the current pipeline.

qPython

.qsp.disableDataTracing[]

sp.disable_data_tracing()

Note

Disables data tracing from the current pipeline. This does not clear any captured trace data. Data captured during tracing can still be accessed via .qsp.getDataTrace.

See Enable Data Tracing for more details

Get Data Trace

Returns the data from a data trace at the time of invocation.

qPython

.qsp.getDataTrace[]

When data tracing is enabled, getDataTrace returns a point in time snapshot of the last data values emitted by each node in the pipeline.

sp.get_data_trace()

When data tracing is enabled, get_data_trace returns a point in time snapshot of the last data values emitted by each node in the pipeline.

Returns:

The returned object is a dictionary of operator IDs to their respective data or errors. If a node has an error message, any data that is captured is the last input to that operator that caused the error.

See Enable Data Tracing for more details

Reset Data Trace

Resets the current data cache state and clears any data that has been captured during a data tracing session.

qPython

.qsp.resetDataTrace[]

sp.reset_data_trace()

See Enable Data Tracing for more details

Profiling

Profiling measures how long a pipeline spends in each operator, which is useful for finding bottlenecks. Like data tracing, it carries a performance cost and should be used for debugging rather than left enabled in production.

It inserts probes around writers and operators that measure the number of times each is called, the number of records processed and the cumulative time spent in each. Readers are not currently profiled, but may be included in a future release.

Performance Implications

Adding profiling to a pipeline may have an impact on the pipeline performance. Profiling should be reserved for debugging purposes and not used in production deployments where possible.

qPython

.qsp.enableProfiling[]

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[(::)]
    .qsp.write.toVariable[`output]

publish ([] til 10)

.qsp.getProfilingTimes[]

id       avgEventTime         avgRecordTime        maxEventTime
-----------------------------------------------------------------------
map      0D00:00:00.000001973 0D00:00:00.000000263 0D00:00:00.000009412
variable 0D00:00:00.000024878 0D00:00:00.000003317 0D00:00:00.000041237

from kxi import sp
import pykx as kx

sp.enable_profiling()

sp.run(sp.read.from_callback('publish')
    | sp.map(lambda x: x)
    | sp.write.to_variable('output'))

kx.q('publish ([] til 10)')

sp.get_profiling_times()

id       avgEventTime         avgRecordTime        maxEventTime
-----------------------------------------------------------------------
map      0D00:00:00.000001973 0D00:00:00.000000263 0D00:00:00.000009412
variable 0D00:00:00.000024878 0D00:00:00.000003317 0D00:00:00.000041237

Enable Profiling

Enables profiling of operator latencies as they flow through a pipeline.

Performance Implications

Adding profiling to a pipeline may have an impact on the pipeline performance. Profiling should be reserved for debugging purposes and not used in production deployments where possible.

qPython

.qsp.enableProfiling[]

sp.enable_profiling()

Disable Profiling

Disables profiling of operator latencies in the current pipeline.

qPython

.qsp.disableProfiling[]

sp.disable_profiling()

Note

Disables profiling from the current pipeline. This does not clear any captured profiling data. Data captured while profiling was enabled can still be accessed via .qsp.getProfilingTimes / sp.get_profiling_times.

Get Profiling Times

Gets the operator latencies for a pipeline.

qPython

.qsp.getProfilingTimes[]

Returns:

A table of operator IDs to their respective latencies.

When profiling is enabled, .qsp.getProfilingTimes returns, for each operator, avgEventTime (the average time spent per call to the operator), avgRecordTime (the average time spent per record processed by the operator) and maxEventTime (the longest time spent in a single call to the operator), computed over all calls since profiling was enabled.

sp.get_profiling_times()

Returns:

A table of operator IDs to their respective latencies.

When profiling is enabled, sp.get_profiling_times returns, for each operator, avgEventTime (the average time spent per call to the operator), avgRecordTime (the average time spent per record processed by the operator) and maxEventTime (the longest time spent in a single call to the operator), computed over all calls since profiling was enabled.

See Profiling for example usage.

Record Counting

Record counting tracks the volume of data flowing through pipeline operators, which is useful for monitoring throughput and dataflow without the overhead of full data or profiling tracing.

The count stores the sum of the counts of each record for each operator, divided by the records' keys. The operators chosen to perform record counting depends on the record counting level, described in Set Record Counting.

qPython

Examples:

Example 1: Counting records with the default level.

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[{ select max price from x }]
    .qsp.write.toVariable[`output]

publish ([] date: .z.d; sym: 10?3?`3; price:10?100f);
.qsp.getRecordCounts[]

                |
---             | --
callback_publish| 10
variable_output | 1

Example 2: Setting record counting to level 2.

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.map[{ select max price from x }]
    .qsp.write.toVariable[`output]

.qsp.setRecordCounting 2;

publish ([] date: .z.d; sym: 10?3?`3; price:10?100f);
.qsp.getRecordCounts[]

                |
---             | --
map             | 10
callback_publish| 10
variable_output | 1

Examples:

Example 1: Counting records with the default level.

from kxi import sp
import pykx as kx

sp.run(sp.read.from_callback('publish')
    | sp.map(lambda x: x)
    | sp.write.to_variable('output'))

kx.q('publish ([] date: .z.d; sym: 10?3?`3; price:10?100f)')
sp.get_record_counts()

                |
---             | --
callback_publish| 10
variable_output | 1

Example 2: Setting record counting to level 2.

from kxi import sp
import pykx as kx

sp.run(sp.read.from_callback('publish')
    | sp.map(lambda x: x)
    | sp.write.to_variable('output'))

sp.set_record_counting(2)

kx.q('publish ([] date: .z.d; sym: 10?3?`3; price:10?100f)')
sp.get_record_counts()

                |
---             | --
map             | 10
callback_publish| 10
variable_output | 1

Set Record Counting

Sets the record counting level for tracking dataflow in a pipeline. Setting the level of record counting determines which nodes in the pipeline to count data flow for.

Available Levels

1. Record counting is disabled for all operators.

2. Count records flowing through readers and writers.

3. Count records flowing through all operators.

qPython

.qsp.setRecordCounting[level]

Parameters:

name	type	description	default
level	long	Level of record counting.	Required

Examples:

.qsp.setRecordCounting[2]

sp.set_record_counting(level)

Parameters:

name	type	description	default
level	long	Level of record counting.	Required

Examples:

sp.set_record_counting(2)

Changing levels

Changing levels resets the RecordCounts cache.

Get Record Counts

Gets dataflow information for a pipeline.

qPython

.qsp.getRecordCounts[]

Returns:

A dictionary of operator IDs.

When record counting is enabled, .qsp.getRecordCounts returns information on the total amount of data that has flowed through the pipeline since enabled or since the cache was last reset.

sp.get_record_counts()

Returns:

A dictionary of operator IDs.

When record counting is enabled, sp.get_record_counts returns information on the total amount of data that has flowed through the pipeline since enabled or since the cache was last reset.

The returned object is a dictionary of operator IDs to their respective counts, where counts are partitioned into stream keys. The operators tracked depends on the record counting level.

See Record Counting for full examples.

Reset Record Counts

Resets the current record counts cache, so subsequent data counts begin from zero.

qPython

.qsp.resetRecordCounts[]

sp.reset_record_counts()

Partitions

Functions for inspecting how data partitions are distributed across workers in a pipeline.

Get Partition Count

Gets count of all partitions in the pipeline (across all workers).

qPython

.qsp.getPartitionCount[]

Returns:

Type	Description
int

Examples:

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.write.toConsole[]

input: til 10
publish input
.qsp.getPartitionCount[]

0

sp.get_partition_count()

Returns:

Type	Description
kx.LongAtom

Examples:

from kxi import sp
import pykx as kx

sp.run(sp.read.from_callback('publish')
    | sp.write.to_console())

kx.q('publish til 10')
sp.get_partition_count()

0

Get Partitions

Gets partitions assigned to a worker.

qPython

.qsp.getPartitions[]

Returns:

Type	Description
list

Examples:

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.write.toConsole[]

input: til 10
publish input

.qsp.getPartitions[]

::

sp.get_partitions()

Returns:

Type	Description
kx.List[kx.SymbolAtom]

Examples:

from kxi import sp
import pykx as kx

sp.run(sp.read.from_callback('publish')
    | sp.write.to_console())

kx.q('publish til 10')
sp.get_partitions()

::