Создание OPC дашборда с помощью Apache Superset в Docker

В этой статье мы поговорим о передаче данных из Kafka в БД и рассмотрим наши данные в рамках Apache Superset. В данном случае мы будем использовать 2 БД - MongoDB и Postgres.

С чем мы будем работать:

Python
Docker
Kafka
OPC UA
Offset Explorer
MongoDB
Postgresql
Apache Superset

Docker-compose

Файл Compose:

services:
  data_generator:
    build:
        dockerfile: ./container/datagenerator/Dockerfile
    container_name: opc_data_generator
    restart: on-failure
    ports:
      - "4840:4840"

  zookeeper:
    image: confluentinc/cp-zookeeper:5.3.1
    container_name: zookeeper
    hostname: zookeeper
    ports:
    - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181  kafka:
    image: confluentinc/cp-kafka:5.3.1
    container_name: kafka
    hostname: kafka
    ports:
    - "9092:9092"
    restart: always
    environment:
      KAFKA_ADVERTISED_LISTENERS: LISTENER_DOCKER_INTERNAL://kafka:19092,LISTENER_DOCKER_EXTERNAL://${DOCKER_HOST_IP:-127.0.0.1}:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_DOCKER_INTERNAL:PLAINTEXT,LISTENER_DOCKER_EXTERNAL:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_DOCKER_INTERNAL
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_BROKER_ID: 1
      KAFKA_LOG4J_LOGGERS: "kafka.controller=INFO,kafka.producer.async.DefaultEventHandler=INFO,state.change.logger=INFO"
      KAFKA_DEFAULT_REPLICATION_FACTOR: 1
      KAFKA_NUM_PARTITIONS: 3
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    depends_on:
    - zookeeper  opc_producer:
    build:
        dockerfile: ./container/opc/Dockerfile
    entrypoint:
      - python
      - ./opc_producer.py
    container_name: opc_producer
    restart: always
    depends_on:
      - kafka

Установка MongoDB с помощью Docker Compose

Создайте рабочую папку и файл с помощью команды:

mkdir -p /tmp/mongodb && touch /tmp/mongodb/dd.yml

Введите в файл следующий текст:

mongo:
    image: mongo
    restart: always
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: example
    ports:
      - "27017:27017"
    volumes:
      - mongodb:/data/db

Чтобы просмотреть набор данных в Mongo, воспользуемся Mongo Express:

mongo-express:
    image: mongo-express
    restart: always
    ports:
      - 8081:8081
    environment:
      ME_CONFIG_MONGODB_ADMINUSERNAME: root
      ME_CONFIG_MONGODB_ADMINPASSWORD: example
      ME_CONFIG_MONGODB_URL: mongodb://root:example@mongo:27017/

После установки веб-сервиса подключитесь через localhost и проверьте, работает ли сервис с помощью http://localhost:8081/.

Установка PostgreSQL с помощью Docker Compose

Для мониторинга данных мы будем использовать порт PostgreSQL, установленный по умолчанию - 5432.

Для того чтобы настроить пользователя и пароль Postgres, а также создать учетную запись Superset, в папке проекта создайте файл .env:

POSTGRES_USER=username
POSTGRES_PASSWORD=password
POSTGRES_DB=postgres
SUPERSET_CONFIG=superset.config
SUPERSET_ENV=production
SUPERSET_ADMIN_USERNAME=admin
SUPERSET_ADMIN_PASSWORD=admin
SUPERSET_ADMIN_FIRST_NAME=Admin
SUPERSET_ADMIN_LAST_NAME=User
POSTGRES_HOST=postgres
POSTGRES_PORT=5432
SQLALCHEMY_ENABLE_POOL_PRE_PING=true
SUPERSET_SECRET_KEY=openssl rand -base64 42

Обязательно создайте persist volumes для нашей БД, а также базу данных на основе SQL-оператора и поместите ее в папку db_setup:

ALTER DATABASE postgres SET timezone TO 'Asia/Kuala_Lumpur';
CREATE TABLE opc_sensors(
    datatime TIMESTAMP,
    opc_pressure FLOAT,
    opc_temperature FLOAT
);

Docker compose для PostgreSQL будет выглядеть следующим образом:

db:
    image: postgres
    restart: always
    ports:
      - "5432:5432"
    env_file:
      - .env
    volumes:
      - postgres_db:/var/lib/postgresql/data
      - ./db_setup:/docker-entrypoint-initdb.d

Для управления базой данных мы будем использовать pgAdmin:

pgadmin:
    image: dpage/pgadmin4
    container_name: pgadmin4_container
    restart: always
    ports:
      - "5050:80"
    environment:
      PGADMIN_DEFAULT_EMAIL: admin@test.com
      PGADMIN_DEFAULT_PASSWORD: admin
    volumes:
      - pgadmin-data:/var/lib/pgadmin
    depends_on:
      - db

После установки веб-сервиса подключитесь через localhost и проверьте, работает ли сервис с помощью http://localhost:5050/.

Установка Apache Superset с помощью Compose

superset:
    image: apache/superset
    container_name: superset
    env_file:
      - .env
    ports:
      - 8088:8088
    volumes:
      - superset:/app
    depends_on:
      - db
    command: >
      sh -c "superset fab create-admin --username admin --firstname Superset --lastname Admin --email admin@superset.com --password admin &&
      superset db upgrade &&
      superset init &&
      superset run -p 8088 -h 0.0.0.0"

Далее мы создадим Superset приложение, используя официальный образ Apache Superset. Для того, чтобы убедиться в том, что наше приложение работает, необходимо создать администратора Superset и инициализировать его перед запуском приложения с помощью приведенной выше команды.

После установки веб-сервиса подключитесь через localhost и проверьте, работает ли сервис с помощью http://localhost:8088.

Создайте Persist Volumes

volumes:
  mongodb:
  postgres_db:
  pgadmin-data:
  superset:

Создайте Kafka Consumer в Docker

Создадим контейнер opc_consumer, который использует образ Python. Поскольку мы будем получать данные из Kafka, контейнер должен зависеть от контейнеров Kafka и DB, которые мы создали прежде:

opc_consumer:
    build:
        dockerfile: ./container/opc_consumer/Dockerfile
    container_name: opc_consumer
    restart: always
    depends_on:
      - kafka
      - db

FROM python:3.7-slim

RUN pip install \
    asyncua==1.0.2 \
    pymongo \
    confluent_kafka\
    psycopg2-binaryWORKDIR /opt/opc_mockup/opcCOPY ../../opc /opt/opc_mockup/opcCMD ["python", "./opc_streaming.py"]

Далее создадим Dockerfile для сборки нашего Python-приложения. Установим абсолютно все пакеты, необходимые для приложения. Скопируем файл приложения Python и запустим его с помощью CMD.

Создадим opc_streaming.py внутри папки opc:

from confluent_kafka import Consumer, KafkaException, KafkaError
from pymongo import MongoClient

from datetime import datetime
import json
import psycopg2conf = {
    'bootstrap.servers': 'kafka:19092',  # Kafka broker address
    'group.id': 'console-consumer-92606',        # Consumer group ID
    'auto.offset.reset': 'earliest',        # Start consuming from the beginning of the topic
    'enable.auto.commit': False             # Disable auto commit
}consumer = Consumer(conf)consumer.subscribe(['opc_server'])  # topic name

Теперь напишем наше приложение на Python, импортировав все необходимые библиотеки. Настроим Kafka Consumer, следуя инструкциям, приведенным выше.

Убедитесь в том, что Вы правильно указали идентификатор Consumer. Это можно сделать в Offset Explorer.

Для нашей базы данных NoSQL в качестве базы данных журналов мы будем использовать MongoDB. Ранее мы уже настраивали конфигурацию базы данных в файле compose. Вы можете изменить имя базы данных по своему усмотрению.

# Set the MongoDB connection details
mongodb_uri = 'mongodb://root:example@mongo:27017'
db_name = 'opc_sensors'
collection_name = 'sensors'

# Create a MongoDB client and connect to the database
client = MongoClient(mongodb_uri)
db = client[db_name]
collection = db[collection_name]

Подключитесь к базе данных PostgreSQL, которую мы затем будем использовать в Apache Superset в целях визуализации данных:

# Set up PostgreSQL connection
conn = psycopg2.connect(
    host='db',
    port='5432',
    dbname='postgres',
    user='username',
    password='password'
)

conn.autocommit = True
cursor = conn.cursor()table_name = 'opc_sensors'# Prepare SQL statement
sql_insert = "INSERT INTO opc_sensors (datatime, opc_pressure,opc_temperature) VALUES (%s, %s, %s)"

Вставьте данные в MongoDB с помощью collection.insert_one. Далее выполните созданный нами SQL-оператор:

while True:
    try:
        msg = consumer.poll(1.0)  # Poll for new messages (1.0 second timeout)

        if msg is None:
            continue        if msg.error():
            if msg.error().code() == KafkaError._PARTITION_EOF:
                # Reached end of partition, continue to the next one
                continue
            else:
                raise KafkaException(msg.error())
        value = json.loads(msg.value().decode('utf-8'))  # Assuming the message value is a string
        # Process the received message
        print(f'Received message: {value}')        # Insert the message value into MongoDB
        collection.insert_one(value)
        # Insert data into PostgreSQL
        cursor.execute(sql_insert, ((datetime.fromtimestamp(value['timestamp'])), value['opc_pressure'], value['opc_temperature']))
        # Manually commit the offset to mark the message as processed
        consumer.commit(msg)    except KeyboardInterrupt:
        breakconsumer.close()  # Close the Kafka consumer and PostgreSQL connection
conn.close()

Теперь мы можем визуализировать данные и построить график:

Подключитесь к PostgreSQL.

Наведите курсор на + и нажмите на «Подключить базу данных».

Выберите PostgreSQL

Заполните поля

Нажмите на «Создать дашборд»

Найдите нашу таблицу и нажмите на «Создать набор данных» и на «Создать диаграмму».

Выберите наш набор данных и выберите «Линейная диаграмма»

Создайте диаграмму в соответствии с тем, как показано ниже. Не стесняйтесь и смело экспериментируйте с цветами, формами и т.д. Когда все будет готово, нажмите на кнопку «Создать».