Parquet 与 Arrow

按行还是按列？

CSV 文件或按行存储的数据库会把每条记录放在一起：先是成交 1 的所有字段，然后是成交 2，依此类推。而分析很少需要这样。要算一千万笔成交的平均价格，只需要 price 这一列，但按行布局会迫使读取者跨过其他所有字段。列式布局让每一列连续存放，于是查询只读它点名的列，相似的值挨在一起因而压缩得很好，对一列做运算也只需扫过一块紧凑的内存。这就是量化数据湖在磁盘上用 Parquet、在内存里用 Arrow 的原因。

# The same table as rows and as columns
rows = [{"symbol": "AAPL", "price": 184.25, "qty": 100},
        {"symbol": "MSFT", "price": 410.5, "qty": 50},
        {"symbol": "AAPL", "price": 184.5, "qty": 25}]

columns = {name: [row[name] for row in rows] for name in rows[0]}
print(columns)                                       # {'symbol': ['AAPL', 'MSFT', 'AAPL'], 'price': [184.25, 410.5, 184.5], 'qty': [100, 50, 25]}
print(sum(columns["price"]) / len(rows))             # 259.75   the average only touched one column

back = [{name: columns[name][i] for name in columns} for i in range(len(rows))]
print(back == rows)                                  # True  nothing is lost either way

Parquet：行组、统计信息与编码

Parquet 文件把表拆分成行组（比如每组一百万行）；在一个行组内，每列单独存成一个数据块，文件会记录每个数据块的最小值和最大值。一个查询 price 在 180 到 190 之间的请求，会先读取统计信息，跳过所有范围不可能包含这种价格的行组（谓词下推），并且只读取被请求的列（投影下推）。列还会被编码：像代码这样取值很少的列，会被存成一个小字典加每行一个很小的整数编码，然后再压缩。重复性强的行情数据文件往往比 CSV 小十倍。

# Dictionary encoding turns a few distinct values into a small dictionary plus integer codes
def dictionary_encode(values):
    positions, dictionary, codes = {}, [], []
    for value in values:
        if value not in positions:
            positions[value] = len(dictionary)
            dictionary.append(value)
        codes.append(positions[value])
    return dictionary, codes

symbols = ["AAPL", "MSFT", "AAPL", "AAPL", "MSFT", "NVDA"] * 1000
dictionary, codes = dictionary_encode(symbols)
print(dictionary, codes[:6])                         # ['AAPL', 'MSFT', 'NVDA'] [0, 1, 0, 0, 1, 2]
print([dictionary[c] for c in codes] == symbols)     # True  decoding restores the column
print(len(dictionary), "distinct values for", len(symbols), "rows")      # 3 distinct values for 6000 rows

# Statistics let a reader skip whole row groups without opening them
def groups_to_read(stats, low, high):
    return [i for i, (minimum, maximum) in enumerate(stats) if maximum >= low and minimum <= high]

stats = [(180.0, 183.5), (183.0, 186.0), (185.5, 190.0), (189.0, 195.0)]        # (min, max) of price per row group
print(groups_to_read(stats, 184.0, 185.0))           # [1]        only the second group can hold a price in this range
print(groups_to_read(stats, 189.5, 192.0))           # [2, 3]
print(groups_to_read(stats, 200.0, 210.0))           # []         no group needs to be read at all

Arrow、pandas 与少读数据

Apache Arrow 为内存定义了同样的列式布局，因此不同程序和语言之间传递表时无需转换：pandas、Polars、DuckDB 和 Spark 都能零拷贝地查看同一张 Arrow 表。实践中会使用 pandas.read_parquet(path, columns=["symbol", "price"], filters=[("price", ">", 180)])，它把列的选择和过滤下推给读取器，也可以用 pyarrow.parquet.read_table。按日期这样的列做分区写出（df.to_parquet(path, partition_cols=["date"])），这样查询某一天只需打开一个目录。行组要大到足以很好地压缩，又要小到让统计信息能跳过许多行组；并按过滤所用的列排序数据，让每个行组只覆盖一个狭窄的范围。

# Sorting by the filter column makes the row groups cover narrow, non-overlapping ranges
import random

rng = random.Random(7)
prices = [round(rng.uniform(100, 200), 2) for _ in range(10_000)]

def group_stats(values, size):
    groups = [values[i : i + size] for i in range(0, len(values), size)]
    return [(min(g), max(g)) for g in groups]

def groups_to_read(stats, low, high):
    return [i for i, (a, b) in enumerate(stats) if b >= low and a <= high]

unsorted_groups = len(groups_to_read(group_stats(prices, 1000), 150.0, 151.0))
sorted_groups = len(groups_to_read(group_stats(sorted(prices), 1000), 150.0, 151.0))
print(unsorted_groups, sorted_groups)                # 10 1   without sorting every group's range overlaps the query

• 列式存储把每一列放在一起：查询只读需要的列，相似的值压缩得好，对列做运算很快。
• Parquet 把数据分成带最小/最大统计信息的行组，并用字典编码重复值；读取器会跳过不需要的行组和列。
• Arrow 是 pandas、Polars 和 DuckDB 共用的内存版本；按日期分区，按过滤列排序，追加而不是修改。

下一节：时序数据库，为 tick 和指标等带时间戳的数据设计的存储。

实验：评分内容

• 用 to_columns 把行变成列25 分
• 用 to_rows 还原各行25 分
• 用 dictionary_encode 压缩重复值25 分
• 用 groups_to_read 跳过可跳过的部分25 分