MariaDB SQL 操作

MariaDB SQL 操作

概述

本节介绍如何在 Node-RED 中执行 MariaDB SQL 操作。学习完成后，您将能够：

• 安装和配置 MySQL/MariaDB 节点
• 编写 INSERT 语句存储设备数据
• 编写 SELECT 语句查询历史记录
• 理解关系数据库在 IoT 中的使用场景

Installing MySQL Node

# 通过 Node-RED 面板安装
# Manage Palette → Install → "node-red-node-mysql"

# 或命令行安装
docker exec -it nodered sh
cd /data
npm install node-red-node-mysql
exit
docker restart nodered

Database Configuration

创建数据库和表

-- 在 MariaDB 中执行
CREATE DATABASE iot_demo;
USE iot_demo;

-- 设备数据表
CREATE TABLE sensor_data (
    id INT AUTO_INCREMENT PRIMARY KEY,
    device_id VARCHAR(50),
    temperature FLOAT,
    humidity FLOAT,
    reading_time INT,  -- Unix timestamp
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_device (device_id),
    INDEX idx_time (reading_time)
);

-- 告警记录表
CREATE TABLE alerts (
    id INT AUTO_INCREMENT PRIMARY KEY,
    device_id VARCHAR(50),
    alert_type VARCHAR(50),
    message TEXT,
    severity ENUM('info','warning','critical'),
    is_read BOOLEAN DEFAULT FALSE,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Node-RED MySQL 节点配置

{
    "host": "mariadb",
    "port": "3306",
    "db": "iot_demo",
    "user": "root",
    "password": "YourPassword2024!"
}

INSERT Operations

插入传感器数据

[Inject: 模拟数据] → [Function: 构建 SQL] → [MySQL Node] → [Debug]

// Function 节点: 构建 INSERT SQL
var data = msg.payload;

// 获取 Unix 时间戳
var timestamp = Math.floor(Date.now() / 1000);

// 构建 SQL 语句
var sql = "INSERT INTO sensor_data " +
    "(device_id, temperature, humidity, reading_time) VALUES " +
    "('" + data.device + "', " +
    data.temperature + ", " +
    data.humidity + ", " +
    timestamp + ")";

// MySQL 节点读取 msg.topic 作为 SQL 语句
msg.topic = sql;

return msg;

插入告警记录

// Function 节点: 插入告警
var alert = msg.payload;

var sql = "INSERT INTO alerts " +
    "(device_id, alert_type, message, severity) VALUES " +
    "('" + alert.device + "', " +
    "'" + alert.type + "', " +
    "'" + alert.message.replace(/'/g, "''") + "', " +
    "'" + alert.severity + "')";

msg.topic = sql;
return msg;

SELECT Operations

查询最近记录

// Function 节点: 查询最新温度数据
var deviceId = msg.payload.device || "SENSOR-01";

var sql = "SELECT * FROM sensor_data " +
    "WHERE device_id = '" + deviceId + "' " +
    "ORDER BY reading_time DESC LIMIT 10";

msg.topic = sql;
return msg;

聚合查询

// Function 节点: 查询统计信息
var sql = "SELECT " +
    "device_id, " +
    "AVG(temperature) as avg_temp, " +
    "MAX(temperature) as max_temp, " +
    "MIN(temperature) as min_temp, " +
    "COUNT(*) as reading_count " +
    "FROM sensor_data " +
    "WHERE reading_time > UNIX_TIMESTAMP() - 3600 " +
    "GROUP BY device_id";

msg.topic = sql;
return msg;

Complete Data Flow

数据流: MQTT → MariaDB

[ESP32] ──MQTT──→ [Mosquitto]
                       │
                       ▼
              [MQTT In: sensor/+]
                       │
                       ▼
              [JSON: 解析]
                       │
                       ▼
              [Function: 构建 INSERT]
                       │
                       ▼
              [MySQL: INSERT]
                       │
                       ▼
              [Function: 验证结果]
                       │
                       ▼
              [Debug: 日志]

// Function: 验证写入结果
if (msg.payload && msg.payload.affectedRows > 0) {
    node.log("Data inserted successfully: " + msg.payload.insertId);
    msg.payload = {
        status: "success",
        insertId: msg.payload.insertId,
        timestamp: Date.now()
    };
} else {
    node.error("Insert failed");
    msg.payload = {
        status: "error",
        message: "Database write failed"
    };
}
return msg;

SQL Injection Protection

// 使用参数化查询防止 SQL 注入
// MySQL 节点支持 msg.payload 作为数组传参

var data = msg.payload;

// 方法 1: 使用模板字符串（推荐）
msg.topic = "INSERT INTO sensor_data (device_id, temperature) VALUES (?, ?)";
msg.payload = [data.device_id, data.temperature];

// 方法 2: 先验证再拼接
function sanitize(value) {
    if (typeof value === 'string') {
        return value.replace(/['"\\;]/g, '').substring(0, 50);
    }
    return value;
}

var safeDevice = sanitize(data.device_id);
var sql = "INSERT INTO sensor_data (device_id, temperature) VALUES " +
    "('" + safeDevice + "', " + Number(data.temperature) + ")";
msg.topic = sql;

return msg;

Using Adminer

Adminer 提供 Web 界面管理 MariaDB：

services:
  adminer:
    image: adminer:latest
    container_name: adminer
    restart: unless-stopped
    ports:
      - "8080:8080"

访问 http://localhost:8080:

System: MySQL
Server: mariadb (容器名)
Username: root
Password: YourPassword2024!
Database: iot_demo

Backup and Export

# 通过 Node-RED 触发备份
# Inject → Function → Exec Node

# Function: 构建备份命令
var date = new Date().toISOString().split('T')[0];
var filename = "/backup/iot_demo_" + date + ".sql";

msg.payload = "mysqldump -u root -pYourPassword2024! iot_demo > " + filename;
return msg;

InfluxDB vs MariaDB Decision

场景	推荐数据库	原因
传感器时序数据	InfluxDB	高压缩、高效时序查询
设备元数据	MariaDB	关系模型、支持更新
告警记录	MariaDB	灵活查询、支持复杂条件
用户和权限	MariaDB	ACID 事务、数据一致性
实时监控数据	InfluxDB	高性能写入、自动过期

最佳实践

✅ 推荐做法:

• IoT 时序数据用 InfluxDB，业务数据用 MariaDB
• 使用参数化查询防止 SQL 注入
• 定期备份数据库
• 为查询字段创建索引
• 使用连接池管理数据库连接

❌ 避免做法:

• 在 MariaDB 中存储高频时序数据
• SQL 语句直接拼接用户输入
• 使用 root 账户连接应用
• 频繁执行全表扫描查询
• 忽略数据库连接失败处理

Summary

1. MySQL/MariaDB 节点提供完整的 SQL 操作能力
2. INSERT 用于存储设备数据和业务记录
3. SELECT 支持复杂查询和聚合分析
4. Adminer 提供 Web 数据库管理界面
5. 参数化查询防止 SQL 注入

References

• node-red-node-mysql 文档
• MariaDB SQL 语法
• Adminer 工具