Data Transformations
.qsp.transform fill (Beta) fill in null values in a table renameColumns renames columns in the incoming data replaceInfinity replaces infinite values with min/max values replaceNull replaces null values with the median value schema transforms data to match a provided schema timeSplit decomposes time columns into subdivisions of mins/hours etc
.qsp.transform.fill
(Beta Feature) Fill null values in table columns
Beta Features
To enable beta features, set the environment variable KXI_SP_BETA_FEATURES
to true
.
.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
.qsp.transform.renameColumns
Renames columns in the incoming data
.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
.qsp.transform.replaceInfinity
Replaces positive/negative infinite values with the max/min values from each columns
.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);
.qsp.transform.replaceNull
Replaces null values using the median from each of the columns
.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);
.qsp.transform.schema
Apply a table schema to data passing through the operator
.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 | A schema to apply to the incoming data. This can be a table name represented in a mounted Insights assembly, an empty kdb+ table, or a map of column names to column types. | Required |
options:
name | type | description | default |
---|---|---|---|
inputType | symbol | The data type of the stream (one of arrays , table or auto ), if this is fixed for all events, this allows for the SP to make optimization at compilation time. |
auto |
schemaType | symbol | How to interpret the provided schema object. By default the schema is treated as the desired literal output, alternatively, this can be set to schema. When using schema the provided schema object should be in a special table format of ([] name: $(); datatype: short$()). |
literal |
parse | symbol | Indicates whether parsing of input string data into other datatypes is required. One of auto , on or off . If on or off , optimization of processing can be done. |
auto |
For all common arguments, refer to configuring operators
This applies a given schema to the data passing through the operator.
Input Considerations
Events containing table data
Columns in incoming messages that are not in the target schema will be dropped. Columns in the target schema
that are missing will be added with nulls. If no common columns are found, or incoming data cannot
be cast to the to the target datatype, a null event will be returned and an error will be logged
to stdout
with the corresponding message offset, if available.
Events containing array data
Incoming array data must have the same length as the number of columns in the target schema. Additionally, each element of the array can be a typed list or a scalar that can be scalar-extended to the length of the outbound table data. The schema will be applied to the incoming data in the column order specified in the schema configuration (whether it's a table or a dictionary). If the schema cannot be applied to the incoming data due to array length or invalid type conversions the operator will push a null event further down the pipeline and log an error with message offset, if available.
Expected type formats
The parse
option allows for string representations to be converted to typed values. For numeric values to
be parsed correctly, they must be provided in the expected format. String values in unexpected formats may
be processed incorrectly.
- Strings representing bytes are expected as exactly two base 16 digits, e.g.
"ff"
- Strings representing integers are expected in base 10, e.g.
"255"
- Strings representing boolean values have a number of supported options, e.g.
"t"
,"1"
- More information on the available formats.
Specifying tokenization using the schema object (schemaType=schema
)
All entries in the tokenization column must be one of auto
, on
, or off
.
auto
- In each batch of data, check if the corresponding column needs to be tokenized, and if so have this done.on
- Always tokenize the corresponding column in the incoming data, without doing any checks. Offers performance improvements overauto
.off
- Never tokenize the corresponding column in the incoming data. Offers performance improvements overauto
.
Note: the parse
option is ignored when a schema object is passed in.
Specifying tokenization using the parse
option
The parse
option is only used when a schema literal is passed in (schemaType=literal
). It defines
the same tokenization option for all input columns:
auto
- In each batch of data, detect and parse columns that require tokenization.parse
- In each batch of data, parse all columns without doing any checks. Offers performance improvements overauto
.off
- No parsing is done for input data. Offers performance improvements overauto
.
Note: Providing parse
as a boolean is deprecated. For backwards-compatibility, 1b maps to auto
and 0b maps to off
.
Providing a symbol as the input schema
If a symbol is provided as the input schema to this operator, the table name must exist in a mounted assembly in the SP Worker.
When the Stream Processor is deployed inside of an assembly, that information will be automatically mounted so any table in
the assembly in which the SP is also deployed can be referenced. If the Stream Processor is deployed outside of an assembly,
an assembly configuration map must be mounted using the kubeConfig
parameter of our deployment REST API.
Convert an input stream to the correct datatypes:
.qsp.run
.qsp.read.fromCallback[`publish]
.qsp.decode.json[]
.qsp.transform.schema[([] date: `date$(); price: `float$(); sym:`$()); .qsp.use``parse!(::;1b)]
.qsp.write.toConsole[]
publish enlist "[{\"date\":\"2020-05-22\",\"price\":\"178.12\",\"sym\":\"AAPL\"}]"
| date price sym
-----------------------------| ----------------------
2021.11.25D17:46:03.169434200| 2020.05.22 178.12 AAPL
.qsp.transform.timeSplit
Decompose temporal columns into constituent elements
.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)