Get Stage State

Get the intermediate state data of a specified operator (stage) in a stream processing pipeline.

Prerequisites

Stream data processing pipelines have been created with the calculator library of the Stream Processing service.

Request Format

POST https://{apigw-address}/streaming/v2.0/stage-state?action=get

Request Parameters (URI)

Name

Location (Path/Query)

Mandatory/Optional

Data Type

Description

orgId

Query

Mandatory

String

The organization ID. How to get the orgId>>

Request Parameters (Body)

Name

Mandatory/Optional

Data Type

Description

pipelineId

Mandatory

String

The stream processing pipeline ID, which can be found on the EnOS Management Console > Stream Processing > Stream Operation page.

stageInstanceName

Mandatory

String

The stage instance name, which can be found on the Stream Development page. Click the stream processing pipeline name, select the target operator, and copy the stage instance name on the Info page.

assetIds

Mandatory

String

The asset ID. Supports the query of multiple asset IDs, separated by commas.

pointIds

Mandatory

String

The measurement point ID. Supports the query of multiple measurement point IDs, separated by commas. The upper limit of the number of measurement points that can be queried is 100,000.

Response Parameters

Name

Data Type

Description

data

List<JSONObject>

The list of operator intermediate state data. For more information, see items. Depending on the operator type being queried (distinguished by the stageInstanceName prefix in the request parameters), different lists of operator intermediate state data will be returned. Note: The data formats of different types of operators are also different.

items

Type 1:Generic Type

Operators of this type: LastRecordAppender, LastChangedRecordAppender, LatestRecordMerger, RecordCapturer

Name

Data Type

Description

modelId

String

The model ID.

modelIdPath

String

The model ID path.

attr

Object

The struct that contains the output results of the operator.

assetId

String

The asset ID.

pointId

String

The measurement point ID.

time

Long

The measurement point data timestamp (UNIX time, accurate to the second).

value

String

The measurement point data.

quality

Integer

The quality tagging for the measurement point data.

dq

Long

The quality indicator of the measurement point data.

Type 2:Simple Timestamp Type

Operators of this type: LatePointTagger

Name

Data Type

Description

assetId

String

The asset ID.

pointId

String

The measurement point ID.

time

Long

The measurement point data timestamp (UNIX time, accurate to the second).

Type 3:Window Aggregation Type

Operators of this type: FixedTimeWindowAggregator, SlidingTimeWindowAggregator

Name

Data Type

Description

assetId

String

The asset ID.

pointId

String

The measurement point ID.

%s::windowStrategy::%s

Object

The state information required by the window function calculation. %s represents the string determined according to information such as assets and measurement points.

%s::updated

Object

The context information for the window function. %s represents the string determined according to information such as assets and measurement points.

%s::watermark

Long

The watermark of the current asset of the window function. %s represents the string determined according to information such as assets and measurement points.

Type 4:Simple Window Aggregation Type

Operators of this type: SimplifiedTimeWindowAggregator

Name

Data Type

Description

assetId

String

The asset ID.

pointId

String

The measurement point ID.

time

Long

The measurement point data timestamp (UNIX time, accurate to the second).

windowSum

Object

The state information required by the window function calculation.

Error Code

Code

Error Information

Description

61102

Missing param xx

The parameter format or value is not valid.

61105

Too many points

The number of requested measurement points exceeds the limit.

61106

Wrong pipelineId or OU has no privilege

The provided pipeline ID is not valid, or the pipeline ID and organization ID do not match.

61107

Not supported stage type

The specified stage type is not supported.

61199

Other errors.

Samples

Request Sample

url: https://{apigw-address}/streaming/v2.0/stage-state?action=get&orgId=yourOrgId

method: POST

requestBody:
{
    "pipelineId":"yourPipelineId",
    "stageInstanceName":"yourStageInstanceName",
    "assetIds":"yourAssetIds",
    "pointIds":"yourPointIds"
}

Return Sample

{
  "code": 0,
  "msg": "OK",
  "data": {
    "items": [
  {
        "modelId": "modelId",
        "modelIdPath": "/modelIdPath",
        "assetId": "assetId",
        "pointId": "pointId",
        "time": 1582648020580,
        "value": 3.1,
        "quality": 0,
        "attr": {}
    },
    {
        "modelId": "modelId",
        "modelIdPath": "/modelIdPath",
        "assetId": "assetId2",
        "pointId": "pointId2",
        "time": 1582648020580,
        "value": 3.2,
        "quality": 1,
        "attr": {}
    }
    ]
  }
}

Java SDK Sample

import com.alibaba.fastjson.JSONObject;
import com.envision.apim.poseidon.config.PConfig;
import com.envision.apim.poseidon.core.Poseidon;
import com.envision.apim.poseidon.request.PoseidonRequest;
import org.junit.Before;
import org.junit.Test;

public class Sample {
    private static final String API_Gateway_URL = "https://{domain_url}";
    private Poseidon poseidon;

    private static class Request extends PoseidonRequest {

        public void setBodyParams(String key, Object value) {
            bodyParams().put(key, value);
        }

        public void setMethod(String method) {
            this.method = method;
        }

        private String method;

        @Override
        public String baseUri() {
            return "";
        }

        @Override
        public String method() {
            return method;
        }
    }

    @Before
    public void init() {
        poseidon = Poseidon.config(
                PConfig.init()
                        .appKey("AccessKey of your APP")
                        .appSecret("SecretKey of your APP")
        ).method("POST");
    }

    @Test
    public void GetStageState() {
        Request request = new Request();
        request.setBodyParams("assetIds", "yourAssetId");
        request.setBodyParams("pointIds", "yourPointId");
        request.setBodyParams("pipelineId", "yourPipelineId");
        request.setBodyParams("stageInstanceName", "yourStageInstanceName");

        JSONObject response = poseidon
                .url(API_Gateway_URL + "/streaming/v2.0/stage-state")
                .queryParam("orgId", "yourOrgId")
                .queryParam("action", "get")
                .getResponse(request, JSONObject.class);
        System.out.println(response);
    }
}