背景

今年在公司內(nèi)部主導(dǎo)了兩個(gè)的行情數(shù)據(jù)系統(tǒng)的構(gòu)建，兩者均使用到了常見的時(shí)序數(shù)據(jù)壓縮算法。

這里簡(jiǎn)單總結(jié)一下過程中積累的一些經(jīng)驗(yàn)。

讓我們先來思考一個(gè)問題：壓縮算法生效的前提是什么？

數(shù)據(jù)本身至少要符合以下兩種特性其一：

• 數(shù)據(jù)存在冗余
• 數(shù)據(jù)符合特定的概率分布

在時(shí)序數(shù)據(jù)領(lǐng)域，數(shù)據(jù)冗余度與相似度較高，因此天生適合進(jìn)行壓縮。

但對(duì)于不同類型的數(shù)據(jù)，其所適用的壓縮算法也大相徑庭。

下面我們逐一介紹這些數(shù)據(jù)相應(yīng)的壓縮算法。

整數(shù)

整型數(shù)據(jù)是構(gòu)建各種應(yīng)用的基石，時(shí)序型應(yīng)用也不例外。
在行情數(shù)據(jù)中，存在大量的整型數(shù)據(jù)，例如：逐筆成交中的時(shí)間戳、成交量。

根據(jù)壓縮算法的不同，可以將整型數(shù)據(jù)分為以下 3 類：

• 無符號(hào)整型 —— Varint
• 有符號(hào)整型 —— ZigZag
• 時(shí)間戳 —— Delta2 + Simple8b

Varint

一個(gè) 32 位的無符號(hào)整型能表達(dá) 0 - 4294967295 之間的任意數(shù)字
但這些數(shù)字在日常生活中出現(xiàn)的概率并不是均勻分布的，一個(gè)著名的例子是本福特定律，該定律常被用于辨別數(shù)據(jù)的真?zhèn)巍?/p>

通常情況下，較小的數(shù)字出現(xiàn)的概率會(huì)高于極大的數(shù)據(jù)。
以年齡為例，無論人口如何分布，大部分人的年齡都位于 0 ~ 100 之間。

表示 128 僅需要 7bit 足矣，如果使用 32bit 的無符號(hào)整型進(jìn)行存儲(chǔ)，意味著至少浪費(fèi)了 24bit。

幸運(yùn)的是，我們能通過一種自適應(yīng)編碼方式來減少這種浪費(fèi) —— Varint。

public class VarIntCodec {

    static int encodeInt(int v, byte[] bytes, int offset) {
        if (v < 0) {
            throw new IllegalStateException();
        } else if (v < 128) {
            bytes[offset++] = (byte) v;
        } else if (v < 16384) {
            bytes[offset++] = (byte) (v | 0x80);
            bytes[offset++] = (byte) ((v >>> 7) & 0x7F);
        } else if (v < 2097152) {
            bytes[offset++] = (byte) (v | 0x80);
            bytes[offset++] = (byte) ((v >>> 7) | 0x80);
            bytes[offset++] = (byte) (v >>> 14);
        } else if (v < 268435456) {
            bytes[offset++] = (byte) (v | 0x80);
            bytes[offset++] = (byte) ((v >>> 7) | 0x80);
            bytes[offset++] = (byte) ((v >>> 14) | 0x80);
            bytes[offset++] = (byte) (v >>> 21);
        } else {
            bytes[offset++] = (byte) (v | 0x80);
            bytes[offset++] = (byte) ((v >>> 7) | 0x80);
            bytes[offset++] = (byte) ((v >>> 14) | 0x80);
            bytes[offset++] = (byte) ((v >>> 21) | 0x80);
            bytes[offset++] = (byte) (v >>> 28);
        }
        return offset;
    }

    static int decodeInt(byte[] bytes, int[] offset) {
        int val;
        int off = offset[0];
        byte b0, b1, b2, b3;
        if ((b0 = bytes[off++]) >= 0) {
            val = b0;
        } else if ((b1 = bytes[off++]) >= 0) {
            val = (b0 & 0x7F) + (b1 << 7);
        } else if ((b2 = bytes[off++]) >= 0) {
            val = (b0 & 0x7F) + ((b1 & 0x7F) << 7) + (b2 << 14);
        } else if ((b3 = bytes[off++]) >= 0) {
            val = (b0 & 0x7F) + ((b1 & 0x7F) << 7) + ((b2 & 0x7F) << 14) + (b3 << 21);;
        } else {
            val = (b0 & 0x7F) + ((b1 & 0x7F) << 7) + ((b2 & 0x7F) << 14) + ((b3 & 0x7F) << 21) + (bytes[off++] << 28);
        }
        offset[0] = off;
        return val;
    }

}

通過觀察代碼可以得知，Varint 編碼并不是沒有代價(jià)的：每 8bit 需要需要犧牲 1bit 作為標(biāo)記位。

對(duì)于值較大的數(shù)據(jù)，Varint 需要額外的空間進(jìn)行編碼，從而導(dǎo)致更大的空間消耗。

因此使用 Varint 的前提有兩個(gè)：

• 數(shù)據(jù)較小
• 沒有負(fù)數(shù)

ZigZag

Varint 編碼負(fù)數(shù)的效率很低：

對(duì)于 big-endian 數(shù)據(jù)來說，Varint 是通過削減 leading-zero 來實(shí)現(xiàn)的壓縮。

而負(fù)數(shù)的首個(gè) bit 永遠(yuǎn)非零，因此不但無法壓縮數(shù)據(jù)，反而會(huì)引入不必要的空間開銷。

ZigZag 通過以下方式來彌補(bǔ)這一缺陷：

增加小負(fù)數(shù)的 leading-zero 數(shù)量，然后再進(jìn)行 Varint 編碼。

其原理很簡(jiǎn)單，增加一個(gè) ZigZag 映射：

• Varint 編碼前增加映射 (n << 1) ^ (n >> 31)
• Varint 解碼后增加映射 (n >>> 1) ^ -(n & 1)

當(dāng) n = -1 時(shí)，Varint 編碼前映射：

n = -1      -> 11111111111111111111111111111111
a = n << 1  -> 11111111111111111111111111111110
b = n >> 31 -> 11111111111111111111111111111111
a ^ b       -> 00000000000000000000000000000001

當(dāng) n = -1 時(shí)，Varint 解碼后映射：

m = a ^ b    -> 00000000000000000000000000000001
a = m >>> 1  -> 00000000000000000000000000000000
b = -(m & 1) -> 11111111111111111111111111111111
a ^ b        -> 11111111111111111111111111111111

ZigZag 映射能夠有效增加小負(fù)數(shù)的 leading-zero 數(shù)量，進(jìn)而提高編碼效率。

ZigZag 本身也并不是完美的的：

• 占用了部分非負(fù)數(shù)的編碼空間
• 對(duì)于大負(fù)數(shù)沒有優(yōu)化效果

Delta2

時(shí)間戳是時(shí)序數(shù)據(jù)中的一類特殊的數(shù)據(jù)，其值往往比較大，因此不適用于 Varint 編碼。
但是時(shí)間戳本身具有以下兩個(gè)特性：

• 非負(fù)且單調(diào)遞增
• 數(shù)據(jù)間隔較為固定

前面提到過：提高整型數(shù)據(jù)壓縮率的一個(gè)重要手段是增加 leading-zero 數(shù)量。
說人話就是：盡可能存儲(chǔ)較小的數(shù)字。

因此，相較于存儲(chǔ)時(shí)間戳，存儲(chǔ)前后兩條數(shù)據(jù)的時(shí)間間隔是一個(gè)更好的選擇。

第一種方式是使用 Delta 編碼：

• 第一個(gè)數(shù)據(jù)點(diǎn)，直接存儲(chǔ) t0t0
• 第 n 個(gè)數(shù)據(jù)點(diǎn)，則存儲(chǔ) tn−t0tn−t0

但是 Delta 編碼的數(shù)據(jù)冗余仍然較多，因此可以改進(jìn)為 Delta2 編碼：

• 第一個(gè)數(shù)據(jù)點(diǎn)，直接存儲(chǔ) t0t0
• 第 n 個(gè)數(shù)據(jù)點(diǎn)，則存儲(chǔ) tn−tn−1tn−tn−1

編碼后的 int64 的時(shí)間戳，可以用 int32 甚至更小的數(shù)據(jù)類型進(jìn)行存儲(chǔ)。

Simple8b

朋友，你覺得 Varint 與 Delta2 已經(jīng)是整型壓縮的極限了嗎？

某些特殊的時(shí)序事件，可能以很高的頻率出現(xiàn)，比如行情盤口報(bào)價(jià)。

這類數(shù)據(jù)的時(shí)間戳進(jìn)行 Delta2 編碼后，可能會(huì)出現(xiàn)以下兩種情況：

• 場(chǎng)景1：很多連續(xù)的 0 或 1 區(qū)間
• 場(chǎng)景2：大部分?jǐn)?shù)據(jù)分布在 0 ~ 63 的范圍內(nèi)

對(duì)于場(chǎng)景1，游程編碼（RLE）是個(gè)不錯(cuò)的選擇，但普適性較差。

對(duì)于場(chǎng)景2，每個(gè)值僅需 6bit 存儲(chǔ)即可，使用 Varint 編碼仍有空間浪費(fèi)。

Simple8b 算法能較好的處理這一問題，其核心思想是：

將盡可能多數(shù)字打包在一個(gè) 64bit 長(zhǎng)整型中。

Simple8b 將 64 位整數(shù)分為兩部分：

• selector(4bit) 用于指定剩余 60bit 中存儲(chǔ)的整數(shù)的個(gè)數(shù)與有效位長(zhǎng)度
• payload(60bit) 則是用于存儲(chǔ)多個(gè)定長(zhǎng)的整數(shù)
  根據(jù)一個(gè)查找表，將數(shù)據(jù)模式匹配到最優(yōu)的 selector，然后將多個(gè)數(shù)據(jù)編碼至 payload

Simple8b 維護(hù)了一個(gè)查找表，可以將數(shù)據(jù)模式匹配到最優(yōu)的 selector，然后將多個(gè)數(shù)據(jù)編碼至 payload。編碼算法可以參考這個(gè)Go實(shí)現(xiàn)。

需要指出的是，這個(gè)開源實(shí)現(xiàn)使用回溯法進(jìn)行編碼，其復(fù)雜度為 O(n2)O(n2)（ n 為輸入數(shù)據(jù)長(zhǎng)度）。該實(shí)現(xiàn)對(duì)于離線壓縮來說已經(jīng)足夠，但對(duì)于實(shí)時(shí)壓縮來說稍顯不足。

我們?cè)诖舜a的基礎(chǔ)上，使用查表法維護(hù)了一個(gè)狀態(tài)轉(zhuǎn)移數(shù)組，將編碼速度提升至 O(n)O(n)，使其能夠應(yīng)用于實(shí)時(shí)壓縮。

浮點(diǎn)數(shù)

浮點(diǎn)數(shù)是一類較難壓縮的數(shù)據(jù)，IEEE 705 在編碼上幾乎沒有浪費(fèi)任何一個(gè) bit，因此無法使用類似 Varint 的方式進(jìn)行壓縮：

0131230signexponent (8-bit)fraction (23-bit)= 0.15625

目前常用的壓縮方式可以分為兩類：

• 有損壓縮 —— 丟棄不必要的精度后，使用整數(shù)進(jìn)行表示
• 無損壓縮 —— XOR 編碼

有損壓縮

有損壓縮需要配合業(yè)務(wù)領(lǐng)域知識(shí)來使用，因?yàn)槭紫纫_定需要保留的精度。

當(dāng)精度確定之后，就可以將浮點(diǎn)數(shù)轉(zhuǎn)換為整數(shù)，然后使用 Varint 進(jìn)行編碼。

public static ScaledResult scaleDecimal(float[] floats, final int scaleLimit) throws CodecException {

    BigDecimal[] decimals = new BigDecimal[floats.length];
    for (int i=0; i<floats.length; i++) {
        decimals[i] = new BigDecimal(Float.toString(floats[i]));
    }

    BigDecimal base = null;
    int maxScale = 0;
    for (BigDecimal decimal : decimals) {
        Preconditions.checkState(decimal.signum() >= 0);
        int scale = decimal.scale();
        if (scale == 1 && maxScale == 0 && decimal.stripTrailingZeros().scale() <= 0) {
            scale = 0;
        }
        maxScale = Math.max(maxScale, scale);
        base = base == null || decimal.compareTo(base) < 0 ? decimal : base;
    }
    final int scale = Math.min(maxScale, scaleLimit);
    final long scaledBase = base.scaleByPowerOfTen(scale).setScale(0, RoundingMode.HALF_UP).longValueExact();
    long[] data = new long[decimals.length];
    for (int i=0; i<decimals.length; i++) {
        long scaledValue = decimals[i].scaleByPowerOfTen(scale).setScale(0, RoundingMode.HALF_UP).longValueExact();
        data[i] = scaledValue - scaledBase;
    }
    return new ScaledResult(scale, scaledBase, data);
}

這個(gè)壓縮算法的優(yōu)點(diǎn)主要是直觀，并且天然跨語言，比較適合前后端交互。

不過也存在以下局限性：

• 只能壓縮精度已知的數(shù)據(jù)
• 數(shù)據(jù)中不能同時(shí)出現(xiàn)正數(shù)與負(fù)數(shù)
• 編碼速度較慢

編碼速度慢很大的原因，是在 float 轉(zhuǎn)換為 BigDecimal 這一步使用Float.toString 保留數(shù)據(jù)精度。該方法不僅復(fù)雜，還生成了許多中間對(duì)象，因此效率自然不高。

一個(gè) workaround 方式是使用高效的解析庫(kù) Ryu ，并將其返回值改為 BigDecimal，減少中間對(duì)象的生成。這一優(yōu)化能夠?qū)⒕幋a速度提高約 5 倍。

無損壓縮

有損壓縮的一個(gè)缺點(diǎn)是泛用性較差，并且無法處理小數(shù)位較大的情況。

然而高冗余的浮點(diǎn)數(shù)據(jù)，在時(shí)序數(shù)據(jù)中比比皆是，比如某些機(jī)器監(jiān)控指標(biāo)，其值往往只在 0 ~ 1 之間波動(dòng)。

2015 年 Fackbook 發(fā)表了一篇關(guān)于內(nèi)存時(shí)序數(shù)據(jù)庫(kù)的 論文，其中提出了一種基于異或算法的浮點(diǎn)數(shù)壓縮算法。

Facebook 的研究人員在時(shí)序浮點(diǎn)數(shù)中發(fā)現(xiàn)個(gè)規(guī)律：

在同個(gè)時(shí)間序列中，大部分浮點(diǎn)數(shù)的占用的有效 bit 位是類似的，并且往往只有中間的一個(gè)連續(xù)區(qū)塊存儲(chǔ)著不同的數(shù)據(jù)。

因此他們想了個(gè)辦法提取并只保存這部分?jǐn)?shù)據(jù)：

1.將 float 轉(zhuǎn)換為 bits

2.計(jì)算兩個(gè)相鄰 bits 的異或值 xor

3.記錄 xor 的 leading-zero 與連續(xù)區(qū)塊長(zhǎng)度 block-size

4.記錄 1 bit 的標(biāo)記位 flag：

• 如果 leading-zero 與 block-size 與之前相同，記為 false
• 如果 leading-zero 與 block-size 與之前不同，記為 true

5.當(dāng) flag 為 true 時(shí)，記錄 leading-zero 與 block-size

6.記錄連續(xù)區(qū)塊 block 數(shù)據(jù)，然后進(jìn)入下個(gè)循環(huán)

由于大部分?jǐn)?shù)據(jù)的 leading-zero 與連續(xù)區(qū)塊長(zhǎng)度 block-size 均相等，往往只需要存儲(chǔ) flag 與 block 數(shù)據(jù)，因此大多數(shù)情況都能有效的減少數(shù)據(jù)存儲(chǔ)空間。

這個(gè)算法主要的難點(diǎn)是實(shí)現(xiàn)高效的 BitWriter，這個(gè)可以通過繼承 ByteArrayOutputStream 并增加一個(gè) long 類型的 buffer 實(shí)現(xiàn)。

abstract class BlockMeta<T extends BlockMeta<T>> {

    short leadingZero;
    short tailingZero;
    int blockSize;

    BlockMeta<T> withMeta(int leadingZero, int blockSize) {
        this.leadingZero = (short) leadingZero;
        this.tailingZero = (short) (length() - leadingZero - blockSize);
        this.blockSize = blockSize;
        return this;
    }

    abstract long value();
    abstract int length();

    boolean fallInSameBlock(BlockMeta<? extends BlockMeta<?>> block) {
        return block != null && block.leadingZero == leadingZero && block.tailingZero == tailingZero;
    }
}

static void encodeBlock(BitsWriter buffer, BlockMeta<?> block, BlockMeta<?> prevBlock) {
    if (block.value() == 0) {
        buffer.writeBit(false);
    } else {
        boolean ctrlBit;
        buffer.writeBit(true);
        buffer.writeBit(ctrlBit = ! block.fallInSameBlock(prevBlock));
        if (ctrlBit) {
            buffer.writeBits(block.leadingZero, 6);
            buffer.writeBits(block.blockSize, 7);
        }
        buffer.writeBits(block.value(), block.blockSize);
    }
}

static long decodeBlock(BitsReader reader, BlockMeta<?> blockMeta) {
    if (reader.readBit()) {
        boolean ctrlBit = reader.readBit();
        if (ctrlBit) {
            int leadingZero = (int) reader.readBits(6);
            int blockSize = (int) reader.readBits(7);
            blockMeta.withMeta(leadingZero, blockSize);
        }
        CodecException.malformedData(blockMeta == null);
        long value = reader.readBits(blockMeta.blockSize);
        return value << blockMeta.tailingZero;
    }
    return 0;
}

Facebook 聲稱，使用該算法壓縮 2 小時(shí)的時(shí)序數(shù)據(jù)，每個(gè)數(shù)據(jù)點(diǎn)僅僅需 1.37 byte：

A two-hour block allows us to achieve a compression ratio of
1.37 bytes per data point.

經(jīng)測(cè)試，該算法能夠?qū)⒎謺r(shí)數(shù)據(jù)壓縮為原來的 33%，壓縮率可達(dá) 60% 以上，效果顯著。

字符串

時(shí)序數(shù)據(jù)中的字符串大致可以分為兩類：

• 標(biāo)簽型 —— 數(shù)據(jù)字典
• 非標(biāo)簽型 —— Snappy

標(biāo)簽型

標(biāo)簽型字符串在時(shí)時(shí)序數(shù)據(jù)中更為常見，比如監(jiān)控?cái)?shù)據(jù)中的 IP 與 Metric 名稱便是此類數(shù)據(jù)。

該類數(shù)據(jù)通常作為查詢索引使用，其值也比較固定。因此通常使用數(shù)據(jù)字典進(jìn)行壓縮，將變長(zhǎng)字符串轉(zhuǎn)換為定長(zhǎng)的整型索引。

• 一方面減少了空間占用
• 另一方面有利于構(gòu)建高效的索引

當(dāng)轉(zhuǎn)換成整型后，還能進(jìn)一步的利用 BloomFilter 與 Bitmap 等數(shù)據(jù)結(jié)構(gòu)，進(jìn)一步提升查詢性能。

非標(biāo)簽型

非標(biāo)簽型的字符串較為少見，較少時(shí)序數(shù)據(jù)引擎支持該類型。

這類數(shù)據(jù)的值較為不可控，因此只能使用通用的壓縮算法進(jìn)行壓縮。

為了兼顧效率，通常情況下會(huì)使用 snappy 或 lz4，兩者不相伯仲。

通常情況下兩者的壓縮比較為接近，這方面 snappy 有微弱的優(yōu)勢(shì)。

不夠 lz4 提供了更為靈活的壓縮策略：

• 可調(diào)的壓縮等級(jí)，允許使用者自行調(diào)配速度與壓縮率
• 可以在壓縮速度與解壓速度上進(jìn)行權(quán)衡，當(dāng)解壓與壓縮機(jī)器性能不對(duì)稱時(shí)較為有用

以上就是基于Java實(shí)現(xiàn)簡(jiǎn)單的時(shí)序數(shù)據(jù)壓縮算法的詳細(xì)內(nèi)容，更多關(guān)于Java時(shí)序數(shù)據(jù)壓縮算法的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！