1
-- 分析查询：只查平均薪资
2
SELECT AVG(salary) FROM employees WHERE city = '北京';
3

4
-- 行存：需要读取所有列的数据
5
-- 100 列 × 1M 行 = 读取 100M 个值（假设每列 8B = 800MB）
6

7
-- 列存：只需读取 salary 和 city 两列
8
-- 2 列 × 1M 行 = 读取 2M 个值（= 16MB）
9
-- I/O 减少 50 倍！

1
# Parquet 文件元数据结构
2
class ParquetMetadata:
3
    version: int                    # 文件版本
4
    schema: SchemaDescriptor        # 列定义
5
    num_rows: int                   # 总行数
6
    row_groups: list[RowGroupMetadata]
7

8
class RowGroupMetadata:
9
    num_rows: int                   # 行数
10
    total_byte_size: int            # 总大小
11
    columns: list[ColumnChunkMetadata]
12

13
class ColumnChunkMetadata:
14
    file_offset: int                # 文件偏移
15
    total_compressed_size: int      # 压缩后大小
16
    total_uncompressed_size: int    # 压缩前大小
17
    num_values: int                 # 值数量
18
    encoding: list[str]             # 编码方式
19
    statistics: ColumnStatistics    # 统计信息
20

21
class ColumnStatistics:
22
    min_value: any                  # 最小值
23
    max_value: any                  # 最大值
24
    null_count: int                 # NULL 数量
25
    distinct_count: int             # 唯一值数量

1
# 行式执行（一次一行）
2
def row_oriented_execution(table):
3
    total = 0
4
    for row in table:
5
        if row.city == '北京':
6
            total += row.salary
7
    return total
8

9
# 向量化执行（一次一批）
10
def vectorized_execution(salary_column, city_column, batch_size=1024):
11
    total = 0
12
    for i in range(0, len(salary_column), batch_size):
13
        # 一次处理 1024 个值
14
        salary_batch = salary_column[i:i+batch_size]
15
        city_batch = city_column[i:i+batch_size]
16

17
        # 向量化过滤
18
        mask = city_batch == '北京'  # 一次比较 1024 个值
19

20
        # 向量化聚合
21
        total += salary_batch[mask].sum()
22

23
    return total

1
// SIMD 向量化聚合（AVX2）
2
// 一次处理 8 个 int32 值
3
void vectorized_sum_avx2(int32_t *values, int n, int32_t *result) {
4
    __m256i sum = _mm256_setzero_si256();  // 8 × 32bit = 256bit
5

6
    for (int i = 0; i < n; i += 8) {
7
        __m256i vec = _mm256_loadu_si256((__m256i*)(values + i));
8
        sum = _mm256_add_epi32(sum, vec);  // 一次加 8 个值
9
    }
10

11
    // 横向求和
12
    int32_t temp[8];
13
    _mm256_storeu_si256((__m256i*)temp, sum);
14
    *result = 0;
15
    for (int i = 0; i < 8; i++) *result += temp[i];
16
}

1
# Parquet 编码流程（以整数列为例）
2
def parquet_encode_int_column(values):
3
    # 1. 统计分析
4
    cardinality = len(set(values))
5
    is_sorted = all(values[i] <= values[i+1] for i in range(len(values)-1))
6

7
    # 2. 选择列级编码
8
    if cardinality < len(values) * 0.01:
9
        # 低基数：字典编码
10
        dict_page, indices = dictionary_encode(values)
11
        # 索引再用 RLE + 位打包
12
        encoded = rle_encode(bit_pack_encode(indices))
13
        return dict_page, encoded
14
    elif is_sorted:
15
        # 排序整数：Delta + 位打包
16
        delta = delta_encode(values)
17
        return bit_pack_encode(delta)
18
    else:
19
        # 通用：Plain 编码
20
        return plain_encode(values)
21

22
    # 3. 块级压缩（对编码后的字节流）
23
    # compressed = zstd_compress(encoded_bytes)

1
// Arrow 列式内存布局
2
struct ArrowArray {
3
    int64_t  length;          // 元素数量
4
    int64_t  null_count;      // NULL 数量
5
    int64_t  offset;          // 偏移量
6
    byte    *buffers[];       // 缓冲区数组
7
    // 对于 int32 列：
8
    //   buffers[0] = 有效性位图（1 bit/值）
9
    //   buffers[1] = 数据缓冲区（4 bytes/值，连续存储）
10
};
11

12
// 零拷贝：多个进程可 mmap 同一 Arrow 内存区域
13
// 无需序列化/反序列化，直接读取 buffers 中的数据
14
// Spark → Arrow → Python: 避免了 JVM 到 Python 的数据拷贝

1
# 数据湖读取流程（以 Iceberg 为例）
2
def iceberg_read(table, snapshot_id=None):
3
    # 1. 获取快照元数据
4
    snapshot = table.snapshots.get(snapshot_id or table.current_snapshot)
5

6
    # 2. 从 Manifest List 获取数据文件列表
7
    manifest_list = read_manifest_list(snapshot.manifest_list_path)
8

9
    # 3. 过滤：利用文件级统计跳过不匹配的文件
10
    matching_files = []
11
    for manifest in manifest_list:
12
        for file in manifest.data_files:
13
            if zone_map_matches(file.stats, query_filter):
14
                matching_files.append(file)
15

16
    # 4. 读取匹配的 Parquet 文件
17
    return read_parquet_files(matching_files, columns, filters)

1
# Zone Map 过滤
2
def zone_map_filter(query_filter, row_group_stats):
3
    """利用 Zone Map 跳过不相关的 Row Group"""
4
    skipped = 0
5
    read = 0
6

7
    for rg_stats in row_group_stats:
8
        # 检查过滤条件是否可能匹配
9
        if query_filter.column == 'salary':
10
            if query_filter.op == '>' and query_filter.value > rg_stats.salary_max:
11
                skipped += 1  # 所有值都小于阈值，跳过
12
                continue
13
            if query_filter.op == '<' and query_filter.value < rg_stats.salary_min:
14
                skipped += 1  # 所有值都大于阈值，跳过
15
                continue
16

17
        read += 1  # 可能匹配，需要读取
18

19
    print(f"跳过 {skipped} 个 Row Group，读取 {read} 个")
20
    return read
21

22
# 示例：SELECT * WHERE salary > 50000
23
# Row Group 1: salary min=30000, max=45000 → 跳过
24
# Row Group 2: salary min=40000, max=60000 → 读取
25
# Row Group 3: salary min=55000, max=80000 → 读取
26
# Row Group 4: salary min=20000, max=35000 → 跳过
27
# 4 个 Row Group 中跳过 2 个，I/O 减少 50%

1
# 使用 PyArrow 创建 Parquet 文件
2
import pyarrow as pa
3
import pyarrow.parquet as pq
4

5
# 创建表
6
table = pa.table({
7
    'id': range(1_000_000),
8
    'name': ['user_%d' % i for i in range(1_000_000)],
9
    'age': [20 + i % 60 for i in range(1_000_000)],
10
    'city': ['北京' if i % 3 == 0 else '上海' if i % 3 == 1 else '广州'
11
             for i in range(1_000_000)],
12
    'salary': [5000 + i * 10 for i in range(1_000_000)],
13
})
14

15
# 写入 Parquet 文件
16
pq.write_table(
17
    table,
18
    'employees.parquet',
19
    row_group_size=128 * 1024 * 1024,  # 128MB Row Group
20
    compression='zstd',                 # ZSTD 压缩
21
    compression_level=3,
22
    use_dictionary=True,                # 字典编码
23
    write_statistics=True,              # 写入统计信息
24
)
25

26
# 查看文件元数据
27
parquet_file = pq.ParquetFile('employees.parquet')
28
print(f"行数: {parquet_file.metadata.num_rows}")
29
print(f"Row Group 数: {parquet_file.metadata.num_row_groups}")
30
print(f"压缩后大小: {parquet_file.metadata.total_compressed_size / 1024 / 1024:.1f} MB")
31
print(f"压缩前大小: {parquet_file.metadata.total_uncompressed_size / 1024 / 1024:.1f} MB")

1
# 使用谓词下推读取 Parquet
2
# 只读取 salary > 50000 的行
3
table = pq.read_table(
4
    'employees.parquet',
5
    columns=['name', 'salary'],  # 列裁剪
6
    filters=[('salary', '>', 50000)],  # 谓词下推
7
)
8

9
# Spark SQL 中的谓词下推
10
# spark.read.parquet('employees.parquet')
11
#     .filter('salary > 50000')
12
#     .select('name', 'salary')

1
# 查看 Parquet 列统计信息
2
parquet_file = pq.ParquetFile('employees.parquet')
3

4
for i, rg in enumerate(parquet_file.metadata.row_groups):
5
    for j, col in enumerate(rg.columns):
6
        stats = col.statistics
7
        print(f"RG{i} Col{j}: min={stats.min}, max={stats.max}, "
8
              f"null_count={stats.null_count}, distinct={stats.distinct_count}")