Create Export Application¶

An Export Application is designed to compile and transfer a block of data from the Kelvin Platform to a third party provider. Some common uses are to export to Amazon S3, Azure Data Lake Storage, or Databricks for ML analysis and training.

1

$ kelvin app create --app-type exporter <EXPORT_NAME>

This will give a response similar to this;

1
2
3
4

[kelvin.sdk][2025-09-02 04:26:43][I] Refreshing metadata..
[kelvin.sdk][2025-09-02 04:26:43][I] Creating new application "demoexporter"
[kelvin.sdk][2025-09-02 04:26:43][R] Successfully created new application: "demoexporter".
[kelvin.sdk][2025-09-02 04:26:43][I] Kelvin code samples are available at: https://github.com/kelvininc/app-samples

This will automatically create an Export Application bootstrap within a new directory and populated with some default files and configurations.

Folder Structure¶

You can now open the folder in your favorite IDE or editor and start to modify the files to create your Export Application.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

$ cd demoexporter
$ tree ./
├── .dockerignore
├── Dockerfile
├── app.yaml
├── main.py
├── requirements.txt
└── ui_schemas
    ├── configuration.json
    └── io_default.json

Below is a brief description of each file.

app.yaml¶

The app.yaml is the main configuration file that holds both Application definitions as well as the deployment/runtime configuration.

It is composed of the following sections:

spec_version key¶

The spec_version key is automatically injected and specifies the Export Application JSON Schema (latest) version which both defines and validates the app.yaml structure.

1

spec_version: 5.0.0

type¶

This defines the type for the application.

• app: A Smart App that allows mapping inputs and outputs to data streams, sending control changes, recommendations, and data tags.
• docker: An docker application that does not connect to the platform's data streams.
• importer: Connects to an external system to import data into the platform as well as receive control changes to act on the external system.
• exporter: Connects to the platform to export data to an external system.

1

type: exporter

info¶

The root section holds the Export Application basic information required to make itself uploadable to Kelvin's App Registry.

1
2
3
4

name: azure-data-lake-uploader
title: Azure Data Lake Uploader
description: Uploads timeseries data to Azure Data Lake Gen2
version: 1.0.0

The name is the Export Application's unique identifier.

The title and description will appear on the Kelvin UI when creating a Connector once the Export Application is uploaded.

The version defines the version of this Export Application and is used in the Kelvin UI.

flags¶

This is where you are able to set some of the Application's capabilities.

1
2
3
4
5

flags:
  enable_runtime_update:

    # enables configuration updates at runtime
    configuration: false 

exporter_io¶

This is the main section that defines the types and function of the Data Streams that are allowed for this Export Application.

1
2
3
4
5
6
7
8

# definition of which kind of datastreams are allowed to be deployed
exporter_io:
  - name: default
    data_types:    # default: [number, string, boolean, object name]
      - number
      - string

    control: true # the app allows control

ui_schemas¶

This is where the Export configuration are defined for the Kelvin UI.

The actual information is kept in a json file in the schemas folder of the project. The file location is defined in the app.yaml file like this;

1
2
3
4
5
6
7
8
9

# optional to generate UI schemas
ui_schemas:

  # app configuration schema
  configuration: "ui_schemas/configuration.json" # default: "ui_schemas/configuration.json"

  # Exporter_io configuration schema
  io_configuration:
    default: "ui_schemas/io_configuration/io_default.json" # default: "ui_schemas/io_default.json"

Below is some information on the content options in the json files.

configuration.json

The configuration.json file will come with default blank schemas when first created.

1
2
3
4
5

{
    "type": "object",
    "properties": {},
    "required": []
}

An example of a Configurations file filled in would look something like this;

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

{
    "type": "object",
    "properties": {
      "upload_interval": {
        "type": "number",
        "default": 60,
        "title": "Upload Interval",
        "minimum": 0
      }
    },
    "required": ["upload_interval"]
}

Which will be displayed on the Kelvin UI like this:

io_default.json

These are for the definition of any objects data_types.

If not defined they use the default ui_schemas/io_default.json.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

{
    "type": "object",
    "properties": {
      "address": {
        "type": "number",
        "title": "Address",
        "minimum": 1,
        "maximum": 49999
      }
    },
    "required": ["address"]
}

defaults¶

This section hold four main sections;

• system : Is used to set different system requirements/constraints within the Export Application running environment. i.e. Resources, Environment Variables, Volumes, Ports, etc.

1
2
3

defaults:
  system: {}
  configuration: {}

defaults/system section

The system section is [optional].

This is where developers can set the system settings that the Export Application needs to be able to function as intended.

This includes opening ports, setting environment variables, limited resource usage, attaching volumes and setting the privileged tag which gives extended privileges on the host system.

1
2
3
4

defaults:
  system:
    environment_vars: []
    volumes: []

1
2
3
4
5
6
7
8
9

defaults:
  system:
    resources:
      requests:   # Reserved
        cpu: 100m
        memory: 256Mi
    limits:     # Limits
        cpu: 200m
        memory: 512Mi

1
2
3
4
5
6
7
8

defaults:
environment_vars:
  - name: AZURE_ACCOUNT_NAME
    value: <% secrets.azure-account-name %>
  - name: AZURE_ACCOUNT_KEY
    value: <% secrets.azure-account-key %>
  - name: AZURE_STORAGE_CONTAINER
    value: <% secrets.azure-storage-container %>

1
2
3
4
5
6

defaults:
  system:
    volumes:
      - name: data
        target: data.db
        type: persistent

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

defaults:
  system:
    ports:
      - name: http
        type: host # Exposed on the host
        host:
        port: 80

      - name: opcua
        type: service # Exposed as a service for other containers
        service:
        port: 48010
        exposed_port: 30120
        exposed: true

1
2
3

defaults:
  system:
    privileged: true

defaults/configuration

These are the default global Export Application configuration values.

Configurations can also be optionally defined in the ui_schemas that provides a link to a JSON file containing all the information about how to display Configurations in the Kelvin UI.

1
2
3

defaults:
  configuration:
    upload_interval: 60

config.yaml¶

You use config.yaml to override default configurations defined in your app.yaml file during local development and testing.

This file provides a secure way to manage sensitive data like credentials, API keys, and environment-specific settings without committing them to version control or uploading them to the cloud platform.

The configuration loading follows a strict hierarchy: config.yaml → app.yaml defaults → graceful failure.

This ensures your SmartApp always prioritizes local overrides while falling back to safe defaults when appropriate.

The file itself will contain any hierarchy from the configurations key only.

If the app.yaml has this;

1
2
3
4
5
6
7
8
9

# app.yaml
defaults:
  configuration:
    threshold: 50.0
    interval: 60
    mqtt:
      host: <% secrets.mqtt_host %>
      username: <% secrets.mqtt_username %>
      password: <% secrets.mqtt_password %>

Then the local config.yaml will look like this;

1
2
3
4
5
6
7
8

# config.yaml
threshold: 75.0
interval: 30
mqtt:
  host: "localhost_from_config"
  username: "user_from_config"
  password: "password_from_config"
debug_mode: true

Python¶

The main.py is used as the entry point of the Export Applications. When it runs, main.py is typically the first script that gets executed, and it usually contains the main logic or orchestrates the flow of the Export Applications. However, naming a file "main.py" is just a convention, and it's not mandatory. The name helps developers quickly identify where the primary logic of the Export Applications begins.

The code example generated upon kelvin app create should be deleted and replaced as this is designed for a Kelvin SmartApp™.

Here is an example script that will compile the data from Kelvin Platform and export it to Azure Data Lake Gen2.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

import asyncio
from datetime import datetime

import aiofiles
import aiofiles.os
from kelvin.application import KelvinApp, filters

from timeseries import TimeseriesDataStore
from uploader import AzureDataLakeStorageUploader


async def upload(app: KelvinApp, data_store: TimeseriesDataStore, uploader: AzureDataLakeStorageUploader):

    # Create export dir
    await aiofiles.os.makedirs("export/", exist_ok=True)

    while True:
        batch_size = app.app_configuration.get("batch_size", 1000)
        upload_interval = app.app_configuration.get("upload_interval", 30)

        # Create filename
        export_file = f"export/{datetime.now().isoformat()}.parquet"

        try:
            # Export to parquet file
            _, chunk_size = await data_store.export_parquet(file_path=export_file, limit=batch_size)

            # Upload file if exists
            if await aiofiles.os.path.exists(export_file):
                await uploader.upload(file_path=export_file)

                # We should only trim data store if upload was successfully
                await data_store.trim(limit=batch_size)

                # Skip sleep if batch_size was full
                if chunk_size < batch_size:
                    print("No more data to upload, waiting for next interval.")
                    await asyncio.sleep(upload_interval)
                else:
                    print("Chunk was full, continuing to process without sleeping.")
            else:
                print("No data to upload at this time.")
                await asyncio.sleep(upload_interval)

        except Exception as e:
            print(f"Error occurred during upload: {e}")
            await asyncio.sleep(upload_interval)
        finally:
            # Remove file if exists
            if await aiofiles.os.path.exists(export_file):
                await aiofiles.os.remove(export_file)

async def main() -> None:

    # Configure the Azure Data Lake Uploader
    uploader = AzureDataLakeStorageUploader()

    # Configure the Timeseries Database
    data_store = TimeseriesDataStore("data.db")
    await data_store.setup()

    # Creating instance of Kelvin App Client
    app = KelvinApp()

    # Connect the App Client
    await app.connect()

    # Create task to continuously upload data
    asyncio.create_task(upload(app=app, data_store=data_store, uploader=uploader))

    # Subscribe to the asset data streams
    async for msg in app.stream_filter(filters.is_asset_data_message):
        # Insert msg to local data store
        await data_store.insert(timestamp=msg.timestamp, asset=msg.resource.asset, datastream=msg.resource.data_stream, payload=msg.payload)


if __name__ == "__main__":
    asyncio.run(main())

Supporting Files¶

1
2
3
4
5
6
7
8

FROM python:3.10-slim

ENV PYTHONUNBUFFERED=1
WORKDIR /opt/kelvin/app
COPY . /opt/kelvin/app
RUN pip install -r requirements.txt

ENTRYPOINT python main.py