如何将GridDB与Google Cloud Storage集成实现IoT数据归档与处理？

项目背景与需求 #

我正在开展一个IoT项目，使用GridDB存储并处理大量时序数据，需要将GridDB与Google Cloud Storage（GCS）集成，实现以下目标：

• 在GridDB中存储实时IoT数据
• 定期将旧数据迁移至GCS归档
• 确保数据完整性，且归档数据可高效检索用于分析

现有代码实现 #

GridDB实时数据存储代码 #

import griddb_python as griddb
import datetime

# 初始化GridDB连接
factory = griddb.StoreFactory.get_instance()
gridstore = factory.get_store(
    host='0.0.0.0',
    port=31999,
    cluster_name='Zaigham_Cluster',
    username='*',
    password='*'
)

# 创建时序容器
conInfo = griddb.ContainerInfo(
    "iot_data",
    [
        griddb.ContainerField("timestamp", griddb.Type.TIMESTAMP),
        griddb.ContainerField("sensor_id", griddb.Type.STRING),
        griddb.ContainerField("value", griddb.Type.FLOAT)
    ],
    griddb.ContainerType.TIME_SERIES
)

ts = gridstore.put_container(conInfo)
ts.set_auto_commit(False)

# 插入示例数据
now = datetime.datetime.now()
for i in range(10):
    ts.put([now, f"sensor_{i}", i * 10.0])
    now += datetime.timedelta(seconds=1)

ts.commit()

初步尝试的GCS归档代码 #

from google.cloud import storage
import csv

# 初始化Google Cloud Storage客户端
client = storage.Client()
bucket_name = 'zaigham'
bucket = client.bucket(bucket_name)
blob = bucket.blob('archived_iot_data.csv')

# 查询GridDB中的旧数据
query = ts.query("SELECT * WHERE timestamp < NOW() - INTERVAL '1 DAY'")
rs = query.fetch()

# 写入CSV并上传至GCS
with open('/tmp/archived_iot_data.csv', 'w') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(["timestamp", "sensor_id", "value"])
    while rs.has_next():
        data = rs.next()
        writer.writerow([data[0], data[1], data[2]])

blob.upload_from_filename('/tmp/archived_iot_data.csv')

优化方案：高效归档与数据保障 #

1. 高效数据传输优化 #

避免本地文件中转，直接通过内存流上传，减少IO开销：

from google.cloud import storage
import csv
from io import StringIO
import datetime

client = storage.Client()
bucket = client.bucket('zaigham')

# 查询数据
query = ts.query("SELECT * WHERE timestamp < NOW() - INTERVAL '1 DAY'")
rs = query.fetch()

# 内存中生成CSV
csv_buffer = StringIO()
writer = csv.writer(csv_buffer)
writer.writerow(["timestamp", "sensor_id", "value"])

while rs.has_next():
    data = rs.next()
    # 格式化timestamp为ISO标准格式，便于后续分析
    writer.writerow([data[0].isoformat(), data[1], data[2]])

# 重置缓冲区指针
csv_buffer.seek(0)
# 按时间戳命名文件，避免覆盖
blob = bucket.blob(f'archived_iot_data_{datetime.datetime.now().strftime("%Y%m%d_%H%M%S")}.csv')
blob.upload_from_file(csv_buffer, content_type='text/csv')

2. 定期执行机制 #

使用Python的schedule库实现定时任务，或部署为Google Cloud Function配合Cloud Scheduler触发：

import schedule
import time

def archive_task():
    # 放入上述优化后的归档代码
    print("归档任务执行完成")

# 每天凌晨2点执行归档
schedule.every().day.at("02:00").do(archive_task)

while True:
    schedule.run_pending()
    time.sleep(60)

3. 数据完整性保障 #

• 事务处理：归档后删除GridDB中的旧数据需放在事务中，确保归档成功才删除：

ts.set_auto_commit(False)
try:
    # 执行归档逻辑（上述内存流上传代码）
    # ...
    # 删除已归档数据
    delete_query = ts.query("DELETE WHERE timestamp < NOW() - INTERVAL '1 DAY'")
    delete_query.execute()
    ts.commit()
except Exception as e:
    ts.rollback()
    print(f"归档失败，回滚操作：{e}")

• 校验机制：上传后验证GCS中文件的行数与GridDB中查询到的行数一致：

# 提前统计查询结果行数
row_count = 0
temp_rs = query.fetch()
while temp_rs.has_next():
    temp_rs.next()
    row_count +=1

# 上传后获取GCS文件内容并统计行数
blob.reload()
downloaded_content = blob.download_as_text()
uploaded_row_count = len(downloaded_content.splitlines()) -1 # 减去表头行

if uploaded_row_count == row_count:
    print("数据归档完整性校验通过")
else:
    print(f"数据不一致：查询到{row_count}行，上传了{uploaded_row_count}行")

4. 归档数据高效检索优化 #

• 按时间分片命名文件：如archived_iot_data_20240501_000000.csv，便于按时间范围快速定位
• 压缩存储：上传时使用GZIP压缩，减少存储成本与传输时间：

import gzip
from io import BytesIO

# 内存中生成压缩CSV
compressed_buffer = BytesIO()
with gzip.GzipFile(fileobj=compressed_buffer, mode='w') as gz:
    gz.write(csv_buffer.getvalue().encode('utf-8'))
compressed_buffer.seek(0)

blob = bucket.blob(f'archived_iot_data_{datetime.datetime.now().strftime("%Y%m%d_%H%M%S")}.csv.gz')
blob.upload_from_file(compressed_buffer, content_type='application/gzip')

• 对接BigQuery分析：在BigQuery中创建外部表，指定GCS路径模式（如gs://zaigham/archived_iot_data_*.csv.gz），即可直接用SQL分析归档数据，无需手动导入

内容的提问来源于stack exchange，提问作者Zaigham Ali Anjum