基于GridDB搭建基础IoT项目的初始步骤及相关问题咨询

一、确认GridDB运行状态 #

先确保你的GridDB Docker容器正常运行：

• 查看运行中容器：docker ps
• 若未运行，启动容器：docker start <你的GridDB容器ID/名称>
• 安装Python GridDB SDK：pip install griddb-python

二、模拟传感器数据采集 #

如果没有真实硬件，用Python脚本模拟温湿度传感器数据生成：

import time
import random
from datetime import datetime

def generate_sensor_data():
    while True:
        # 模拟温度（20-35℃）、湿度（40-70%）
        temperature = round(random.uniform(20, 35), 2)
        humidity = round(random.uniform(40, 70), 2)
        timestamp = datetime.now().isoformat()
        yield {"timestamp": timestamp, "temperature": temperature, "humidity": humidity}
        time.sleep(1)  # 每秒生成一条数据

# 测试生成数据
if __name__ == "__main__":
    for data in generate_sensor_data():
        print(data)

三、将传感器数据存入GridDB #

编写脚本连接GridDB，创建容器并插入数据：

import griddb_python as griddb
from datetime import datetime
import random
import time

# GridDB连接参数
factory = griddb.StoreFactory.get_instance()
try:
    # 连接集群（默认用户名admin，密码admin）
    store = factory.get_store(
        host="localhost",
        port=10001,
        cluster_name="defaultCluster",
        username="admin",
        password="admin"
    )

    # 定义容器结构（IoT传感器数据模型）
    container_info = griddb.ContainerInfo(
        "sensor_data",
        [
            ["timestamp", griddb.Type.TIMESTAMP],
            ["temperature", griddb.Type.DOUBLE],
            ["humidity", griddb.Type.DOUBLE]
        ],
        griddb.ContainerType.COLLECTION,
        True
    )

    # 获取或创建容器
    container = store.put_container(container_info)
    container.set_auto_commit(True)

    # 生成并插入数据
    def insert_sensor_data():
        while True:
            temp = round(random.uniform(20, 35), 2)
            humi = round(random.uniform(40, 70), 2)
            ts = datetime.now()
            # 插入数据
            container.put([ts, temp, humi])
            print(f"插入数据: 时间={ts}, 温度={temp}℃, 湿度={humi}%")
            time.sleep(1)

    insert_sensor_data()

except griddb.GSException as e:
    for i in range(e.get_error_stack_size()):
        print(f"Error[{i}]: {e.get_error_message(i)}")

四、基础数据查询操作 #

1. 查询最新10条传感器数据 #

# 承接上面的连接代码，获取container后执行查询
query = container.query("SELECT * ORDER BY timestamp DESC LIMIT 10")
rs = query.fetch()
print("最新10条数据:")
while rs.has_next():
    data = rs.next()
    print(f"时间: {data[0]}, 温度: {data[1]}℃, 湿度: {data[2]}%")

2. 查询指定时间段内的数据 #

from datetime import timedelta

# 查询过去10分钟内的数据
ten_min_ago = datetime.now() - timedelta(minutes=10)
query = container.query(f"SELECT * WHERE timestamp > TIMESTAMP('{ten_min_ago.isoformat()}')")
rs = query.fetch()
print("过去10分钟数据:")
while rs.has_next():
    data = rs.next()
    print(f"时间: {data[0]}, 温度: {data[1]}℃, 湿度: {data[2]}%")

3. 查询温度超过30℃的异常数据 #

query = container.query("SELECT * WHERE temperature > 30.0 ORDER BY timestamp")
rs = query.fetch()
print("温度超30℃的异常数据:")
while rs.has_next():
    data = rs.next()
    print(f"时间: {data[0]}, 温度: {data[1]}℃, 湿度: {data[2]}%")

五、测试数据集导入 #

模拟CSV测试数据集（可保存为sensor_test_data.csv） #

timestamp,temperature,humidity
2024-05-01T08:00:00,22.5,45.2
2024-05-01T08:01:00,22.7,45.5
2024-05-01T08:02:00,23.0,46.0
2024-05-01T08:03:00,23.2,46.3
2024-05-01T08:04:00,23.5,46.7
2024-05-01T08:05:00,23.8,47.0
# 可自行扩展更多行，比如生成1000条模拟数据

CSV数据导入脚本 #

import griddb_python as griddb
import csv
from datetime import datetime

# GridDB连接参数
factory = griddb.StoreFactory.get_instance()
try:
    store = factory.get_store(
        host="localhost",
        port=10001,
        cluster_name="defaultCluster",
        username="admin",
        password="admin"
    )

    # 获取已创建的sensor_data容器
    container = store.get_container("sensor_data")
    container.set_auto_commit(True)

    # 读取CSV并导入
    with open("sensor_test_data.csv", "r") as f:
        reader = csv.DictReader(f)
        for row in reader:
            ts = datetime.fromisoformat(row["timestamp"])
            temp = float(row["temperature"])
            humi = float(row["humidity"])
            container.put([ts, temp, humi])
    print("CSV数据导入完成")

except griddb.GSException as e:
    for i in range(e.get_error_stack_size()):
        print(f"Error[{i}]: {e.get_error_message(i)}")

内容的提问来源于stack exchange，提问作者gareth