第08课：Parquet与Avro格式

💾 Parquet与Avro格式

列式存储在分析场景下具有更高压缩率和更快查询，Parquet和Avro是两种最重要的开放数据格式。

🐍 Python实战：核心实战

# Parquet格式模拟：列式存储、压缩、谓词下推
import hashlib, time, random, json
from datetime import datetime

class ColumnChunk:
    def __init__(self, name, dtype):
        self.name = name; self.dtype = dtype; self.values = []
        self.stats = {'min':None,'max':None,'null_count':0,'distinct':0}
        self.uncompressed = 0; self.compressed = 0
    def add(self, v):
        self.values.append(v); self.uncompressed += len(str(v).encode())
    def compute_stats(self):
        non_null = [v for v in self.values if v is not None]
        self.stats['null_count'] = len(self.values)-len(non_null)
        if non_null:
            self.stats['min'] = min(non_null); self.stats['max'] = max(non_null); self.stats['distinct'] = len(set(non_null))
        self.compressed = int(self.uncompressed * 0.3)
    def can_skip(self, op, val):
        mn, mx = self.stats.get('min'), self.stats.get('max')
        if mn is None: return False
        if op == '>' and mx <= val: return True
        if op == '<' and mn >= val: return True
        return False

class RowGroup:
    def __init__(self): self.columns = {}; self.row_count = 0
    def add_column(self, name, dtype, values):
        chunk = ColumnChunk(name, dtype)
        for v in values: chunk.add(v)
        chunk.compute_stats(); self.columns[name] = chunk
        self.row_count = max(self.row_count, len(values))

class ParquetFile:
    def __init__(self, schema): self.schema = schema; self.row_groups = []
    def write_row_group(self, records):
        rg = RowGroup()
        for col, dtype in self.schema.items():
            rg.add_column(col, dtype, [r.get(col) for r in records])
        self.row_groups.append(rg)
    def read_column(self, col, predicates=None):
        result = []; skipped = 0
        for rg in self.row_groups:
            skip = False
            if predicates:
                for pc, (op, val) in predicates.items():
                    if pc in rg.columns and rg.columns[pc].can_skip(op, val): skip = True; break
            if skip: skipped += 1; continue
            if col in rg.columns: result.extend(rg.columns[col].values)
        return result, skipped
    def stats(self):
        total = sum(rg.row_count for rg in self.row_groups)
        comp = sum(sum(c.compressed for c in rg.columns.values()) for rg in self.row_groups)
        uncomp = sum(sum(c.uncompressed for c in rg.columns.values()) for rg in self.row_groups)
        return {'rows':total,'compressed':comp,'uncompressed':uncomp,'ratio':round(uncomp/comp,2) if comp else 0}

schema = {'id':'int','name':'string','amount':'float','region':'string'}
pf = ParquetFile(schema)
for rg_idx in range(5):
    records = [{'id':rg_idx*1000+i+1,'name':f'客户_{rg_idx*1000+i+1}','amount':round(random.uniform(100,10000),2),'region':random.choice(['华东','华南','华北','西南','东北'])} for i in range(1000)]
    pf.write_row_group(records)

s = pf.stats()
print(f"Parquet: {s['rows']}行, 压缩{1/s['ratio']:.1%}, 压缩比{s['ratio']}x")
amounts, _ = pf.read_column('amount')
print(f"列裁剪amount: {len(amounts)}值")
result, skipped = pf.read_column('amount', {'amount':('>',8000)})
print(f"谓词下推(>8000): {len(result)}值, 跳过{skipped}行组")
for col, chunk in pf.row_groups[0].columns.items():
    print(f"  {col}: [{chunk.stats['min']},{chunk.stats['max']}], distinct={chunk.stats['distinct']}, 压缩{chunk.uncompressed/chunk.compressed:.1f}x")
print("✅ 验证通过 - Parquet模拟运行正常")

📖 原理深入：现代数据存储架构

🏆 最佳实践：数据存储设计检查清单

🔬 深度案例：生产环境实战经验

📚 延伸阅读与参考

🤔 常见问题与解答