<jupyter_start><jupyter_text>Parquet Row Group首先我们来了解下 Parquet 中的一些重要概念 (https://github.com/apache/parquet-format):[EN]- Block (HDFS block): This means a block in HDFS and the meaning is unchanged for describing this file format. The file format is designed to work well on top of HDFS.- File: A HDFS file that must include the metadata for the file. It does not need to actually contain the data.- Row group: A logical horizontal partitioning of the data into rows. There is no physical structure that is guaranteed for a row group. A row group consists of a column chunk for each column in the dataset.- Column chunk: A chunk of the data for a particular column. They live in a particular row group and are guaranteed to be contiguous in the file.- Page: Column chunks are divided up into pages. A page is conceptually an indivisible unit (in terms of compression and encoding). There can be multiple page types which are interleaved in a column chunk.[CN]考虑一个 DataFrame events, 有 event_id: string(32) 和 event_time: datetime64 两列. 一共 1,000,000 (1M) 行.- Block: HDFS 分布式文件系统中的数据块, 一般是 128 MB 一个. 也有用 64 MB 或是 256 MB 的.- File: 一个具体的文件. 里面包含了 metadata.- Row group: 一个文件内部从逻辑上被分为了很多个 Row Group, 每个 Row Group 的起始位置和结束位置的指针是保存在 File metadata 中的. 一个 Row Group 会包含很多 "行" (虽然数据不是按行存储的) 的数据. 比如我们可以说一个 Parquet 文件有 100 个 Row Group, 每个 Row Group 有 10,000 行数据.- Column chunk: 一个 Row Group 中会有很多 Column. Column chunk 内的数据是连续存储的. 比如从 0 ~ 1,000,000,000 bytes 是 event_id, 1,000,000,001 ~ 1,005,000,000 是 event_time.Row Group 对用列式存储进行分析非常有帮助. 我们考虑一个例子: 我们需要读取一个大文件, 其中有个 column 是叫 event_time, 我们只需要 event_time between (start, end) 的数据. 有了 Row Group, 你可以通过 File metadata 直接定位到每个 Row Group 的起始点, 并读取 Row Group 的 metadata, 里面就有这个 Row Group 的 event_time 最大最小值. 这样可以避免扫描许多 Row Group, 从而提高查询速度.Ref:- https://arrow.apache.org/docs/python/parquet.htmlfiner-grained-reading-and-writing<jupyter_code>%pip install pathlib_mate
%pip install sfm
import numpy as np
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.compute as pc
from datetime import datetime
from pathlib_mate import Path
from sfm.timer import DateTimeTimer
n_rows = 1_000_000

def gen_data(n_rows):
    event_type_list = ["event_type_1", "event_type_2", "event_type_3"]
    data_table = pa.table({
        "event_id": np.arange(n_rows),
        "event_type": pa.DictionaryArray.from_arrays(
            np.random.randint(low=0, high=len(event_type_list), size=n_rows), 
            event_type_list,
        ),
        "event_time": pd.date_range("2000-01-01", periods=n_rows, freq="1s")
    })
    return data_table

with DateTimeTimer():
    data_table = gen_data(n_rows)
data_table.take([0, 1, 2]).to_pandas()
data_table.take([n_rows - 3, n_rows - 2, n_rows - 1]).to_pandas()<jupyter_output><empty_output><jupyter_text>Optimize Parquet file Using Row Group用 Row Group 对 查询进行优化的原理是, 如果你有 1GB 的数据, 而你的查询结果可能只涉及到 10MB 的数据, 那么 Row Group 对于每个 Column 的统计信息就可以帮助跳过哪些不包含目标数据的 Row Group. 很有可能你只需要扫描不到 100MB 数据就可以得到结果了. 而 IO 相比基于内存向量的查询要慢几个数量级, 所以最后扫描的速度相比把整个数据集读到内存要快得多的多.用 Row Group 进行优化, 建议按照这个流程:1. 决定哪几个 Column 你会进行 Range Query, 通常是 timestamp 列. 选择最重要的 Range Query Column.2. 在写入磁盘之前, 对 Range Query Column 进行排序.知识点:- ``row_group_size`` 的默认值为 64 * 1024 * 1024 = 64M- ``pyarrow.parquet.ParquetFile.num_row_groups`` 属性能获得 parquet 文件的 row group 总数- ``pyarrow.parquet.ParquetFile.metadata.row_group(row_group_id: int)`` 方法能获得具体某个 row group 的 metadata- ``pyarrow.parquet.ParquetFile.metadata.row_group(row_group_id: int).column(column_id: int)`` 方法能获得具体某个 row group 中某个 column 的 metadata- ``pyarrow.parquet.ParquetFile.metadata.row_group(row_group_id: int).column(column_id: int).statistics.min / max`` 属性能获得具体某个 row group 中某个 column 的统计最大值, 最小值.Ref:- https://arrow.apache.org/docs/python/generated/pyarrow.parquet.write_table.html<jupyter_code>pq_file = Path("events_with_row_group.parquet")
pq.write_table(data_table, pq_file.abspath, row_group_size=int(n_rows / 1000))
pq_file.size_in_text
start = datetime(2000, 1, 5)
end = datetime(2000, 1, 6)

with DateTimeTimer():
    parquet_file = pq.ParquetFile(pq_file.abspath)

    tables = list()
    for rg_id in range(parquet_file.num_row_groups):
        rg_meta = parquet_file.metadata.row_group(rg_id)
        event_time_col = rg_meta.column(2)
        if not ((event_time_col.statistics.max < start) or (event_time_col.statistics.min > end)):
            t = parquet_file.read_row_group(rg_id)
            t_filtered = pc.filter(
                t,
                pc.and_(
                    pc.greater_equal(t["event_time"], start),
                    pc.less(t["event_time"], end),
                ),
            )
            tables.append(t_filtered)

    table_filtered = pa.concat_tables(tables)
print(table_filtered.shape)
print("min event_time is:", pc.min(table_filtered["event_time"]))
print("max event_time is:", pc.max(table_filtered["event_time"]))
with DateTimeTimer():
    t = pq.read_table(pq_file.abspath)
    t_filtered = pc.filter(
        t,
        pc.and_(
            pc.greater_equal(t["event_time"], start),
            pc.less(t["event_time"], end),
        ),
    )
print(table_filtered.shape)
print("min event_time is:", pc.min(table_filtered["event_time"]))
print("max event_time is:", pc.max(table_filtered["event_time"]))<jupyter_output>(86400, 3)
min event_time is: 2000-01-05 00:00:00
max event_time is: 2000-01-05 23:59:59
{'fraction_non_alphanumeric': 0.09990697674418604, 'fraction_numerical': 0.02902325581395349, 'mean_word_length': 4.134670487106018, 'pattern_counts': {'":': 3, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 3, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '205653', 'license': 'MIT', 'n_tokens_mistral': 2257, 'n_tokens_neox': 2177, 'n_words': 536, 'path': '01-pyarrow-examples/91-parquet-Format/02-parquet-row-group.ipynb', 'repo_name': 'MacHu-GWU/learn_big_data_on_aws-project', 'size': '5375'}