Pipelines

Register, list, detail and unregister stream processing pipelines

Overview

Streams are the backbone of all processing. Joule Management API provides the ability to register, list, detail and undeploy stream pipelines.

For further information on the structure of the content payload refer to this documentation.

Example stream content

{
  "stream": {
    "name": "basic_tumbling_window_pipeline",
    "enabled": true,
    "eventTimeType": "EVENT_TIME",
    "sources": [
      "nasdaq_quotes_stream"
    ],
    "processing unit": {
      "pipeline": [
        {
          "time window": {
            "emitting type": "tumblingQuoteAnalytics",
            "aggregate functions": {
              "FIRST": [
                "ask"
              ],
              "LAST": [
                "ask"
              ]
            },
            "policy": {
              "type": "tumblingTime",
              "window size": 1000
            }
          }
        }
      ]
    },
    "emit": {
      "select": "symbol, ask_FIRST, ask_LAST"
    },
    "group by": [
      "symbol"
    ]
  }
}

Register a stream

Register a use case stream and validate it before starting up

POST/joule/management/stream/register

Body

application/json

namestring

enabled*boolean

sourcesarray of string

bufferSize*integer (int32)

checkPointProcessing*boolean

groupbyKeyGroupbyKey (object)

groupByAttributesarray of string

streamTimeTypeStreamTimeType (enum)

INGEST_TIMEEVENT_TIMEPROCESS_TIMEUSER_DEFINED

propertiesPathstring

consumerQueueSize*integer (int32)

consumingFrequency*integer (int64)

publishingQueueSize*integer (int32)

publishingFrequency*integer (int64)

telemetryAuditingSpecificationTelemetryAuditingSpecification (object)

validFromstring (date-time)

validTostring (date-time)

asyncProcessing*boolean

initialisationSpecificationInitialisationSpecification (object)

processingUnitSpecificationProcessingUnitSpecification (object)

selectProcessorSelectProcessor (object)

Response

201500501502

Successful registered stream

Request

JavaScriptCurlPython

const response = await fetch('/joule/management/stream/register', {
    method: 'POST',
    headers: {
      "Content-Type": "application/json"
    },
    body: JSON.stringify({
      "enabled": false,
      "bufferSize": 0,
      "checkPointProcessing": false,
      "consumerQueueSize": 0,
      "consumingFrequency": 0,
      "publishingQueueSize": 0,
      "publishingFrequency": 0,
      "asyncProcessing": false
    }),
});
const data = await response.json();

List all streams

List all streams registered within Joule node

GET/joule/management/stream/list

Response

201500

Successful list streams

Request

JavaScriptCurlPython

const response = await fetch('/joule/management/stream/list', {
    method: 'GET',
    headers: {},
});
const data = await response.json();

Get stream specification

Get the stream processing specification of a registered stream

GET/joule/management/stream/detail

Query parameters

Response

201404500501

Successful detail provided for stream

Body

application/json

namestring

enabled*boolean

sourcesarray of string

bufferSize*integer (int32)

checkPointProcessing*boolean

groupbyKeyGroupbyKey (object)

groupByAttributesarray of string

streamTimeTypeStreamTimeType (enum)

INGEST_TIMEEVENT_TIMEPROCESS_TIMEUSER_DEFINED

propertiesPathstring

consumerQueueSize*integer (int32)

consumingFrequency*integer (int64)

publishingQueueSize*integer (int32)

publishingFrequency*integer (int64)

telemetryAuditingSpecificationTelemetryAuditingSpecification (object)

validFromstring (date-time)

validTostring (date-time)

asyncProcessing*boolean

initialisationSpecificationInitialisationSpecification (object)

processingUnitSpecificationProcessingUnitSpecification (object)

selectProcessorSelectProcessor (object)

Request

JavaScriptCurlPython

const response = await fetch('/joule/management/stream/detail?name=text', {
    method: 'GET',
    headers: {},
});
const data = await response.json();

Response

201

{
  "name": "text",
  "enabled": false,
  "sources": [
    "text"
  ],
  "bufferSize": 0,
  "checkPointProcessing": false,
  "groupbyKey": {
    "keys": [
      "text"
    ],
    "partitionKey": 0
  },
  "groupByAttributes": [
    "text"
  ],
  "streamTimeType": "INGEST_TIME",
  "propertiesPath": "text",
  "consumerQueueSize": 0,
  "consumingFrequency": 0,
  "publishingQueueSize": 0,
  "publishingFrequency": 0,
  "telemetryAuditingSpecification": {
    "name": "text",
    "enabled": false,
    "rawTelemetrySpecification": {
      "cloneEvents": false,
      "auditFrequency": 0
    },
    "processedTelemetrySpecification": {
      "cloneEvents": false,
      "auditFrequency": 0
    }
  },
  "validFrom": "2024-12-21T16:08:36.948Z",
  "validTo": "2024-12-21T16:08:36.948Z",
  "asyncProcessing": false,
  "initialisationSpecification": {
    "tableImportProcessor": {
      "parquetImportConfigurations": [
        {
          "schema": "text",
          "files": [
            "text"
          ],
          "table": "text",
          "asView": false,
          "index": {
            "schema": "text",
            "fields": [
              "text"
            ],
            "unique": false
          }
        }
      ],
      "schema": "text",
      "name": "text",
      "enabled": false
    },
    "modelImporter": {
      "empty": false,
      "name": "text",
      "enabled": false
    },
    "name": "text",
    "enabled": false
  },
  "processingUnitSpecification": {
    "name": "text",
    "enabled": false,
    "metricsEngineSpecification": {
      "metricsDefinitionSpecification": {
        "metrics": [
          {
            "name": "text",
            "query": "text",
            "tableDefinition": "text",
            "metricKey": "text",
            "compactionSpecification": {
              "frequency": 0,
              "timeUnit": "NANOSECONDS",
              "userDefinedCompactionQuery": "text"
            },
            "cleanOnStart": false,
            "enabled": false
          }
        ],
        "name": "text",
        "enabled": false
      },
      "policy": {
        "timeUnit": "NANOSECONDS",
        "frequency": 0,
        "startupDelay": 0
      },
      "name": "text",
      "enabled": false
    },
    "processorPipeline": [
      {
        "uUID": "text",
        "metrics": {
          "metric": 0
        }
      }
    ]
  },
  "selectProcessor": {
    "context": {
      "present": false,
      "empty": false
    },
    "uUID": "text",
    "name": "text",
    "properties": {
      "property": "text",
      "empty": false
    },
    "cloneEvent": false,
    "stateful": false,
    "groupbyKey": {
      "keys": [
        "text"
      ],
      "partitionKey": 0
    },
    "jouleDBQueryInterface": {
      "name": "text",
      "enabled": false,
      "storeName": "text",
      "getInitialImage": false,
      "initialImageQuery": "text",
      "store": {
        "metaData": {
          "description": "text",
          "specification": {
            "name": "text",
            "enabled": false
          }
        },
        "storeType": "KEY_VALUE",
        "initialImage": {
          "present": false,
          "empty": false
        }
      },
      "storeType": "KEY_VALUE"
    },
    "metrics": {
      "metric": 0
    },
    "enabled": false,
    "eventsProcessed": 0,
    "eventsDiscarded": 0,
    "eventsFailed": 0,
    "eventsReceived": 0,
    "eventsIgnored": 0,
    "eventsQueued": 0,
    "responseEventType": "text",
    "projectionExpression": "text",
    "selectAttributes": [
      {
        "originalExpression": "text",
        "metricFamily": "text",
        "metrics": "text",
        "metricKey": "text",
        "metricTokens": [
          "text"
        ],
        "alias": "text",
        "type": "SEARCH_PATH",
        "searchPath": [
          "text"
        ],
        "mathProcessor": {
          "context": {
            "present": false,
            "empty": false
          },
          "uUID": "text",
          "name": "text",
          "properties": {
            "property": "text",
            "empty": false
          },
          "cloneEvent": false,
          "stateful": false,
          "groupbyKey": {
            "keys": [
              "text"
            ],
            "partitionKey": 0
          },
          "jouleDBQueryInterface": {
            "name": "text",
            "enabled": false,
            "storeName": "text",
            "getInitialImage": false,
            "initialImageQuery": "text",
            "store": {
              "metaData": {
                "description": "text",
                "specification": {
                  "name": "text",
                  "enabled": false
                }
              },
              "storeType": "KEY_VALUE",
              "initialImage": {
                "present": false,
                "empty": false
              }
            },
            "storeType": "KEY_VALUE"
          },
          "metrics": {
            "metric": 0
          },
          "enabled": false,
          "eventsProcessed": 0,
          "eventsDiscarded": 0,
          "eventsFailed": 0,
          "eventsReceived": 0,
          "eventsIgnored": 0,
          "eventsQueued": 0,
          "responseDataType": "DOUBLE",
          "expression": "text",
          "language": "text"
        },
        "queryPredicate": "text",
        "parameters": [
          "text"
        ],
        "aliase": "text"
      }
    ]
  }
}

Unregister stream

Unregister a stream processing specification from platform

DELETE/joule/management/stream/unregister

Query parameters

Response

201500501502

Successfully unregistered stream

Request

JavaScriptCurlPython

const response = await fetch('/joule/management/stream/unregister?name=quoteStream', {
    method: 'DELETE',
    headers: {},
});
const data = await response.json();