摘要

本文深入讲解DolphinDB与Grafana集成的实时监控大屏技术。从Grafana安装配置到数据源连接，从实时数据展示到告警配置，从大屏设计到性能优化，全面介绍实时监控大屏的核心方法。通过丰富的代码示例，帮助读者掌握DolphinDB + Grafana实时监控的核心技能。

---

一、实时监控概述

1.1 监控大屏架构

1.2 Grafana优势

优势	说明
开源免费	社区版免费使用
多数据源	支持多种数据库
丰富图表	多种可视化组件
告警功能	内置告警系统
模板变量	动态仪表盘

1.3 适用场景

场景	说明
生产监控	实时生产数据展示
设备监控	设备状态实时追踪
能耗监控	能耗数据可视化
质量监控	质量指标实时展示

---

二、Grafana安装配置

2.1 安装Grafana

# Ubuntu/Debian
sudo apt-get install -y software-properties-common
sudo add-apt-repository "deb https://packages.grafana.com/oss/deb stable main"
wget -q -O - https://packages.grafana.com/gpg.key | sudo apt-key add -
sudo apt-get update
sudo apt-get install grafana

# 启动服务
sudo systemctl start grafana-server
sudo systemctl enable grafana-server

# 访问
# http://localhost:3000
# 默认账号: admin/admin

2.2 安装DolphinDB数据源插件

# 下载插件
git clone https://github.com/dolphindb/grafana-datasource.git

# 复制到Grafana插件目录
sudo cp -r grafana-datasource /var/lib/grafana/plugins/

# 重启Grafana
sudo systemctl restart grafana-server

2.3 配置数据源

# Grafana数据源配置
apiVersion: 1
datasources:
  - name: DolphinDB
    type: dolphindb-datasource
    access: proxy
    url: http://localhost:8848
    isDefault: true
    editable: true

---

三、DolphinDB数据准备

3.1 创建监控数据表

// 创建实时监控数据表
share streamTable(100000:0, 
    `device_id`timestamp`temperature`humidity`pressure`status,
    [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, INT]) as monitor_stream

// 启用持久化
enableTablePersistence(monitor_stream, true, true, 1000000)

// 创建分布式存储表
db = database("dfs://monitor_db", VALUE, 1..100)
schema = table(1:0, 
    `device_id`timestamp`temperature`humidity`pressure`status,
    [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, INT])
db.createPartitionedTable(schema, `sensor_data, `device_id)

// 订阅写入
subscribeTable(, "monitor_stream", "persist", -1,
    def(msg) {
        loadTable("dfs://monitor_db", "sensor_data").append!(msg)
    }, 10000, 5000)

3.2 创建聚合视图

// 创建实时聚合表
share table(1:0, 
    `device_id`time_window`avg_temp`max_temp`min_temp`cnt,
    [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, LONG]) as monitor_agg

// 时间序列聚合引擎
aggEngine = createTimeSeriesEngine("monitor_agg_engine", 60000,
    <[avg(temperature) as avg_temp,
      max(temperature) as max_temp,
      min(temperature) as min_temp,
      count(*) as cnt]>,
    monitor_agg, `timestamp, `device_id)

// 订阅聚合
subscribeTable(, "monitor_stream", "agg", -1, aggEngine, true)

3.3 创建API接口

// Grafana查询接口
def grafanaQuery(fromTime, toTime, deviceIds) {
    t = loadTable("dfs://monitor_db", "sensor_data")
    
    return select device_id, timestamp, temperature, humidity, pressure
           from t
           where timestamp between fromTime and toTime
           and device_id in deviceIds
}

// 注册函数
addFunctionView(grafanaQuery)

---

四、Grafana仪表盘配置

4.1 创建仪表盘

{
  "dashboard": {
    "title": "工业设备实时监控",
    "tags": ["industrial", "monitor"],
    "timezone": "browser",
    "panels": [],
    "refresh": "5s",
    "time": {
      "from": "now-1h",
      "to": "now"
    }
  }
}

4.2 配置面板

// 温度曲线面板
{
  "type": "graph",
  "title": "设备温度实时曲线",
  "targets": [
    {
      "datasource": "DolphinDB",
      "query": "select timestamp, temperature from monitor_stream where device_id = '$device'",
      "refId": "A"
    }
  ],
  "xaxis": {
    "mode": "time"
  },
  "yaxes": [
    {
      "label": "温度(°C)",
      "min": 0,
      "max": 100
    }
  ]
}

4.3 模板变量

// 设备选择变量
{
  "name": "device",
  "type": "query",
  "datasource": "DolphinDB",
  "query": "select distinct device_id from sensor_data",
  "refresh": 1,
  "multi": true,
  "includeAll": true
}

---

五、实时数据展示

5.1 实时曲线图

// DolphinDB查询函数
def getTemperatureCurve(deviceId, startTime, endTime) {
    t = loadTable("dfs://monitor_db", "sensor_data")
    
    return select timestamp as time, temperature as value
           from t
           where device_id = deviceId
           and timestamp between startTime and endTime
           order by timestamp
}

// 注册
addFunctionView(getTemperatureCurve)

5.2 实时统计卡片

// 实时统计
def getRealtimeStats() {
    t = loadTable("dfs://monitor_db", "sensor_data")
    
    return select count(*) as total_records,
                  avg(temperature) as avg_temp,
                  max(temperature) as max_temp,
                  min(temperature) as min_temp,
                  count(distinct device_id) as device_count
           from t
           where timestamp > now() - 3600000
}

addFunctionView(getRealtimeStats)

5.3 设备状态面板

// 设备状态统计
def getDeviceStatus() {
    t = loadTable("dfs://monitor_db", "sensor_data")
    
    return select device_id,
                  last(temperature) as current_temp,
                  last(status) as status,
                  count(*) as record_count
           from t
           where timestamp > now() - 300000
           group by device_id
}

addFunctionView(getDeviceStatus)

---

六、告警配置

6.1 Grafana告警规则

{
  "name": "温度过高告警",
  "conditions": [
    {
      "evaluator": {
        "type": "gt",
        "params": [80]
      },
      "operator": {
        "type": "and"
      },
      "query": {
        "params": ["A", "5m", "now"]
      },
      "reducer": {
        "type": "avg"
      }
    }
  ],
  "notifications": [
    {
      "type": "email",
      "settings": {
        "addresses": "admin@company.com"
      }
    }
  ]
}

6.2 DolphinDB告警表

// 告警记录表
share table(1:0, 
    `alert_time`device_id`alert_type`alert_level`value`message,
    [TIMESTAMP, SYMBOL, STRING, INT, DOUBLE, STRING]) as alert_log

// 告警触发
def checkAlert(data) {
    for (row in data) {
        if (row.temperature > 80) {
            insert into alert_log values (
                now(), row.device_id, "temperature_high", 2,
                row.temperature, "温度超过阈值: " + string(row.temperature)
            )
        }
    }
}

// 订阅告警检测
subscribeTable(, "monitor_stream", "alert", -1,
    def(msg) { checkAlert(msg) }, true)

---

七、大屏设计最佳实践

7.1 布局设计

7.2 颜色规范

状态	颜色	说明
正常	绿色	设备运行正常
警告	黄色	指标接近阈值
异常	红色	指标超限
离线	灰色	设备离线

7.3 刷新策略

数据类型	刷新间隔
实时曲线	5秒
统计卡片	10秒
设备状态	30秒
历史数据	1分钟

---

八、实战案例

8.1 完整监控大屏系统

// ========== 工业设备实时监控大屏 ==========

// 1. 创建数据表
share streamTable(100000:0, 
    `device_id`timestamp`temperature`humidity`pressure`status,
    [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, INT]) as monitor_stream

enableTablePersistence(monitor_stream, true, true, 1000000)

// 2. 创建分布式表
db = database("dfs://monitor_db", VALUE, 1..100)
schema = table(1:0, 
    `device_id`timestamp`temperature`humidity`pressure`status,
    [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, INT])
db.createPartitionedTable(schema, `sensor_data, `device_id)

// 3. 订阅写入
subscribeTable(, "monitor_stream", "persist", -1,
    def(msg) {
        loadTable("dfs://monitor_db", "sensor_data").append!(msg)
    }, 10000, 5000)

// 4. 创建聚合引擎
share table(1:0, 
    `device_id`time_window`avg_temp`max_temp`min_temp`cnt,
    [SYMBOL, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, LONG]) as monitor_agg

aggEngine = createTimeSeriesEngine("monitor_agg_engine", 60000,
    <[avg(temperature) as avg_temp,
      max(temperature) as max_temp,
      min(temperature) as min_temp,
      count(*) as cnt]>,
    monitor_agg, `timestamp, `device_id)

subscribeTable(, "monitor_stream", "agg", -1, aggEngine, true)

// 5. 创建告警表
share table(1:0, 
    `alert_time`device_id`alert_type`alert_level`value`message,
    [TIMESTAMP, SYMBOL, STRING, INT, DOUBLE, STRING]) as alert_log

// 6. Grafana查询接口
def getTemperatureCurve(deviceId, startTime, endTime) {
    t = loadTable("dfs://monitor_db", "sensor_data")
    return select timestamp as time, temperature as value
           from t
           where device_id = deviceId
           and timestamp between startTime and endTime
           order by timestamp
}

def getRealtimeStats() {
    t = loadTable("dfs://monitor_db", "sensor_data")
    return select count(*) as total_records,
                  avg(temperature) as avg_temp,
                  max(temperature) as max_temp,
                  min(temperature) as min_temp
           from t
           where timestamp > now() - 3600000
}

def getDeviceStatus() {
    t = loadTable("dfs://monitor_db", "sensor_data")
    return select device_id,
                  last(temperature) as current_temp,
                  last(status) as status
           from t
           where timestamp > now() - 300000
           group by device_id
}

addFunctionView(getTemperatureCurve)
addFunctionView(getRealtimeStats)
addFunctionView(getDeviceStatus)

// 7. 模拟数据生成
def generateMockData() {
    while (true) {
        data = table(
            take(1..10, 10) as device_id,
            take(now(), 10) as timestamp,
            rand(20.0..30.0, 10) as temperature,
            rand(40.0..60.0, 10) as humidity,
            rand(1000.0..1020.0, 10) as pressure,
            take(1, 10) as status
        )
        monitor_stream.append!(data)
        sleep(5000)
    }
}

submitJob("mock_data", "模拟数据", generateMockData)

print("实时监控大屏系统启动完成")
print("Grafana访问: http://localhost:3000")

---

九、总结

本文详细介绍了DolphinDB + Grafana实时监控大屏：

1. Grafana安装：安装配置、插件安装
2. 数据准备：监控表、聚合引擎、API接口
3. 仪表盘配置：面板配置、模板变量
4. 实时展示：曲线图、统计卡片、状态面板
5. 告警配置：告警规则、告警记录
6. 大屏设计：布局设计、颜色规范、刷新策略

思考题：

1. 如何设计高性能的实时监控大屏？
2. 如何优化Grafana查询性能？
3. 如何实现多租户监控大屏？

---

参考资料

• Grafana官方文档
• DolphinDB Grafana插件
• DolphinDB流计算

---