OPCUA Reader

The output format is OPCUADataChangeEvent:

Sample OPCUADataChangeEvent:

{
  dataValue: 3.14
  dataIsNull: false
  dataIsNotNull: true
  dataTypeNodeId: "ns=0;i=11"
  sourceTime: 131517670918430000
  sourcePicoSeconds: null
  serverTime: 131517670918430000
  serverPicoSeconds: null
  statusCodeValue: 0
  statusCodeIsGood: true
  statusCodeIsBad: false
  statusCodeIsUncertain: false
  statusCodehasOverflowSet: false
  metadata: 
{
    "Description": {
        "locale": null,
        "text": null
    },
    "monitoringMode": "Reporting",
    "requestedQueueSize": 10,
    "readValueId": {
        "typeId": {
            "namespaceIndex": 0,
            "identifier": 626,
            "type": "Numeric",
            "null": false,
            "notNull": true
        },
        "nodeId": {
            "namespaceIndex": 2,
            "identifier": "HelloWorld/ScalarTypes/Double",
            "type": "String",
            "null": false,
            "notNull": true
        },
        "binaryEncodingId": {
            "namespaceIndex": 0,
            "identifier": 628,
            "type": "Numeric",
            "null": false,
            "notNull": true
        },
        "xmlEncodingId": {
            "namespaceIndex": 0,
            "identifier": 627,
            "type": "Numeric",
            "null": false,
            "notNull": true
        },
        "attributeId": 13,
        "indexRange": null,
        "dataEncoding": {
            "namespaceIndex": 0,
            "name": null,
            "null": true,
            "notNull": false
        }
    },
    "ValueRank": -1,
    "requestedSamplingInterval": 2000.0,
    "revisedSamplingInterval": 2000.0,
    "filterResult": {
        "bodyType": "ByteString",
        "encoded": null,
        "encodingTypeId": {
            "namespaceIndex": 0,
            "identifier": 0,
            "type": "Numeric",
            "null": true,
            "notNull": false
        }
    },
    "BrowseName": {
        "namespaceIndex": 2,
        "name": "Double",
        "null": false,
        "notNull": true
    },
    "ArrayDimensions": "-1",
    "NodeId": {
        "namespaceIndex": 2,
        "identifier": "HelloWorld/ScalarTypes/Double",
        "type": "String",
        "null": false,
        "notNull": true
    },
    "DataType": "Double",
    "clientHandle": 11,
    "monitoredItemId": 11,
    "revisedQueueSize": 10
}
}

The following sample application writes OPC-UA data to an HBase table. The table contains the most recently reported value for each node. The application is divided into two flows so that the source and CQ can run in a Forwarding Agent on the OPC-UA server.

CREATE APPLICATION OPCUAapp;

CREATE FLOW OPCUAFlow;

CREATE SOURCE OPCUASource USING OPCUAReader (
  OPCUAEndpointURL:'opc.tcp://mfactorengineering.com:4840',
  nodeIdList:'ns=1;s=EVR2.state.Empty_Box_Timer'
)
OUTPUT TO NotificationStream;

CREATE TYPE OPCUADataChange (
  data java.lang.Object,
  dataIsNull java.lang.Boolean,
  dataIsNotNull java.lang.Boolean,
  dataTypeNodeId java.lang.Object,
  sourceTime java.lang.Long,
  serverTime java.lang.Long,
  sourcePicoSeconds java.lang.Long,
  serverPicoSeconds java.lang.Long,
  statusCodeValue java.lang.Long,
  statusCodeGood java.lang.Boolean,
  statusCodeBad java.lang.Boolean,
  statusCodeUncertain java.lang.Boolean,
  statusCodeHasOverflowSet java.lang.Boolean,
  dataType java.lang.String,
  valueRank java.lang.Integer,
  arrayDimensions java.lang.Object,
  BrowseName java.lang.Object,
  Description java.lang.Object,
  monitoringMode java.lang.String,
  nodeIdNS java.lang.Integer,
  nodeIdIdentifier java.lang.Object,
  displayName java.lang.String  KEY
);
CREATE STREAM OPCUAStream OF OPCUADataChange;

CREATE CQ OPCUADataCQ
INSERT INTO OPCUAStream
SELECT dataValue,
  dataIsNull,
  dataIsNotNull,
  dataTypeNodeId,
  sourceTime,
  serverTime,
  sourcePicoSeconds,
  serverPicoSeconds,
  statusCodeValue,
  statusCodeIsGood,
  statusCodeIsBad,
  statusCodeIsUncertain,
  statusCodehasOverflowSet,
  META(x,"DataType"),
  META(x,"ValueRank"),
  TO_JSON_NODE(META(x,"ArrayDimensions")),
  TO_JSON_NODE(META(x,"BrowseName")),
  META(x,"Description"),
  META(x,"monitoringMode"),
  TO_JSON_NODE(META(x,"NodeId")).get("namespaceIndex").asInt(),
  TO_JSON_NODE(META(x,"NodeId")).get("identifier"),
  TO_JSON_NODE(META(x,"displayName")).get("text").asText()
FROM NotificationStream x;

END FLOW OPCUAFlow;

CREATE FLOW HBASEFlow;
CREATE TARGET HBaseTarget USING HBaseWriter (
  HBaseConfigurationPath:"/path/to/hbase/conf/hbase-site.xml",
  Tables: "OPCUAEvents.data",
  PKUpdateHandlingMode: "DELETEANDINSERT"
)
INPUT FROM OPCUAStream;
END FLOW HBASEFlow;

END APPLICATION OPCUAapp;