...
IoTDB + SDT 的压缩+查询模式将参考 OSISoft Pi
参考资料:
算法论文.
基本概念
Error Compression Deviation (CD) 压缩偏差
一个数据点与线性趋势的最大差异值 E
运行程序前,预先设定的E 绝对压缩偏差值
在偏差范围内,数据被压缩并扔掉
在偏差范围外,数据不被压缩,保存下来
- Compression minimum (comp min)
- segment 之间的最小时间间隔距离
- 若 compMin = 0,记录每一个超过CD的点
- 可以设定compMin 防止noisy point 占用存储空间
- Compression maximum (comp max)
- segment 之间的最大时间间隔距离
- 不论CD 的值,若两点之间的时间距离 >= compMax,pi server会记录当前数据点并结束当前segment
- 为了更精准的压缩数据
upperDoor,当前保留数据点值+CD为支点的门
起始默认为关闭状态 Integer.min
和curUpperSlope 进行比较,取最大值,只能逆时针打开
lowerDoor,当前保留数据点值-CD为支点的门
起始默认为关闭状态 Integer.max
和curLowerSlope进行比较,取最小值,只能顺时针打开
curUpperSlope
当前数据点到segment starttime 的上斜率
curLowerSlope
当前数据点到segment starttime 的下斜率
存储上一个数据点
两扇门超过平行的时候
upperDoor >= lowerDoor
...
...
以下为,IoTDB + SDT 设计
SDT Encoder
- List<Long> timestamps
- List<T> values 支持
- List<Long> longValues
- List<Integer> intValues
- List<Double> doubleValues
- List<Float> floatValues
- encode(time, value, timeOut, valueOut) 对不同数据类型的values 需要提供不同的接口
- timestamps.add(time)
- values.add(value)
- flush(timeOut, valueOut)
- Sdt 对timestamps 和values 进行压缩(根据CD,compMax,compMin)
- 写入 byteCache
- 只针对内存中的数据进行压缩,压缩后,更新stats
SDT Decoder
- getAllSatisfiedPageData 调用 Decoder.readLong
- 返回page data的时候,Sdt 不需要进行解压
- Sdt decoder 直接返回timebuffer + valuebuffer, 在pageReader 中遍历返回的结果,pagedata.put(t, v)
...
- public boolean hasNext(ByteBuffer buffer)
- pageReader, while sdtDecoder.hasNext(timeBuffer)
- 遍历timeBuffer,返回一个时间戳
- 调用方式
- public long readLong(ByteBuffer buffer)
- pageReader, while sdtDecoder.hasNext(timeBuffer)
- long aLong = sdtDecoder.readLong(valueBuffer);
- pageData.put(t, aLong)
- 调用方式
PageWriter
需要提供的变量
- private Encoder sdtEncoder;
- private PublicBAOS timeOut, valueOut;
...
- 刷入磁盘
PageReader
需要提供的变量
- private Decoder sdtDecoder;
- private ByteBuffer buffer;
...
- 不需要解压的查询
- 范围查询
Select s1 where time > 3 and time < 10
Group by - 需要解压进行插值计算的查询
- 点查询
Select s1 where time = 3- 需要注意 time = 3 的前后两点,是否跨page
- 调用 SDTDecoder
- S1 time = 3 的数据点经过压缩,并存在了磁盘,直接返回该点
- S1 time = 3 的数据点不存在,找到s1 time = 3 的前后两个数据点并进行插值计算
- 点查询
点查询
- Ie time = x,page1, page2, page3 ..
- return null
- 在该ts 第一个和最后一个时间戳之外,不进行插值计算
- 遍历pageReaders 的时候,记录 maxPrevPair, minNextPair
- 若存在time = 3 的数据点,直接返回
- 若最后遍历完getValue == null,则通过 maxPrevPair, minNextPair 进行插值计算
- 遍历pageData 的时候,记录maxPrevPage, minNextPage
- 若生成pageReaders.size == 0,则通过 maxPrevPage, minNextPage 进行unpack page,decoder 返回插值计算的结果
- 若x < page1.startime || x > page3.endtime
- 若x >= page2.startime && x <= page2.endtime
- 若x >= page2.endtime && x <= page3.starttime
范围查询
- 与当前逻辑一致,直接返回压缩后的结果集,不进行插值计算
...