Streaming OPC UA Data to Kinesis via SiteWise Edge Gateway

Bridging OPC UA telemetry to Kinesis Data Streams through SiteWise Edge Gateway and a custom Greengrass component.

Takahiro Iwasa

2022-08-27

6 min read

SiteWise Edge Gateway can bridge OPC UA data sources to Kinesis Data Streams, making shop-floor telemetry available to downstream AWS services.

Setting Up the Backend

OPC UA Server

To create a dummy OPC UA server, use the opcua-asyncio Python library.

Install the required package:

pip install asyncua

Download the example script (server-minimal.py) to your EC2 instance:

curl -OL https://raw.githubusercontent.com/FreeOpcUa/opcua-asyncio/master/examples/server-minimal.py

Run the script:

python server-minimal.py

Greengrass V2 Core Device

Set up a Greengrass V2 core device by following the official documentation. The installation procedure is outside the scope of this article.

After installation, perform the following:

• Deploy the aws.greengrass.StreamManager component.
• Add the IAM policy below to the Token Exchange Role to allow data transmission to Kinesis Data Streams:

1
{
2
  "Version": "2012-10-17",
3
  "Statement": [
4
    {
5
      "Effect": "Allow",
6
      "Action": [
7
        "kinesis:PutRecord",
8
        "kinesis:PutRecords"
9
      ],
10
      "Resource": "arn:aws:kinesis:*:$YOUR_AWS_ACCOUNT_ID:stream/$KINESIS_DATA_STREAM_NAME"
11
    }
12
  ]
13
}

SiteWise Edge Gateway

Set up a SiteWise Edge Gateway.

Choose Advanced setup.

The Data processing pack is not required.

Skip the Configure publisher step.

Specify the local endpoint of your OPC UA server and set the Greengrass stream name (e.g., SiteWise_Stream_Kinesis).

Review the configuration and click Create.

Kinesis Data Stream

Create a Kinesis Data Stream as the destination for your OPC UA data.

Kinesis Data Firehose

To persist the streamed data, configure a Kinesis Data Firehose delivery stream as follows:

• Source: The Kinesis Data Stream created above.
• Destination: An S3 bucket.

Creating a Greengrass Component

Create a custom Greengrass component to forward data from the Greengrass stream to Kinesis Data Streams. See the official documentation for details about component development.

Directory Structure

Organize the component’s files in the following structure:

- kinesis_data_stream.py
- recipe.yaml
- requirements.txt
- stream_manager_sdk.zip

Implementing the Component

Create requirements.txt. The stream-manager library provides access to Greengrass Stream Manager.

1
cbor2~=5.4.2
2
stream-manager==1.1.1

Create kinesis_data_stream.py with the following code. It uses the Greengrass Stream Manager SDK to forward data to a Kinesis Data Stream.

1
"""
2
Script to use Greengrass Stream Manager to stream data to a Kinesis Data Stream
3
See also https://github.com/aws-greengrass/aws-greengrass-stream-manager-sdk-python/blob/main/samples/stream_manager_kinesis.py
4
"""
5

6
import argparse
7
import asyncio
8
import logging
9
import time
10

11
from stream_manager import (
12
    ExportDefinition,
13
    KinesisConfig,
14
    MessageStreamDefinition,
15
    ReadMessagesOptions,
16
    ResourceNotFoundException,
17
    StrategyOnFull,
18
    StreamManagerClient,
19
)
20

21
logging.basicConfig(level=logging.INFO)
22
logger = logging.getLogger()
23

24

25
def main(
26
        stream_name: str,
27
        kinesis_stream_name: str,
28
        batch_size: int = None
29
):
30
    try:
31
        # Create a client for the StreamManager
32
        client = StreamManagerClient()
33

34
        # Try deleting the stream (if it exists) so that we have a fresh start
35
        try:
36
            client.delete_message_stream(stream_name=stream_name)
37
        except ResourceNotFoundException:
38
            pass
39

40
        exports = ExportDefinition(
41
            kinesis=[KinesisConfig(
42
                identifier="KinesisExport" + stream_name,
43
                kinesis_stream_name=kinesis_stream_name,
44
                batch_size=batch_size,
45
            )]
46
        )
47
        client.create_message_stream(
48
            MessageStreamDefinition(
49
                name=stream_name,
50
                strategy_on_full=StrategyOnFull.OverwriteOldestData,
51
                export_definition=exports
52
            )
53
        )
54

55
        while True:
56
            time.sleep(1)
57

58
    except asyncio.TimeoutError:
59
        logger.exception("Timed out while executing")
60
    except Exception:
61
        logger.exception("Exception while running")
62
    finally:
63
        # Always close the client to avoid resource leaks
64
        if client:
65
            client.close()
66

67

68
def parse_args() -> argparse.Namespace:
69
    parser = argparse.ArgumentParser()
70
    parser.add_argument('--greengrass-stream', required=True, default='SiteWise_Stream_Kinesis')
71
    parser.add_argument('--kinesis-stream', required=True)
72
    parser.add_argument('--batch-size', required=False, type=int, default=500)
73
    return parser.parse_args()
74

75

76
if __name__ == '__main__':
77
    args = parse_args()
78
    logger.info(f'args: {args.__dict__}')
79
    main(args.greengrass_stream, args.kinesis_stream, args.batch_size)

Create recipe.yaml with the following content. This example names the component jp.co.xyz.StreamManagerKinesis.

This recipe includes the following configurable parameters in ComponentConfiguration (lines 14-16):

• GreengrassStream: Name of the Greengrass stream
• KinesisStream: Kinesis Data Stream name to receive OPC UA data
• BatchSize: Batch size for data transfer (minimum: 1, maximum: 500)

1
# Replace $ArtifactsS3Bucket with your value to complete component registration.
2

3
RecipeFormatVersion: 2020-01-25
4

5
ComponentName: jp.co.xyz.StreamManagerKinesis
6
ComponentVersion: 1.0.0
7
ComponentDescription: Streams data in Greengrass stream to a Kinesis Data Stream.
8
ComponentPublisher: self
9
ComponentDependencies:
10
  aws.greengrass.StreamManager:
11
    VersionRequirement: '^2.0.0'
12
ComponentConfiguration:
13
  DefaultConfiguration:
14
    GreengrassStream: SiteWise_Stream_Kinesis
15
    KinesisStream: ''
16
    BatchSize: 100 # minimum 1, maximum 500
17

18
Manifests:
19
  - Platform:
20
      os: linux
21
    Lifecycle:
22
      Install: pip3 install --user -r {artifacts:decompressedPath}/component/requirements.txt
23
      Run: |
24
        export PYTHONPATH=$PYTHONPATH:{artifacts:decompressedPath}/stream_manager_sdk
25
        python3 {artifacts:decompressedPath}/component/kinesis_data_stream.py \
26
          --greengrass-stream {configuration:/GreengrassStream} \
27
          --kinesis-stream {configuration:/KinesisStream} \
28
          --batch-size {configuration:/BatchSize}
29
    Artifacts:
30
      - URI: s3://$ArtifactsS3Bucket/artifacts/jp.co.xyz.StreamManagerKinesis/1.0.0/component.zip
31
        Unarchive: ZIP

Uploading Component Artifact

Archive the component files and upload them to your S3 bucket using the following commands:

S3_BUCKET=<YOUR_BUCKET_NAME>
VERSION=1.0.0

zip component.zip kinesis_data_stream.py requirements.txt
aws s3 cp component.zip s3://$S3_BUCKET/artifacts/jp.co.xyz.StreamManagerKinesis/$VERSION/
rm component.zip

Registering Component

To register the component:

1. Copy the contents of the recipe.yaml file created earlier.
2. Replace $ArtifactsS3Bucket with the actual S3 bucket name where the component artifact resides.

Deploying Component

Click Deploy.

Click Configure component.

Update the component’s configuration with the required details. Set the KinesisStream parameter to the Kinesis Data Stream name you created earlier.

Review the configuration and deploy the component.

Once the deployment is complete, the component will stream data from the Greengrass Stream to the specified Kinesis Data Stream.

Testing

The objects landing in the S3 bucket confirm the setup is working:

aws s3 cp s3://<YOUR_BUCKET_NAME>/... ./

1
{
2
  "propertyAlias": "/MyObject/MyVariable",
3
  "propertyValues": [
4
    {
5
      "value": {
6
        "doubleValue": 7.699999999999997
7
      },
8
      "timestamp": {
9
        "timeInSeconds": 1661581962,
10
        "offsetInNanos": 9000000
11
      },
12
      "quality": "GOOD"
13
    }
14
  ]
15
}

If the data appears as expected, the system is successfully streaming OPC UA data to the Kinesis Data Stream and persisting it in S3.

Conclusion

SiteWise Edge Gateway and a custom Greengrass component can forward telemetry from a sample OPC UA server to Kinesis Data Streams and persist it in S3.

SiteWise Edge Gateway polls the OPC UA server and writes the data to a named Greengrass stream such as SiteWise_Stream_Kinesis. Because it does not export directly to Kinesis, the custom component forwards those messages. StreamManagerClient and KinesisConfig from the Stream Manager SDK handle the transfer, while BatchSize—up to 500—controls how many messages are collected for each PutRecords call.

Plan to maintain the component’s IAM role, package, and versions as part of the pipeline rather than treating the component as a one-time setup task.

About the author

Takahiro Iwasa

LinkedIn

GitHub

Software Developer

This blog shares technical notes from hands-on projects—architecture, implementation, and AWS service integrations.