The input adapter regularly pulls data from the HES simulator and prepares the data for further processing. During deployment, you can choose to deploy the meter data generator and HES simulator stacks by choosing ENABLED for the MeterDataGenerator parameter. You also configure the number of meters and the interval between meter reads with the TotalDevices and GenerationInterval parameters, respectively. The default values for these parameters generate 50,000 reads every five minutes.

This solution includes a set of Amazon Managed Grafana dashboards. These dashboards use Amazon Athena to query data stored in a data lake and depend on several of the optional datasets for full functionality.

As shown in Figure 2 the solution’s input adapter loads meter reads from an external source (such as HES or FTP) and prepares them for processing.

A step functions workflow orchestrates the generation and download of the meter-reads file as a compressed file from the external database to the inbound bucket. Another process extracts the file from the inbound bucket and stores it in the uncompressed folder. The inbound bucket stores the compressed and uncompressed files. The solution deletes uncompressed files to save storage and costs.

After the file is extracted, the solution generates an event that invokes a Lambda function for further processing. The file-range-extractor Lambda function extracts range information from the uncompressed file based on the file size and number of chunks (configurable). A range is a group of lines that you want to process together. Extracted range information is sent to Amazon Simple Queue Service (Amazon SQS).

Each worker takes a range from the Amazon SQS queue and processes the respective meter reads (parse and transform) before sending each element to Amazon Kinesis. This process ensures that the content input file can be processed in parallel. The worker transforms the CSV line into JSON and creates a separate object for each reading type. The Amazon Kinesis data stream ingests the data into the staging area. The stream scales on demand.

You can set up connections to other data sources by configuring additional dataflows. A dataflow connects to the external database, loads data from it, and stores data in a purpose-built database that can be accessed from the solution’s central Data Catalog.

The solution comes with two sample dataflows: weather and topology. To add a new dataflow, create a data pipeline that loads data from the source, prepares them, and stores results in an appropriate data store. Then add the data store you’ve configured to the solution’s Data Catalog.

The curated data in the integration stage S3 bucket is partitioned by reading type, year, month, day, and hour, as follows:

s3://IntegrationBucket/reading_type=<reading_type_value>/year=<year>/month=<month>/day=<day>/hour=<hour>/<meter-data-file-in-parquet-format>

You can find all meter reads for the hour of a day on the lowest level of the partition tree. To optimize query performance, the data is stored in a column-based file format (Parquet).

The data lake handles late-arriving meter reads. Late-arriving meter reads are detected as soon as the data reaches the staging stage. If a late read is detected, an event is sent to Amazon EventBridge. The ETL pipeline takes care of moving the late read to the correct partition and ensures that data is stored in an optimized way.