Aggregations

This repository is obsolete, it is no longer being actively maintained by the organisation.

The work/domain has been superseded by the wholesale repository/domain.

Table of content

• Intro
• Architecture
• Domain Road Map
• Dataflow between domains
  • Input into the aggregations domain
    • Delta Lake (metering points)
    • Eventhub input
  • Output from the aggregations domain
    • Format of the message
    • Talking to the postoffice eventhub endpoint via the messaging framework
    • Protobuf
• How do we do aggregations?
  • Coordinator function
  • Databricks workspace
  • Databricks cluster
  • Python code
  • Dataframe results
• Getting started
  • Setting up infrastructure
  • Read more on aggregations infrastructure
• Test
  • Generating test data
  • How can you generate test data in your Delta Lake
• Triggering aggregations via coordinator
• Viewing results of aggregations

Intro

The aggregation domain is in charge of doing calculations on the time series sent to Green Energy Hub and executing the balance and wholesale settlement process.

The main calculations the domain is responsible to process are consumption, production, exchange between grid areas and the current grid loss within a grid area.
All calculations return a result for grid area, balance responsible parties and energy suppliers.

The times series sent to Green Energy Hub is processed and enriched in the Time Series domain before they can be picked up by the Aggregations domain.

The calculated results are packaged and forwarded to the legitimate Market Participants such as:

These are the business processes maintained by this domain:

Domain Overview

Domain Road Map

In the current program increment we are working on the following features:

• Deliver a calculation engine that calculates quantities, amounts and get prices per charge on flex settled metering points, so that we can execute wholesale fixing (D05) and correction settlement (D32)
• We store calculation results, so we can prepare for reporting to the post office, and for later use eg. requests of aggregated data from market actors
• We store basis data for wholesale fixing (D05) and correction settlement (D32), so we can support future control reports
• We can trigger a wholesale fixing (D05) and correction settlement process (D32) for all variants
• Protobuf contract for create event between timeseries and aggregation domain is created, so that we can implement realistic mock data, to ease the integration process later
• We will make it possible for SME's to trigger a process and view the results, so the business gets involved in the validating the solution.

Dataflow between domains

Input into the aggregations domain

Delta Lake (metering points)

The Aggregations domain does its calculation on data residing in a delta lake. This data is read in in the beginning of the aggregation job and used through out the calculations. See here how we read the data in the python code

(TBD) Reading historical data.

Eventhub input

(TBD)

Output from the aggregations domain

The coordinator has the responsibility for sending results from the aggregation jobs out of the Aggregations domain. It collects the result from the job in the CoordinatorService handles it and sends it out to a destination eventhub. This is the current implementation. But results could easily be send to another type of endpoint.

Format of the message

(TBD)

Talking to the Post Office eventhub endpoint via the messaging framework

(TBD) Link to the framework once it has been moved. Right now it is embedded as projects.

Protobuf

(TBD) Link to general description of the use of protobuf.

How do we do aggregations?

The aggregations/calculations of the metering points stored in the delta lake are done by databricks jobs containing python code utilizing pyspark.

Coordinator function

The coordinator has a descriptive name in the sense that it does what it says on the tin. It allows an external entity to trigger an aggregation job via an http interface.

Peek here to see we start and manage databricks from the coordinator Once the calculations are done the databricks job notifies the coordinator about the path of the result. The coordinator receives the path in Coordinatortriggers/ResultReceiver and from there the CoordinatorService fetches a stream of the result that the databricks job put in the blob. The format of the result is JSON which is gzip compressed. The stream is decompressed before the input is further processed. The inputprocessor finds a strategy that matches the name of the aggregation result and hands over the next steps to the strategy.

Implementing aggregation result strategies

The coordinator utilizes a strategy pattern for handling different results returned from the databricks job. The strategy is matched by the name of the result in the blob path returned.(See InputStringParserTest)

To implement a new strategy use the following approach:

public class YourResultDtoStrategy: BaseStrategy<YourResultDto>, IDispatchStrategy

public YourResultDtoStrategy(
            ILogger<YourResultDto> logger,
            Dispatcher dispatcher)
        : base(logger, dispatcher)
        {
        }

public override string FriendlyNameInstance => "your_result";

public override IEnumerable<IOutboundMessage> PrepareMessages(
            IEnumerable<YourResultDto> list,
            ProcessType processType,
            string timeIntervalStart,
            string timeIntervalEnd)
        {
            // Code for transforming list of YourResultDtos to IEnumerable<IOutboundMessage>
        }

Your Strategy is automatically DI registered during startup and will then be called when YourResult is received. If the system can't find an appropriate strategy it will log it with the following message:

IDispatchStrategy not found in input processor map. yourresult

Databricks workspace

This is the instance in which the databricks cluster resides. (TBD) When the instance is in the shared domain, describe that.

Python code

The aggregation job itself is defined by python code. The code is both compiled into a wheel file and a python file triggered by the job. The starting point for the databricks job is in aggregation_trigger.py The specific aggregations in .\source\databricks\geh_stream\aggregation_utils\aggregators these are compiled into a wheel file and installed as a library on the cluster.

Dataframe results

The results of the aggregation dataframes are combined in aggregation_trigger.py and then sent back to the coordinator as json.

Getting started

• As a general approach for getting started with aggregations we recommend setting up the infrastructure and familiarize yourself with the components involved and how to get into your databricks workspace.
• Generate some test data so you have something to calculate on top of.
• Finally: try and do some calculations by triggering the jobs.

Setting up infrastructure

The instances able to run the aggregations are created with infrastructure as code (Terraform). The code for this can be found in ./build. This IaC is triggered by github and the following describes how to get started with provisioning your own infrastructure.

Note: We use a delta lake for the time series data which is not currently commissioned by the IaC. You need to setup and reference one yourself.

(TBD) Link the general description of how Terraform and IaC works.

(TBD) Info about the shared resources and the role of the keyvault.

(TBD) Info about environments.

Read more on aggregations infrastructure

Learn more about the aggregations infrastructure here.

Test

Read about general QA that applies to the entire Green Energy Hub here.

The aggregation domain has Databricks jobs and libraries implemented in Python. Currently, the aggregation domain has a test suite of pytest unit tests. Information about and instructions on how to execute these tests are outlined here.

Generating test data

The time series test data is created using the databricks workbook.

The creation of test data is based on this file generated from the current danish DataHub system. The test data file consists of the following data properties:

How can you generate test data in your Delta Lake

The databricks workbook can be used to generate the amount of data needed and is currently configured to create time series data for more than 50 grid areas and approximately 4 million metering points.

The creation of test data is split into two parts:

1. Create test data based on this file for one hour based on old school iteration that takes a while.
2. Create test data based on latest one full hour loaded as dataframe from Delta Lake.

The reason for the split of data creation is to take full advantage of the distributed architecture of Spark and Delta Lake.

The generated time series data are setup to be stored in a delta lake where from the aggregation job fetches the data to run an aggregation upon.

If test data is created solely based on this file, then the data stored will amount to approximately 202.574.280 entries per day resulting in about 2.16 GB of storage.

Triggering aggregations via coordinator

An example:

https://azfun-coordinator-aggregations-XXXXXX.azurewebsites.net/api/KickStartJob?beginTime=2013-01-01T23%3A00%3A00%2B0100&endTime=2013-01-30T23%3A00%3A00%2B0100&processType=D03

This will ask the coordinator to do an aggregation in the specified time frame with process type D03.

Viewing results of aggregations

If you are using this domain without having a target eventhub for handling the results an alternative approach would be to change CoordinatorService and then perhaps either:

• Dump the result into a file and the inspect it.
• Log it into application log.
• Perhaps send it elsewhere.