Parquetファイルをざっくりと理解してみる

# Parquetファイルをざっくりと理解してみる

本記事は「 [TTDC Advent Calendar 2024](https://qiita.com/advent-calendar/2024/ttdc) 」 2 日目の記事です。

社内でも取り扱うことの多いparquetファイル。まだ触れたことのない方や中身をまり理解せずに使っている方（私）のために、Parquetの特徴や機能をhttps://parquet.apache.org/docs/ より読み解きざっくりと解説していきます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3763657/018f9bb5-18ac-163e-33db-7a3017783b1b.png)

# 目次
- Parquetファイルとは
- 列指向データ形式の特徴
- ファイルフォーマット
- パーティション
- Encoding
- データ型
- 圧縮方式
- 古いバージョン(ver.1.0)でのタイムスタンプの取り扱い
- Metadataを見てみる
- まとめ
- 参考

# Parquetファイルとは
Apache Parquet は、効率的なデータの保存と取得のために設計された、オープンソースの列指向データファイル形式です。複雑なデータを一括処理するための高性能な圧縮およびエンコード スキームを提供し、多くのプログラミング言語や分析ツールでサポートされています。

# 列指向データ形式の特徴
- クエリ（データ検索）速度が早い
  - 列ごとにメタデータが保存されており関係のない列の読み込みをスキップして関係のある列だけ読み込むことができるので、CSVのような行指向ファイルと比較してデータの検索を効率的に実行できます
  - 逆に行データへのアクセス速度はトレードオフで遅くなります。例えば行データをランダムにN行取得するような処理は遅くなります
- 圧縮効率がいい
  - 列方向に圧縮することができるので時系列データのように同じ値が連続するデータの場合は大幅に圧縮することができます
  - その他にも列に格納するデータの型や値の傾向に合わせて列単位で圧縮アルゴリズムを変更することで効率を高めることができます

# ファイルフォーマット
![image-1.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3763657/2820745d-d94e-2f18-c72c-30f0bc4d28fb.png)
https://parquet.apache.org/docs/file-format/

- ざっくり構造
  - ファイルの先頭からRowGroupが順番に並びます（RowGroup0, RowGroup1...）
  - RowGroupの中にColumnがあります
  - Columnの中にPageがあります
  - Pageの中にデータ（Values）があります
  - ファイルの最後にFileMetaDataがあります
    - ファイルの終端から先に読み込み、目的の列がファイルのどの位置にあるか特定して対象のデータのみを参照することができます
- ざっくりRowGroupとPageの機能
  - RowGroup
    - RowGroupが大きいほどColumnが大きくなり、より大きなシーケンシャルIOが可能になるため全体の読み書きは早くなります
    - RowGroupが大きいほどIOサイズが増えるのでメモリが必要になります
    - RowGroupサイズは大きめの512MB-1GBが推奨です
  - Page
    - 圧縮はPage単位で行うことができます。読み込み時は必要な部分だけ解凍してデータ更新時も必要な部分だけ圧縮することで効率よく圧縮解凍ができます
    - Pageサイズが小さいほどきめ細かい読み書きができます
    - Pageサイズが大きいほどスペースや処理のオーバーヘッドがなくなり全体容量は少なく、全体の読み書きのスピードは速くなります
    - ただしI/OチャンクはPageではなくあくまでColumnなので読み書きはColumn単位で圧縮の単位がPageになります
    - Pageサイズは8KBが推奨です
- ネストされたデータ構造のサポート
  - PageヘッダにあるRepititionレベルやDefinitionレベルはネストデータに対して使用するオプションです
  - 詳しくはApache Arrowの[Arrow and Parquet Part 2: Nested and Hierarchical Data using Structs and Lists](https://arrow.apache.org/blog/2022/10/08/arrow-parquet-encoding-part-2/)が参考になります
    
    ネストデータ例
    ```json
    {
        "time": 0.0,
        "speed": 12.5,
        "acceleration": {
            "x": 2.1,
            "y": 3.5,
            "z": -0.1
        }
    }
    ```
# パーティション
- ApacheArrowの[Partitioned Datasets (Multiple Files)](https://arrow.apache.org/docs/python/parquet.html#partitioned-datasets-multiple-files)にある通り、複数のparquetファイルから１つのデータセットを構成することができます
- パーティションを設定するとparquetの中の列をキーとして、その値をディレクトリ名にすることでファイルを分割して構成します
  ```txt
  dataset_name/
    year=2007/
      month=01/
        0.parq
        1.parq
        ...
      month=02/
        0.parq
        1.parq
        ...
      month=03/
      ...
    year=2008/
      month=01/
      ...
    ...
  ```
- Parquetの読み書きを行うApache Arrow、AWS Athena、Sparkなどのエンジンがこのディレクトリ構成を解釈して検索対象のファイルのみを開くことで効率よくクエリを実行できます

# Encoding
下記のエンコーディング方式に対応しています。詳しくは[encodings](https://parquet.apache.org/docs/file-format/data-pages/encodings/)で説明されています。
- PLAIN
- PLAIN_DICTIONARY
- BIT_PACKED
- RLE
- RLE_DICTIONARY
- BYTE_STREAM_SPLIT
- DELTA_BINARY_PACKED
- DELTA_BYTE_ARRAY
- DELTA_LENGTH_BYTE_ARRAY

# データ型
ApacheArrowの[physical-types](https://arrow.apache.org/docs/cpp/parquet.html#physical-types)によると下記のデータ型が存在します。
- Physical type
  - BOOLEAN
  - INT32
  - INT64
  - INT96
  - FLOAT
  - DOUBLE
  - BYTE_ARRAY
  - FIXED_LENGTH_BYTE_ARRAY
- Logical type
  - Physical typeとの組み合わせで下記のバリエーションがあります。
    | Logical type | Physical type                                        |
    |--------------|------------------------------------------------------|
    | NULL         | Any                                                  |
    | INT          | INT32                                                |
    | INT          | INT64                                                |
    | DECIMAL      | INT32 / INT64 / BYTE_ARRAY / FIXED_LENGTH_BYTE_ARRAY |
    | DATE         | INT32                                                |
    | TIME         | INT32                                                |
    | TIME         | INT64                                                |
    | TIMESTAMP    | INT64                                                |
    | STRING       | BYTE_ARRAY                                           |
    | LIST         | Any                                                  |
    | MAP          | Any                                                  |
    | FLOAT16      | FIXED_LENGTH_BYTE_ARRAY                              |


# 圧縮方式
ApacheArrowの[compression](https://arrow.apache.org/docs/cpp/parquet.html#compression)によると下記の圧縮方式が使用できます
- SNAPPY
- GZIP
- BROTLI
- LZ4
- ZSTD


# 古いバージョン(ver.1.0)でのタイムスタンプの取り扱い
- Apache Arrowエンジンを使用して古いバージョン1.0のParquetファイルを書き込む場合、ナノ秒('ns')はマイクロ秒('us')に変換されます
- より新しい Parquet 形式バージョン 2.6 を使用すると、ナノ秒単位のタイムスタンプをキャストせずに保存できますが、新しいバージョンをサポートせず、バージョン1.0を使用しているParquetリーダーがあるため注意が必要です
- pyarrowエンジンでは```coerce_timestamps```で解像度を指定することができます
- また解像度がさがることでデータが失われるのを防ぐために```use_deprecated_int96_timestamps=True```でINT96タイムスタンプを使用することができますが、これは現在非推奨です
- 詳しくはApacheArrowの[Storing timestamps](https://arrow.apache.org/docs/python/parquet.html#storing-timestamps)が参考になります

# Metadataを見てみる
![image-2.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3763657/28aea8e1-7810-4b20-7b87-4a689b61c96c.png)
https://parquet.apache.org/docs/file-format/metadata/

- 様々な情報があります。これらのmetadataはpyarrow.parquet.ParquetFile.metadataやparquet-tools inspectを使用して確認することができます
- pandas, numpyでランダムにデータを作製してparquetファイルを作った場合にどのようなmetadataになるか見てみました
- ```parquet-tools inspect output.parquet```は--detailオプションをつけることで更に詳細に見ることができます
- 結果
  - 圧縮率（Snappy圧縮の場合）
    - カウントアップしているタイムスタンプ型が19%で最大圧縮となりました
    - 次いでint64型が8%、float型はランダムと正規分布で値の分布を変えてみましたがどちらも-0%となりました
  - RowGroupサイズを4000行に指定したことで、全体10000行に対して3つのRowGroupが作成されました
  - PageSizeの設定がどこに効いているかはこの方法では確認できませんでした

環境
```
python 3.10.3

pandas==2.2.3
pyarrow==18.0.0
parquet-tools==0.2.16
```

コード
```python
import pandas as pd
import pyarrow.parquet as pq
import pprint
import numpy as np
from datetime import datetime, timedelta

# データフレームの行数を定義
num_rows = 10000

# measurement_time 列をカウントアップする値で生成 (ミリ秒単位)
measurement_time = [datetime.now() + timedelta(milliseconds=i*10) for i in range(num_rows)]

# 他の3つの列には適当に変化する値を生成
col_speed = (np.random.random(num_rows) * 100).astype(int)  # 0から100の範囲でランダムな値(int)
col_accel = np.random.random(num_rows)   # 0から1の範囲でランダムな値(float)
col_gas = np.random.normal(loc=50, scale=10, size=num_rows)  # 平均50、標準偏差10の正規分布に従うランダム値(float)

# データフレームの作成
df = pd.DataFrame({
    'measurement_time': measurement_time,
    'speed': col_speed,
    'accel': col_accel,
    'gas': col_gas,
})

print(df.head())
df.to_parquet('output.parquet',
              version = "2.6",      # default "2.6"
              row_group_size = 4000,   # default None(minimum of the Table size and 1024 * 1024.)
              data_page_size = 1024 * 1, # 1KB default None(1MB)
            )

# pyarrowの機能でmetadataを見る場合はここを使う
# parquet_file = pq.ParquetFile('output.parquet')
# pprint.pprint(parquet_file.metadata.to_dict())

# --detailオプションで更に詳細に見れる
!parquet-tools inspect output.parquet
```

output
```
            measurement_time  speed     accel        gas
0 2024-11-27 21:22:41.606305     79  0.948418  62.105864
1 2024-11-27 21:22:41.616305     70  0.564227  49.890950
2 2024-11-27 21:22:41.626305     82  0.068329  52.143745
3 2024-11-27 21:22:41.636305     94  0.196393  55.518347
4 2024-11-27 21:22:41.646305     44  0.847549  40.346390

############ file meta data ############
created_by: parquet-cpp-arrow version 18.0.0-SNAPSHOT
num_columns: 4
num_rows: 10000
num_row_groups: 3
format_version: 2.6
serialized_size: 3777


############ Columns ############
measurement_time
speed
accel
gas

############ Column(measurement_time) ############
name: measurement_time
path: measurement_time
max_definition_level: 1
max_repetition_level: 0
physical_type: INT64
logical_type: Timestamp(isAdjustedToUTC=false, timeUnit=nanoseconds, is_from_converted_type=false, force_set_converted_type=false)
converted_type (legacy): NONE
compression: SNAPPY (space_saved: 19%)

############ Column(speed) ############
name: speed
path: speed
max_definition_level: 1
max_repetition_level: 0
physical_type: INT64
logical_type: None
converted_type (legacy): NONE
compression: SNAPPY (space_saved: 8%)

############ Column(accel) ############
name: accel
path: accel
max_definition_level: 1
max_repetition_level: 0
physical_type: DOUBLE
logical_type: None
converted_type (legacy): NONE
compression: SNAPPY (space_saved: -0%)

############ Column(gas) ############
name: gas
path: gas
max_definition_level: 1
max_repetition_level: 0
physical_type: DOUBLE
logical_type: None
converted_type (legacy): NONE
compression: SNAPPY (space_saved: -0%)

```



# まとめ
今回はhttps://parquet.apache.org/docs/ を読み解きparquetファイルの中身をざっくりと理解することができました。まだまだ紹介できなかった機能がありますがもっと理解を深め情報発信できればと思います。

最後まで読んでいただきありがとうございました。本記事の内容に誤りなどあればコメントにてご教授お願いいたします。

# 参考
- https://parquet.apache.org/docs/
- https://arrow.apache.org/docs/python/parquet.html