Nodes are the building blocks of your workflow that perform specific operations on your data. You can connect nodes together to create a data pipeline that processes your data from start to finish.
Nodes are WebAssembly modules and are run in Web Workers. Internally, a Web Worker may be used to run multiple nodes.
A Node Definition is a zip file with a manifest.json. The manifest defines the inputs, outputs, configuration options, and metadata about the node. It also defines the entry point into the code for the node.
In DataCogs, manifest.json files may have comments and trailing commas.
This example shows a node that assigns properties to columns in a dataset. The entrypoint is a file named node.js. When this node is instantiated, a Web Worker will start and load the node.js file.
{
"NodeType": { // The type of the package this manifest describes
"V1": { // The version of the NodeType API this manifest conforms to. Currently this should always be "V1".
"name": "Column Properties",
"description": "Assign properties to columns",
"author": "someone",
"version": "1.0.0", // The version of this node
"module": "node",
/*
The settings below demonstrate a node that is configured by the node code itself at runtime.
*/
"settings": [
{
"name": "Column Count",
"description": "Number of Columns",
"setting_type": {
"type": "IntegerInput",
"default": 0
}
},
{
"name": "Columns",
"description": "For each column, set the data type and role",
"growth_policy": {"Managed": "Column Count"},
"settings": [
{
"name": "Column",
"style": "Table",
"settings": [
{"name":"Name", "setting_type": {"type": "TextInput", "default": ""}},
{"name":"Data Type",
"setting_type": {
"type": "RadioButton",
"default": "Nominal",
// Each Radio Button option is a pair of a value and a description
"options": [
["Nominal", "Categories without any inherent order; they simply name or classify observations (e.g., colors, languages)."],
["Ordinal", "Have a meaningful order or ranking, but the exact differences between ranks are unknown or unequal (e.g., ratings like satisfied, neutral, dissatisfied)."],
["Interval", "Numerical data with equal intervals between values but no true zero point; differences are meaningful, but ratios are not (e.g., temperature in Celsius or Fahrenheit)."],
["Ratio", "Numerical data with equal intervals and a true zero point, allowing for meaningful comparison using differences and ratios (e.g., duration, height, weight)."]
]
}
},
{"name":"Role",
"setting_type": {
"type": "RadioButton",
"default": "Ignore",
"options": [
["Ignore", "Not used in analysis"],
["Identifier", "Unique values assigned to each record to distinguish one observation from another"],
["Target", "A variable you are trying to predict or classify"],
["Feature", "Variables used as input to models to predict or explain the target variable"]
]
}
}
]
}
]
}
],
"inputs": [
{
"name": "Data Stream",
"specification": {
"transport": "Buffer",
"data_format": "application/vnd.apache.arrow.stream"
},
"is_required": true
}
],
"outputs": [
{
"name": "Typed Data Stream",
"specification": {
"transport": "Buffer",
"data_format": "application/vnd.apache.arrow.stream"
},
"tags": ["Data Type", "Role"],
"linked_to_input": {
"input_index": 0
}
}
]
}
}
}
A single manifest may also define multiple nodes. Each node is defined in a separate object within the manifest. This is useful for defining multiple nodes that share common code or resources.
{
"NodeType": {
"query":{ // A unique identifier for the node. This should be stable because it is saved in the graph.
"V1": { // The version of the NodeType API this manifest conforms to.
"name": "Query Data Stream",
"description": "Perform SQL-like queries and transformations on Arrow RecordBatches using Apache Datafusion",
"author": "someone",
"version": "1.0.0",
"module": "node/recordbatch_transform",
"tags": ["data", "transform", "query"],
/*
This node can grow/shrink the number of inputs and outputs based on the settings.
If the user or the node code changes the number of inputs or outputs, the node will physically add or remove connectors.
*/
"settings": [
{
"name": "Inputs",
"description": "The number of input streams",
"setting_type": {
"type": "IntegerInput",
"default": 1,
"range": [1, 10]
}
},
{
"name": "Outputs",
"description": "The number of output streams",
"setting_type": {
"type": "IntegerInput",
"default": 1,
"range": [1, 10]
}
},
{
"name": "Output",
"growth_policy": {"Managed": "Outputs"}, // The number of outputs is managed by the Outputs setting.
"settings": [
{
"name": "Query",
"description": "The query to apply. Use 'inputN' as the name of the input stream, where N is 0 for the first connection 1, for the second, etc. Note that if the input data has upper case letters the column name must be quoted. String values in a WHERE clause must be quoted with single quotes, e.g. SELECT * from input0 WHERE \"City\" = 'London'.",
"setting_type": {
"type": "CodeEditor",
"required": true,
"language": "sql",
"disable_popout": true,
"default": "SELECT * FROM input0"
}
}
]
}
],
"inputs": [
{
"name": "input",
"specification": {
"transport": "Buffer",
"data_format": "application/vnd.apache.arrow.stream"
},
"is_required": true,
"growth_policy": {"Managed": "Inputs"}
}
],
"outputs": [
{
"name": "output",
"specification": {
"transport": "Buffer",
"data_format": "application/vnd.apache.arrow.stream"
},
"growth_policy": {"Managed": "Outputs"}
}
]
}
},
"generate_id":{
"V1": {
"name": "Generate ID Column",
"description": "Generate a new column with unique identifiers",
"author": "someone",
"version": "1.0.0",
"module": "node/recordbatch_transform",
"tags": ["data", "transform", "id"],
"settings": [
{
"name": "Column Name",
"description": "The name of the new column",
"setting_type": {
"type": "TextInput",
"default": "ID"
}
},
{
"name": "Start Value",
"description": "The starting value for the ID column",
"setting_type": {
"type": "IntegerInput",
"default": 0
}
},
{
"name": "Increment",
"description": "The increment value for the ID column",
"setting_type": {
"type": "IntegerInput",
"default": 1
}
}
],
"inputs": [
{
"name": "Data Stream",
"specification": {
"transport": "Buffer",
"data_format": "application/vnd.apache.arrow.stream"
},
"is_required": true
}
],
"outputs": [
{
"name": "Data Stream",
"specification": {
"transport": "Buffer",
"data_format": "application/vnd.apache.arrow.stream"
},
"linked_to_input": {
"input_index": 0
}
}
]
}
}
}
}
Node communication is managed through a MessagePort. The Node API is used for things such as managing the node state, sending and receiving data from the graph, and opening and communicating with other nodes. The Orchestrator manages both the lifecycle of each node, as well as the communication between nodes.
There is a Rust-based wrapper around the Node API that simplifies the node lifecycle for nodes written in Rust. It provides async mechanisms and events for monitoring inputs and configuration changes.