Transform

This page provides the pre-defined data transformation methods available. For custom transformations, a Map Node can be used.

Fill

(Beta Feature) Fills null values in table columns.

Beta Features

To enable beta features, set the environment variable KXI_SP_BETA_FEATURES to true. See Beta Feature Usage Terms.

qPython

.qsp.transform.fill[defaults]
.qsp.transform.fill[defaults;mode]

Parameters:

name	type	description	default
defaults	dict	Keys specify the columns to replace null values for. The value associated with a key is what's used as the default fill for the corresponding column.	Required
mode	symbol	Specifies the approach to use when filling. One of static, down, or up.	static

For all common arguments, refer to configuring operators

This operator fills null values in a table either statically, up, or down, and utilizes specified defaults.

• static - every null value in a column is replaced with its corresponding default.
• up - up (backward) fill, where if the last entry in the current column is null it is replaced with its corresponding default.
• down - down (forward) fill, where for the first table passed in, any column whose first entry is null has it replaced with the column's corresponding default. Plugin state is managed to have down-fill carry over across batches. Currently, only filling vector columns is supported, and all default values to fill with must be atomic. The type of elements in a vector column should match the type of the atomic fill value, otherwise the column is implicitly cast to match the type of the fill value.

Using fill-static on a table with null entries:

input: ([] val1: 0N 1 2 0N 3; val2: "a b c"; val3:  0N 5 0N 5 0N)

.qsp.run
   .qsp.read.fromCallback[`publish]
   .qsp.transform.fill[`val1`val2`val3!(-1;"_"; -10)]
   .qsp.write.toVariable[`output];

publish input
output

val1 val2 val3
--------------
-1   a    -10
1    _    5
2    b    -10
-1   _    5
3    c    -10

Using fill-down on a table with null entries:

input: ([] val1: 0N 1 2 0N 3; val2: "a b c"; val3:  0N 5 0N 5 0N)
dict: `val1`val2`val3!(-1;"_"; -10)

.qsp.run
   .qsp.read.fromCallback[`publish]
   .qsp.transform.fill[dict;`down]
   .qsp.write.toVariable[`output];

publish input
output

val1 val2 val3
--------------
-1   a    -10
1    a    5
2    b    5
2    b    5
3    c    5

sp.transform.fill({'price': 0.0})

Parameters:

name	type	description	default
defaults	dict	Dictionary of keys and values to replace nulls. Keys specify the columns to replace null values for. The value associated with a key is what's used as the default fill for the corresponding column.	Required
mode	symbol	Alternatively you may specify a mode for replacement of nulls. Specifies the approach to use when filling. One of static, down, or up. In static mode every null value in a column is replaced with its corresponding default. In up mode we do an up (backward) fill, where if the last entry in the current column is null it is replaced with its corresponding default. Finally, in down mode we do a down (forward) fill, where for the first table passed in, any column whose first entry is null has it replaced with the column's corresponding default	static

Returns: A fill transformer, which can be joined to other operators or pipelines.

Examples:

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

>>> sp.run(sp.read.from_callback('publish')
        | sp.transform.fill({'x': 1})
        | sp.write.to_variable('out'))
>>> kx.q("publish", pd.DataFrame({'x': [None, 2, None]}))
>>> kx.q('out')

x
-
1
2
1

Rename Columns

Renames columns in the incoming data

qPython

.qsp.transform.renameColumns[nameScheme]

Parameters:

name	type	description	default
nameScheme	dictionary	A dictionary mapping current column names to what they should be renamed to	Required

For all common arguments, refer to configuring operators

This example transforms data with column names "a" and "b" to have columns "c" and "d"

.qsp.run
    .qsp.read.fromCallback[`publish]
    .qsp.transform.renameColumns[`a`b!`c`d]
    .qsp.write.toVariable[`output];

publish ([] a: til 5; b: til 5);
output

c d
---
0 0
1 1
2 2
3 3
4 4

sp.transform.rename_columns({'price': 'x', 'quantity': 'y'})

Parameters:

name	type	description	default
nameScheme	dictionary	A dictionary defining the columns to be renamed and their new names.	Required

Returns: A rename transformer, which can be joined to other operators or pipelines.

Examples: Renames the columns in a batch of data (must be a table)

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

>>> sp.run(sp.read.from_callback('publish')
        | sp.transform.rename_columns({'a': 'x', 'b': 'y'})
        | sp.write.to_console(timestamp='none'))
>>> data = pd.DataFrame({
        'a': np.random.randn(10),
        'b': np.random.randn(10),
        'z': np.random.randn(10)
    })
>>> kx.q('publish', data)
>>> kx.q('out')

x          y          z
---------------------------------
-0.8346907 -0.6609932 -1.807997
0.9907844  -0.2233735 0.7030451
2.021636   0.8098854  0.3018809
0.2035373  1.373164   0.8300498
-0.1250729 1.424175   -0.07679308
2.156653   -0.2182297 -0.458493
-0.2572752 -2.013144  1.027509
2.010659   1.397499   -0.7732002
-1.143055  -1.739128  0.8732918
-0.4201101 0.635765   0.5426539

Replace Infinity

Replaces positive/negative infinite values with the max/min values from each columns.

qPython

.qsp.transform.replaceInfinity[X]
.qsp.transform.replaceInfinity[X; .qsp.use enlist[`newCol]!enlist newCol]

Parameters:

name	type	description
X	symbol or symbol[]	Symbol or list of symbols indicating the columns to act on.

options:

name	type	description	default
newCol	boolean	Boolean indicating if additional columns are to be added indicating which entries were infinities. 1b for yes 0b for no.	0b

For all common arguments, refer to configuring operators

This operator replaces infinite values in the specified columns. This allows the ± infinities to be replaced by the running maximum/minimum for the column.

Operating restrictions

If the first value received for a column is infinite, an error will be thrown as there is no value to replace it with.

This pipeline replaces infinities in the columns x and x1

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.replaceInfinity[`x`x1; .qsp.use ``newCol!11b]
  .qsp.write.toConsole[];

publish ([] x: 1 3 4 0w; x1: 1 -0W 0 -0W; x2: til 4);

sp.transform.replace_infinity(['price', 'quantity']

Positive infinity is replaced with the largest previously seen value in its column. Negative infinity is replaced with the smallest previously seen value in its column.

Parameters:

name	type	description
columns	symbol or symbol[]	Column(s) to apply infinity replace to in string format.

options:

name	type	description	default
new_column	boolean	Boolean indicating whether to add new column (True) which indicates the presence of infinite values for a given row.	0b

Returns: An infinity_replace transformer, which can be joined to other operators or pipelines.

Examples: Replaces the infinite values in a batch of data:

>>> from kxi import sp
>>> import pykx as kx
>>> import pandas as pd
>>> import numpy as np
>>> sp.run(sp.read.from_callback('publish')
        | sp.transform.replace_infinity(['x', 'x1'])
        | sp.write.to_console(timestamp='none'))
>>> data = pd.DataFrame({
        'x': np.random.randn(10),
        'x1': np.random.randn(10),
        'x2': np.random.randn(10)
    })
>>> data['x'][9] = np.inf
>>> kx.q('publish' , data)
>>> kx.q('out')

    x          x1         x2
---------------------------------
1.122557   1.091329   -0.9192559
-0.2272307 0.1437045  -1.314342
1.852872   -0.8079581 -0.04237484
-1.311179  -0.4326425 1.383893
-0.7943273 -1.13575   -0.437631
0.1900052  -1.223518  -0.9580604
0.2486013  -1.104549  0.8776486
1.109844   0.380195   0.898798
0.9993286  0.4093832  -1.477295
1.852872   0.7642802  0.4787892

Replace Null

Replaces null values using the median from each of the columns.

qPython

.qsp.transform.replaceNull[X]
.qsp.transform.replaceNull[X; .qsp.use (!) . flip (
    (`newCol    ; newCol);
    (`bufferSize; bufferSize))]

Parameters:

name	type	description
X	symbol or symbol[] or dictionary	Symbol or list of symbols indicating the columns to act on, or a dictionary of column names and replacement values.

options:

name	type	description	default
newCol	boolean	Boolean indicating if additional columns are to be added indicating which entries were null. 1b for yes 0b for no.	0b
bufferSize	long	Number of data points that must amass before calculating the median.	0

For all common arguments, refer to configuring operators

This operator replace null values in the selected columns. If X is a symbol or list of symbols, nulls will be replaced by the median for the column, as calculated from the buffered data, or from the first batch if bufferSize is not specified. Initially, the batches are collected in a buffer until the required size is exceeded. If this buffer contains only null values for the column, an error will be logged, and null values will not be replaced. If X is a dictionary of column names and replacement values, nulls will be replaced by the value given for the column. When replacement values are given, no buffering is done.

This pipeline replaces nulls columns x and y with the median values.

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.replaceNull[`x`y;.qsp.use`bufferSize`newCol!(10;1b)]
  .qsp.write.toConsole[];

publish ([] x: 0n,10?1f; y: 11?1f; z: til 11);

This pipeline replaces nulls in columns x and x1 with specific values.

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.replaceNull[`x`x1!0 .5;.qsp.use ``newCol!11b]
  .qsp.write.toConsole[];

publish ([] x: 0n , 10?1f; x1: 0n 0n , (8?1f) , 0n);

sp.transform.replace_null(['price', 'quantity']

Parameters:

name	type	description
columns	symbol or symbol[] or dictionary	Column(s) to apply null replace to in string format, or None replace nulls in all float64 columns within a batch.

options:

name	type	description	default
buffer_size	boolean	Number of data points which must amass before a determination is made of the median value for specified columns and after which null replacement is applied.	0b
new_column	long	Boolean indicating whether a new column is to be added to indicate the presence of null values for a given row.	0

Returns: A null_replace transformer, which can be joined to other operators or pipelines.

Examples: Replaces the null values in a batch of data:

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

>>> sp.run(sp.read.from_callback('publish')
        | sp.transform.replace_null(['x', 'x1'])
        | sp.write.to_console(timestamp='none'))
>>> data = pd.DataFrame({
        'x': np.random.randn(10),
        'x1': np.random.randn(10),
        'x2': np.random.randn(10)
    })
>>> data['x'][9] = np.nan
>>> kx.q('publish', data)
>>> kx.q('out')

    x          x1         x2
--------------------------------
0.9570144  0.3184851  1.184641
-2.361744  -0.2467937 -0.84709
0.483128   0.1582659  0.3794346
-0.9834509 0.3171456  0.5043843
2.389175   0.118148   -0.5278073
0.9906622  -1.987247  -0.831912
0.7275034  0.05414338 0.9702141
0.2681021  1.785524   1.305275
0.4028538  1.23407    0.2663236
0.483128   -1.194168  1.019275

Schema

Applies a table schema to data passing through the operator.

Incoming columns that are not in the target schema are dropped, columns in the target schema that are missing from the input are added with nulls, and remaining columns are cast (and optionally parsed from strings) to the target datatypes. If the schema cannot be applied — for example, due to incompatible types, an array event whose length does not match the schema, or no common columns at all — an exception will be thrown.

qPython

.qsp.transform.schema[schema]
.qsp.transform.schema[schema; .qsp.use (!) . flip (
    (`inputType  ; inputType);
    (`schemaType ; schemaType);
    (`parse      ; parse))]

Parameters:

name	type	description	default
schema	symbol or table or dictionary	The target schema. Accepts an empty typed table, a dictionary mapping column names to type identifiers, or a symbol naming a table in a mounted Insights assembly.	Required

options:

name	type	description	default
inputType	symbol	The data type of each event, one of table, arrays or auto. When fixed for every event, the SP can perform additional optimizations at compile time.	auto
schemaType	symbol	How to interpret the schema argument. literal (the default) treats schema as the desired output table. schema treats it as a metadata-style table — see below.	literal
parse	symbol	Whether to parse string input data into typed values. One of auto, on or off. Ignored when schemaType is set to schema.	auto

For all common arguments, refer to configuring operators

When inputType is set to arrays, each incoming message must be a list of column data, with one entry per column in the target schema, matched positionally in the order they appear in schema. Each entry carries that column's values for the entire batch — either as a typed list (one value per row), or as a scalar that is extended to the batch length. A single message can therefore contain many rows.

When schema is a symbol, it is resolved as the name of a table mounted into the pipeline. If the pipeline is deployed as part of a package, this will be done automatically. Otherwise it must be mounted via ConfigMaps or Secrets parameters.

The schemaType option

schemaType controls whether schema is the output table itself or a description of it.

With schemaType set to literal (the default), schema is the desired output table — column names and types come directly from it.

With schemaType set to schema, schema is instead a metadata-style table of the form:

([] name:`symbol$(); datatype:`short$())

It may optionally include a tokenize column to control parsing per column:

value	meaning
auto	Inspect each batch and tokenize the column only if needed.
on	Always tokenize this column. Faster than auto when input is always strings.
off	Never tokenize this column. Faster than auto when input is already typed.

The top-level parse option is only used when schemaType is set to literal and applies the same tokenization choice to every column. When schemaType is set to schema, parse is ignored and tokenization is controlled per column via tokenize.

The parse option

parse controls whether string input values are tokenized into typed values:

• auto — in each batch, detect columns that need parsing and parse them.
• on — always parse every column. Faster than auto when input is always strings.
• off — never parse. Faster than auto when input is already typed.

For parsing to succeed, strings must be in the expected format: bytes as two base-16 digits ("ff"), integers in base 10 ("255"), and booleans in one of the supported truthy formats.

Providing parse as a boolean is deprecated

For backwards compatibility, 1b maps to auto and 0b maps to off. New pipelines should use the symbol form.

Examples:

Apply a literal schema given as an empty typed table:

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.schema[([] sym:`$(); price:`float$())]
  .qsp.write.toConsole[]

publish ([] sym:`AAPL`GOOG; price:178.12 95.05)

Apply the same schema given as a column-name-to-type dictionary:

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.schema[`sym`price!("symbol";"float")]
  .qsp.write.toConsole[]

publish ([] sym:`AAPL`GOOG; price:178.12 95.05)

Apply the same schema with parse set to always tokenize string input. The same parse value applies to every column:

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.schema[`sym`price!("symbol";"float"); .qsp.use``parse!(::;`on)]
  .qsp.write.toConsole[]

publish ([] sym:("AAPL";"GOOG"); price:("178.12";"95.05"))

Apply a schema by table name from a mounted package:

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.schema[`trade]
  .qsp.write.toConsole[]

publish ([] sym:`AAPL`GOOG; price:178.12 95.05)

Apply a metadata-style schema with per-column tokenization. Here sym is never parsed (already a symbol) and price is always parsed (arriving as strings):

schema:([] name:`sym`price; datatype:-11 -9h; tokenize:`off`on)

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.schema[schema; .qsp.use``schemaType!(::;`schema)]
  .qsp.write.toConsole[]

publish ([] sym:`AAPL`GOOG; price:("178.12";"95.05"))

Apply a schema to array-form events. Columns are matched positionally:

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.schema[([] sym:`$(); price:`float$()); .qsp.use``inputType!(::;`arrays)]
  .qsp.write.toConsole[]

publish (`AAPL`GOOG; 178.12 95.05)

sp.transform.schema({'sym': kx.SymbolAtom, 'price': kx.FloatAtom})

Parameters:

name	type	description	default
table	Union[dict, kx.Table]	The target schema. With the default schema_type of literal, a dictionary mapping column names to PyKX type classes (e.g. kx.SymbolAtom, kx.FloatAtom), or an empty kx.Table with the desired columns and types. With schema_type set to schema, this is instead a metadata-style kx.Table — see below.	Required

options:

name	type	description	default
input_type	str	The data type of each event, one of table, arrays or auto. When fixed for every event, the SP can perform additional optimizations at compile time.	auto
schema_type	str	How to interpret the table argument. literal (the default) treats it as the desired output table. schema treats it as a metadata-style table — see below.	literal
parse	str	Whether to parse string input data into typed values. One of auto, on or off. Ignored when schema_type is set to schema.	auto

Returns: A schema transformer operator, which can be joined to other operators or pipelines.

For all common arguments, refer to configuring operators

When input_type is set to arrays, each incoming message must be a list of column data, with one entry per column in the target schema, matched positionally in the order they appear in the schema. Each entry carries that column's values for the entire batch — either as a typed list (one value per row), or as a scalar that is extended to the batch length. A single message can therefore contain many rows.

When table is passed as a kx.SymbolAtom (i.e. a q symbol naming a table), it is resolved against tables defined in a mounted Insights assembly. If the Stream Processor is deployed inside an assembly, those tables are mounted automatically; otherwise, the assembly configuration must be mounted via the kubeConfig parameter of the deployment REST API.

The schema_type option

schema_type controls whether table is the output table itself or a description of it.

With schema_type set to literal (the default), table is the desired output schema — a dictionary mapping column names to PyKX type classes, or an empty kx.Table whose columns and types describe the output.

With schema_type set to schema, table is instead a metadata-style kx.Table of the form:

kx.q('([] name:`symbol$(); datatype:`short$())')

It may optionally include a tokenize column to control parsing per column:

value	meaning
auto	Inspect each batch and tokenize the column only if needed.
on	Always tokenize this column. Faster than auto when input is always strings.
off	Never tokenize this column. Faster than auto when input is already typed.

The top-level parse option is only used when schema_type is set to literal and applies the same tokenization choice to every column. When schema_type is set to schema, parse is ignored and tokenization is controlled per column via tokenize.

The parse option

parse controls whether string input values are tokenized into typed values:

• auto — in each batch, detect columns that need parsing and parse them.
• on — always parse every column. Faster than auto when input is always strings.
• off — never parse. Faster than auto when input is already typed.

For parsing to succeed, strings must be in the expected format: bytes as two base-16 digits ("ff"), integers in base 10 ("255"), and booleans in one of the supported truthy formats.

Examples:

Apply a literal schema given as a PyKX type dictionary:

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

>>> sp.run(sp.read.from_callback('publish')
            | sp.transform.schema({'sym': kx.SymbolAtom, 'price': kx.FloatAtom})
            | sp.write.to_variable('out'))
>>> kx.q('publish', kx.q('([] sym:`AAPL`GOOG; price:178.12 95.05)'))
>>> kx.q('out')

Apply the same schema given as an empty kx.Table:

>>> schema = kx.q('([] sym:`$(); price:`float$())')
>>> sp.run(sp.read.from_callback('publish')
            | sp.transform.schema(schema)
            | sp.write.to_variable('out'))
>>> kx.q('publish', kx.q('([] sym:`AAPL`GOOG; price:178.12 95.05)'))
>>> kx.q('out')

Apply a literal schema with parse set to always tokenize string input. The same parse value applies to every column:

>>> sp.run(sp.read.from_callback('publish')
            | sp.transform.schema({'sym': kx.SymbolAtom, 'price': kx.FloatAtom}, parse='on')
            | sp.write.to_variable('out'))
>>> kx.q('publish', kx.q('([] sym:("AAPL";"GOOG"); price:("178.12";"95.05"))'))
>>> kx.q('out')

Apply a schema by table name from a mounted Insights assembly. Pass the table name as a kx.SymbolAtom:

>>> sp.run(sp.read.from_callback('publish')
            | sp.transform.schema(kx.SymbolAtom('trade'))
            | sp.write.to_variable('out'))
>>> kx.q('publish', kx.q('([] sym:`AAPL`GOOG; price:178.12 95.05)'))

Apply a metadata-style schema with per-column tokenization. Here sym is never parsed (already a symbol) and price is always parsed (arriving as strings):

>>> schema = kx.q('([] name:`sym`price; datatype:-11 -9h; tokenize:`off`on)')
>>> sp.run(sp.read.from_callback('publish')
            | sp.transform.schema(schema, schema_type='schema')
            | sp.write.to_variable('out'))
>>> kx.q('publish', kx.q('([] sym:`AAPL`GOOG; price:("178.12";"95.05"))'))
>>> kx.q('out')

Apply a schema to array-form events. Columns are matched positionally:

>>> schema = kx.q('([] sym:`$(); price:`float$())')
>>> sp.run(sp.read.from_callback('publish')
            | sp.transform.schema(schema, input_type='arrays')
            | sp.write.to_variable('out'))
>>> kx.q('publish', kx.q('(`AAPL`GOOG; 178.12 95.05)'))
>>> kx.q('out')

Time Split

Decompose temporal (e.g. date/time) columns into constituent parts (e.g. minutes, hours, days etc.).

qPython

.qsp.transform.timeSplit[X]
.qsp.transform.timeSplit[X; .qsp.use enlist[`delete]!enlist delete]

Parameters:

name	type	description
X	symbol or symbol[] or ::	The columns to act on. Alternatively use :: to convert all temporal columns.

options:

name	type	description	default
delete	boolean	Boolean indicating if the original temporal column is to be deleted post conversion. This is true by default as many ML algorithms cannot discern the meaning of temporal types.	1b

For all common arguments, refer to configuring operators

This operator is used to decompose temporal columns within kdb+ tables (date/month/timestamp etc.) into a set of constituent elements which can be used to represent this data. This functionality is described here, and is intended to allow the information in temporal data to be better utilized within ML algorithms by providing contextual information such as day of the week/quarter information alongside information such as month/day/hour/second etc.

The operator can be configured to convert data on a per column(s) basis or used on all temporal columns. By default the original column is deleted, this can be modified to allow the original column to be maintained if required by the use-case.

This pipeline replaces all temporal columns (date/month) with decomposed representations.

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.timeSplit[::]
  .qsp.write.toConsole[];

publish ([] month: asc "m"$10?5; date: asc "d"$10?60; number: 10?1f)

This pipeline replaces a specified column with the decomposed representation.

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.timeSplit[`x]
  .qsp.write.toConsole[];

publish ([] x: 10?2020.01.01D00:00:00; y: 10?00:00:00)

This pipeline replaces specified columns with decomposed representation, retaining the originals column.

.qsp.run
  .qsp.read.fromCallback[`publish]
  .qsp.transform.timeSplit[`x`y; .qsp.use ``delete!00b]
  .qsp.write.toConsole[];

publish ([] x: 10?00:00; y: 10?00:00; z: til 10)

sp.transform.time_split()

Parameters:

name	type	description
columns	symbol or symbol[] or ::	The names of the columns to be temporally decomposed, or None to decompose all date/time columns.

options:

name	type	description	default
delete	boolean	Whether the original temporal columns which have been decomposed should be removed from the batch.	1b

Returns: A time_split transformer, which can be joined to other operators or pipelines.

Examples: Apply time split to reduce times to seasonal components:

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

>>> sp.run(sp.read.from_callback('publish')
        | sp.transform.time_split()
        | sp.write.to_console(timestamp='none'))
>>> data = pd.DataFrame({
        'x':[datetime(2000,1,1), datetime(2000,2,2)],
        'x1':['a', 'b']
    })
>>> kx.q('publish', data)
>>> kx.q('out')

x1 x_dayOfWeek x_year x_month x_day x_quarter x_weekday x_hour x_minute x_second
--------------------------------------------------------------------------------
a  0           2000   1       1     1         0         0      0        0
b  4           2000   2       2     1         1         0      0        0