时序数据库在数据库领域的应用前景

    measurement,tag1=value1,tag2=value2 field1=value,field2=value timestamp
    
    

    temperature,device=sensor01,region=west value=25.4,humidity=0.6 1625097600000
    
    

    import time
    from datetime import datetime, timedelta
    
    class TimePartitioner:
    def __init__(self, partition_interval='1d'):
        self.interval = self._parse_interval(partition_interval)
    
    def _parse_interval(self, interval_str):
        unit = interval_str[-1]
        value = int(interval_str[:-1])
        if unit == 's':
            return timedelta(seconds=value)
        elif unit == 'm':
            return timedelta(minutes=value)
        elif unit == 'h':
            return timedelta(hours=value)
        elif unit == 'd':
            return timedelta(days=value)
        else:
            raise ValueError("Unsupported interval unit")
    
    def get_partition(self, timestamp):
        """返回给定时间戳所属的分区ID"""
        if isinstance(timestamp, int):
            dt = datetime.fromtimestamp(timestamp/1000)  # 假设是毫秒时间戳
        else:
            dt = timestamp
        partition_start = datetime(dt.year, dt.month, dt.day)
        if self.interval >= timedelta(days=1):
            # 按天分区
            return partition_start.strftime("%Y%m%d")
        else:
            # 更细粒度分区
            total_seconds = (dt - partition_start).total_seconds()
            partition_num = int(total_seconds // self.interval.total_seconds())
            return f"{partition_start.strftime('%Y%m%d')}_{partition_num}"
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/2Jn9ycQhwvOVSRirdTMLgjazqE3G.png)

    import os
    import struct
    import time
    from collections import defaultdict
    from heapq import merge
    
    class TSMEngine:
    def __init__(self, data_dir="data"):
        self.data_dir = data_dir
        self.memtable = defaultdict(dict)
        self.wal = None
        os.makedirs(data_dir, exist_ok=True)
    
    def write(self, measurement, tags, fields, timestamp):
        # 写入WAL
        if not self.wal:
            self._open_wal()
        wal_entry = self._serialize_point(measurement, tags, fields, timestamp)
        self.wal.write(wal_entry)
    
        # 写入内存表
        series_key = self._make_series_key(measurement, tags)
        self.memtable[series_key][timestamp] = fields
    
    def _open_wal(self):
        wal_path = os.path.join(self.data_dir, f"wal_{int(time.time())}.log")
        self.wal = open(wal_path, 'ab')
    
    def _serialize_point(self, measurement, tags, fields, timestamp):
        # 简化的序列化方法
        tag_str = ','.join(f"{k}={v}" for k,v in sorted(tags.items()))
        field_str = ','.join(f"{k}={v}" for k,v in sorted(fields.items()))
        line = f"{measurement},{tag_str} {field_str} {timestamp}\n"
        return line.encode('utf-8')
    
    def _make_series_key(self, measurement, tags):
        return f"{measurement}|{','.join(f'{k}={v}' for k,v in sorted(tags.items()))}"
    
    def flush(self):
        """将内存表数据刷入磁盘"""
        if not self.memtable:
            return
    
        # 创建新的TSM文件
        tsm_path = os.path.join(self.data_dir, f"data_{int(time.time())}.tsm")
        with open(tsm_path, 'wb') as f:
            for series_key, points in self.memtable.items():
                # 简化的TSM格式: [series_key][timestamp1][field1][field2]...
                f.write(struct.pack('I', len(series_key)))
                f.write(series_key.encode('utf-8'))
                for timestamp, fields in sorted(points.items()):
                    f.write(struct.pack('Q', timestamp))
                    for field_name, value in sorted(fields.items()):
                        if isinstance(value, float):
                            f.write(struct.pack('d', value))
                        elif isinstance(value, int):
                            f.write(struct.pack('q', value))
                        else:
                            raise ValueError("Unsupported field type")
    
        # 重置内存表和WAL
        self.memtable.clear()
        self.wal.close()
        self.wal = None
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/3Z1eFBHG9KEuMJsSIbgtV0rUTiNf.png)

    import struct
    import numpy as np
    
    class GorillaCompressor:
    def __init__(self):
        self.prev_value = None
        self.prev_delta = None
        self.buffer = bytearray()
        self.current_byte = 0
        self.bits_written = 0
    
    def compress_value(self, value):
        if self.prev_value is None:
            # 第一个值，直接存储
            self._write_bits(struct.pack('!d', value), 64)
            self.prev_value = value
            return
    
        delta = value - self.prev_value
        xor = int.from_bytes(struct.pack('!d', delta), 'big') ^ \
              int.from_bytes(struct.pack('!d', self.prev_delta if self.prev_delta else 0), 'big')
    
        if delta == (self.prev_delta if self.prev_delta else 0):
            # 存储'0' bit表示delta相同
            self._write_bit(0)
        else:
            # 存储'1' bit后跟xor值
            self._write_bit(1)
            leading_zeros = 64 - len(bin(xor)[2:]) if xor != 0 else 0
            significant_bits = 64 - leading_zeros if xor != 0 else 0
    
            # 存储leading zeros和significant bits
            self._write_bits(struct.pack('!B', leading_zeros), 6)
            if significant_bits > 0:
                mask = (1 << significant_bits) - 1
                self._write_bits(struct.pack('!Q', xor & mask), significant_bits)
    
        self.prev_value = value
        self.prev_delta = delta
    
    def _write_bit(self, bit):
        if bit:
            self.current_byte |= 1 << (7 - self.bits_written)
        self.bits_written += 1
    
        if self.bits_written == 8:
            self.buffer.append(self.current_byte)
            self.current_byte = 0
            self.bits_written = 0
    
    def _write_bits(self, bytes_data, num_bits):
        for i in range(num_bits):
            byte_index = i // 8
            bit_index = 7 - (i % 8)
            bit = (bytes_data[byte_index] >> bit_index) & 1
            self._write_bit(bit)
    
    def get_compressed_data(self):
        if self.bits_written > 0:
            self.buffer.append(self.current_byte)
        return bytes(self.buffer)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/ueATRX2g13KJUIQLdo0EzDflyVPv.png)

    # 安装InfluxDB
    docker run -d -p 8086:8086 -v influxdb:/var/lib/influxdb influxdb:2.0
    
    # 安装Python客户端
    pip install influxdb-client pandas matplotlib
    
    
    bash

    from influxdb_client import InfluxDBClient, Point
    from influxdb_client.client.write_api import SYNCHRONOUS
    
    # 配置连接信息
    url = "http://localhost:8086"
    token = "admin-token"  # 初始安装后创建的token
    org = "my-org"
    bucket = "iot-data"
    
    # 创建客户端
    client = InfluxDBClient(url=url, token=token, org=org)
    write_api = client.write_api(write_options=SYNCHRONOUS)
    query_api = client.query_api()
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/eBxjfIrTLqZKJVWF3s6c4ahgy5vp.png)

    import random
    import time
    from datetime import datetime, timedelta
    
    def generate_device_data(device_id, num_points=1000):
    """模拟物联网设备数据"""
    base_temp = 20 + random.uniform(-5, 5)
    base_humidity = 50 + random.uniform(-10, 10)
    points = []
    
    now = datetime.utcnow()
    for i in range(num_points):
        timestamp = now - timedelta(seconds=num_points - i)
        # 模拟昼夜温度变化
        hour = timestamp.hour
        temp_variation = 5 * math.sin(hour * math.pi / 12)
        temp = base_temp + temp_variation + random.uniform(-1, 1)
        humidity = base_humidity + random.uniform(-5, 5)
    
        point = Point("environment")\
            .tag("device_id", device_id)\
            .tag("location", random.choice(["east", "west", "north", "south"]))\
            .field("temperature", temp)\
            .field("humidity", max(0, min(100, humidity)))\
            .time(timestamp)
    
        points.append(point)
    
    return points
    
    # 写入模拟数据
    devices = [f"sensor-{i:03d}" for i in range(10)]
    for device in devices:
    data = generate_device_data(device, 1000)
    write_api.write(bucket=bucket, record=data)
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/nYuaRQb7TV312SJX8Eks4vDy0KIf.png)

    # 查询最近24小时的数据
    query = f'''
    from(bucket: "{bucket}")
      |> range(start: -24h)
      |> filter(fn: (r) => r._measurement == "environment")
      |> filter(fn: (r) => r._field == "temperature" or r._field == "humidity")
      |> aggregateWindow(every: 1h, fn: mean)
    '''
    
    result = query_api.query(query)
    data = []
    for table in result:
    for record in table.records:
        data.append({
            "time": record.get_time(),
            "measurement": record.get_measurement(),
            "field": record.get_field(),
            "value": record.get_value(),
            "device": record.values["device_id"],
            "location": record.values["location"]
        })
    
    # 转换为DataFrame分析
    import pandas as pd
    df = pd.DataFrame(data)
    pivot_df = df.pivot_table(index="time", columns=["device", "field"], values="value")
    print(pivot_df.head())
    
    
    python
    
    
![](https://ad.itadn.com/c/weblog/blog-img/images/2025-08-16/G3zrVSKxE52Tbq60XFgvCI9wchmB.png)

 * 模拟了10个物联网设备，每个设备生成1000个数据点
 * 数据包含温度和湿度指标，并添加了昼夜变化模式
 * 使用设备ID和位置作为标签，便于后续过滤和分组

 * 使用InfluxDB的Point对象构建数据点
 * 批量写入提高性能，减少网络往返
 * 自动处理时间戳转换和序列化

 * 使用Flux查询语言获取最近24小时数据
 * 按1小时间隔进行均值聚合
 * 将结果转换为Pandas DataFrame便于分析

 * 批量写入减少I/O操作
 * 使用标签进行高效过滤
 * 在数据库端进行聚合计算，减少数据传输量
 * 利用时间范围查询优化

    CPU使用率,host=web01,dc=us-west cpu=65.2,load1=3.4,load5=2.8 1625097600
    HTTP请求数,service=checkout,env=prod count=1245,error=12,avg_time=234 1625097600