使用 Actian Zen 和 C/C++ 进行嵌入式物联网数据库

日益增多的智能互联物联网设备为您的组织产生了呈指数级增长的数据：传感器、执行器、控制系统和智能摄像头正在捕获实时数据。因此，一个显而易见的问题是，如何管理这些数据？这些边缘设备通常资源受限 — CPU 相对较慢，内存有限，并且存储空间往往很少。

从开发人员的角度来看，边缘设备的资源限制要求代码库体积小巧且存储空间使用效率极高。嵌入式应用程序开发传统上使用 C/C++ 来编写紧凑、高效的程序，并使用相当基础的直接访问方式来操作数据。

新型嵌入式数据库正逐渐证明自己是一种有价值的替代方案，它们能够满足物联网的数据本地管理需求，提供高性能和高可用性。它们可以在不同的硬件和软件平台上运行，以捕获实时数据，高效地处理和汇总这些数据以供其他系统使用，然后分发这些数据以进行分析和做出有意义的决策。

Actian Zen

Actian Zen 是首款面向企业的嵌入式数据库，提供混合数据管理和分析功能，能够对数据中心化物联网解决方案进行缩减和扩展。

处理多个网关和边缘设备意味着在每个应用程序节点上处理多个数据库和数据管理解决方案。通常，这需要自定义数据转换和 ETL 集成代码来在嵌入式数据库和中心数据库之间移动数据，这使得混合数据系统的部署和维护成为一个难题。这不仅会减慢部署周期，还会导致不必要的复杂性和更高的管理成本。

Actian Zen 在广泛的平台上提供统一的数据类型和文件格式。它可以在边缘设备和网关之间协同工作，为开发人员提供更简单、更强大的集成体验，使他们能够创建增值应用程序功能，而无需担心定制化的 ETL 操作。

为了演示在嵌入式环境中设置和使用 Actian Zen 的便捷性，以及在服务器、桌面和嵌入式环境之间进行 Zen 编程迁移的顺畅程度，我们将以树莓派设备作为嵌入式系统，通过一个简单的示例进行介绍。我们选择树莓派是因为它价格便宜、灵活，并且类似于许多现成的物联网解决方案和原型。我们将编写一个简单的 C++ 程序来捕获设备上的时间序列数据，将其写入 Zen 数据库并检索数据。

安装

首先，让我们配置树莓派的基本设置。我们需要在树莓派上安装 Raspbian。Raspbian 是一个基于 Debian 的 GNU/Linux 发行版，是树莓派设备的官方支持和推荐操作系统。以下是刷写 Raspbian 所需的工具。

• Raspberry Pi
• 一张 4GB 的 microSD 卡

您可以在此处下载 Raspbian 操作系统。下载 ZIP 文件后，您可以选择任何软件来刷写镜像。例如，您可以使用Etcher，它可在 Windows、Mac 和 Linux 上免费下载和使用。

刷写过程完成后，将 microSD 卡插入您的树莓派。连接显示器、键盘和 USB 鼠标到树莓派。给树莓派通电并让其启动。在继续之前，请进行基本的期望设置。

本文的其余部分假设以下先决条件

• 您应该熟悉 C/C++ 编程。
• 您了解基本的SQL命令，如SELECT和INSERT。

安装 Zen

要下载适用于 Raspbian 的 Actian Zen，请访问下载页面，并在下拉菜单中选择以下选项：

• 产品：Actian Zen (PSQL)
• 版本：v13 R2 评估版
• 平台：Raspbian ARM 32 位

单击“应用过滤器”按钮以检索相关下载。

检索后，展开“Zen Edge v13 R2 试用版”节点，然后下载适用于 Raspbian ARM 的 Actian Zen Edge 数据库 v13 R2 试用版。

下载完成后，请按照以下步骤进行安装。打开 Raspbian 系统的终端窗口（本地或通过 SSH 远程访问），以默认的树莓派用户身份登录，然后输入以下命令。

$ cd Downloads  
$ sudo cp Zen-IoT-linux-13.30-035.000.armhf.tar.gz /usr/local

使用以下命令进行安装。

$ cd /usr/local  
$ sudo tar -zxf Zen-IoT-linux-13.30-035.000.armhf.tar.gz 
$ cd psql/etc  
$ sudo ./preinstall.sh  
$ sudo ./postinstall.sh 

开发环境设置

在本文中，我们只使用 C++，所以您不必严格需要 IDE。一个简单的文本编辑器就可以工作。另一方面，如果您痴迷于 IDE 进行开发，那么您可以选择 Geany、Visual Studio Code 或您偏爱的编辑器。您只需要一个简单的命令来安装 IDE。

$ sudo apt-get install geany

或者

$ curl -s https://packagecloud.io/install/repositories/headmelted/codebuilds/script.deb.sh | sudo bash 
$ sudo apt-get install code-oss 

IDE 的选择完全取决于您。您可以参考此链接，了解更多关于 C++ 开发的 IDE 和编译器信息。

我们将使用 Btrieve 2 SDK 来提供 Actian Zen 的代码访问 API。转到下载页面，并在下拉菜单中选择以下选项：

• 产品：Actian Zen (PSQL)
• 版本：SDK
• 平台：Btrieve 2

单击“应用过滤器”按钮以检索相关下载。

为您的平台选择适用于 PSQL 13 的 Btrieve 2 Linux SDK。

下载完成后，执行以下命令。

$ cd Downloads  
$ tar xf PSQL-SDK-Btrieve2API-13.30.034.000-Linux-noarch.tar.gz

执行此命令后，Btrieve 2 API 组件将被解压。为了保持目录结构整洁，让我们为开发项目创建一个新文件夹。

$ cd Desktop 
$ mkdir Programs

最后但同样重要的是，我们需要将“include”文件夹中的两个必需文件复制到我们的项目目录。

$ cp include/btrieveC.h ~/Programs  
$ cp include/btrieveCpp.h ~/Programs

时间序列演示

现在我们已经准备好了一切，我们将创建一个简单的 C++ 示例，该示例将捕获时间序列温度数据并将其写入 Actian Zen 数据库。为了模拟真实世界的例子而不使其过于复杂，我们将简单地读取内置的 SOC 温度寄存器（/sys/class/thermal/thermal_zone0/temp），并将其写入数据库。稍后我们还可以查询它以检索数据。让我们开始编码吧。

首先，创建用于捕获 CPU 状态、文件创建和加载的函数。

#include <stdio.h>
#include <string>
#include <iostream>
#include <stdlib.h>
#include <stdint.h>
#include <unistd.h>
#include <regex>
#include <vector>
#include "btrieveCpp.h"
using namespace std;
static char *btrieveFileName = (char *)"coretemp.btr";
typedef struct CpuStatus {
   double temperature;
   double frequency;
   double* cpuLoad;
} record_t;
double toDouble(std::string s) {
   std::replace(s.begin(), s.end(), ',', '.');
   return std::atof(s.c_str());
   }
string do_console_command_get_result(char* command) {
   FILE* pipe = popen(command, "r");
   if (!pipe)
      return "ERROR";
   char buffer[128];
   string result = "";
   while (!feof(pipe))
   {
      if (fgets(buffer, 128, pipe) != NULL)
         result += buffer;
   }
   pclose(pipe);
   return(result);
}
double getTemperature(int zone) {
   char buffer[100];
   sprintf(buffer, "cat /sys/class/thermal/thermal_zone%d/temp", zone);
   string result = do_console_command_get_result(buffer);
   if (result.size() > 0)  result.resize(result.size() - 1);
   return toDouble(result);
}
double getCpuFrequency(int cpu) {
   char buffer[100];
   sprintf(buffer, "cat /sys/devices/system/cpu/cpu%d/cpufreq/cpuinfo_cur_freq", cpu);
   string result = do_console_command_get_result(buffer);
   if (result.size() > 0)  result.resize(result.size() - 1);
   return toDouble(result);
}
vector<string> tokenize(const string& str, char delim) {
   vector<std::string> tokens;
   std::stringstream mySstream(str);
   string temp;
   while (getline(mySstream, temp, delim))
      tokens.push_back(temp);
      return tokens;
}
double *getMpStat() {
   double* result = new double[4];
   std::regex regex(R"(\d{2}:\d{2}:\d{2}\s+(\d)\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+\d+[.,]\d{2}\s+(\d+[.,]\d{2}))");
   string s = do_console_command_get_result((char*)"mpstat -P ALL 1 1");
   vector<string> lines = tokenize(s, '\n');
   for (size_t i = 0; i < lines.size(); ++i)
   {
      smatch matches;
      if (regex_search(lines[i], matches, regex))
      {
         int cpunum = stoi(matches[1].str());
         if (cpunum >= 0 && cpunum < 4) {
            result[cpunum] = 100 - toDouble(matches[2].str());
         }
      }
   }
   return result;
}
void printHelp() {
   printf("\n\nUsage:\n coretempmon [options]\n\n");
   printf("-h            This help\n");
   printf("-s            Silent output, just data");
   printf("-d <milliseconds>  Monitor delay in seconds (default=5)\n");
}
void monitoring(unsigned int delay, int loops, bool silent) {
   if (!silent) {
   printf("Core Temperature Monitor 1.0\n");
   printf("Scan delay is %d seconds\n", delay);
   printf("Stop monitoring using [ctrl]-[c]\n");
   printf("Time Temperature \u2103 Freq_CPU1 CPULoad1  %%CPULoad2 \%%CPULoad3 \%%CPULoad3\n");
   }
   
   bool infinite = (loops == -1);
   int counter = loops;
   record_t data;
      // Getting the CPU stats
   while (infinite || counter-- >0)
   {
      usleep(delay * 1000);
      data.cpuLoad = getMpStat();
      data.frequency = getCpuFrequency(0);
      data.temperature = getTemperature(0);
   
      printf("%0.2f \u2103 %0.0f MHz %0.2f %0.2f %0.2f %0.2f\n", data.temperature, data.frequency, data.cpuLoad[0], data.cpuLoad[1], data.cpuLoad[2], data.cpuLoad[3]);
      data.cpuLoad;
   }
}
static Btrieve::StatusCode
createFile(BtrieveClient *btrieveClient) {
   Btrieve::StatusCode status;
   BtrieveFileAttributes btrieveFileAttributes;
   // If Seting Fixed Record Length fails
   if ((status = btrieveFileAttributes.SetFixedRecordLength(sizeof(record_t))) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      printf("Error: BtrieveFileAttributes::SetFixedRecordLength():%d:%s.\n", status, Btrieve::StatusCodeToString(status));
      goto leave;
   }
   // If Creating file fails
   if ((status = btrieveClient->FileCreate(&btrieveFileAttributes, btrieveFileName, Btrieve::CREATE_MODE_OVERWRITE)) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      printf("Error: BtrieveClient::FileCreate():%d:%s.\n", status, Btrieve::StatusCodeToString(status));
      goto leave;
   }
leave:
   return status;
}
static Btrieve::StatusCode
openFile(BtrieveClient *btrieveClient, BtrieveFile *btrieveFile) {
   Btrieve::StatusCode status;
   // If opening file fails
   if ((status = btrieveClient->FileOpen(btrieveFile, btrieveFileName, NULL, Btrieve::OPEN_MODE_NORMAL)) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      printf("Error: BtrieveClient::FileOpen():%d:%s.\n", status, Btrieve::StatusCodeToString(status));
      goto leave;
   }
leave:
   return status;
}
static Btrieve::StatusCode
loadFile(BtrieveFile *btrieveFile) {
   Btrieve::StatusCode status = Btrieve::STATUS_CODE_NO_ERROR;
   return status;
}
static Btrieve::StatusCode
closeFile(BtrieveClient *btrieveClient, BtrieveFile *btrieveFile) {
   Btrieve::StatusCode status;
   // Closing the File fails.
   if ((status = btrieveClient->FileClose(btrieveFile)) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      printf("Error: BtrieveClient::FileClose():%d:%s.\n", status, Btrieve::StatusCodeToString(status));
      goto leave;
   }
leave:
   return status;
}

定义了所有必需的函数后，您可以调用它们来获取要查询的数据，并可以对数据应用不同的条件。

int main(int argc, char *argv[])
{
   BtrieveClient btrieveClient(0x4232, 0);
   Btrieve::StatusCode status = Btrieve::STATUS_CODE_UNKNOWN;
   BtrieveFile btrieveFile;
   
   // If creating the File fails.
   if ((status = createFile(&btrieveClient)) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      goto leave;
   }
   // If opening the File fails.
   if ((status = openFile(&btrieveClient, &btrieveFile)) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      goto leave;
   }
   // If loading the File fails.
   if ((status = loadFile(&btrieveFile)) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      goto leave;
   }
   else {
      int c;
      unsigned int delay = 5000;
      int loops = -1;
      bool silent = false;
      while ((c = getopt(argc, argv, "shd:n:")) != -1) {
         switch (c) {
         case 'd': {
            delay = atoi(optarg);
            if (delay < 1) { delay = 5000; }
            break;
         }
         case 'n':
            loops = atoi(optarg);
            break;
         case 's':
            silent = true;
            break;
         default:
            abort();
         }
      }
      if (delay > 0) {
         monitoring(delay, loops, silent);
         }
      }
   // If closeing the File fails.
   if ((status = closeFile(&btrieveClient, &btrieveFile)) != Btrieve::STATUS_CODE_NO_ERROR)
   {
      goto leave;
   }
leave:
   // If there wasn't a failure.
   if (status == Btrieve::STATUS_CODE_NO_ERROR)
      return 0;
   return 1;
}

上述代码循环遍历记录，并输出 CPU 温度及其频率。现在从您的编辑器中构建并运行。

请随意尝试其他示例并应用不同的查询。

上面的示例是一个简单的演示示例，用于解释用例。Zen Edge 可以做更多的事情。请随意使用上述演示示例，结合多个边缘设备进行实验，添加一些传感器，并对时间序列数据实现更多分析。