Parquet Content-Defined Chunking

TL;DR: Parquet Content-Defined Chunking (CDC) 은 이제 PyArrow 와 Pandas 에 제공되며, Hugging Face 의 Xet 스토리지 레이어와 같은 콘텐츠 주소형 저장 시스템에서 Parquet 파일의 효율적인 중복 제거를 가능하게 합니다. CDC 는 변경된 데이터만 업로드하거나 다운로드함으로써 데이터 전송 및 저장 비용을 크게 줄입니다. CDC 를 활성화하려면 use_content_defined_chunking 인자를 전달합니다:

import pandas as pd
import pyarrow.parquet as pq
df.to_parquet("hf://datasets/{user}/{repo}/path.parquet", use_content_defined_chunking=True)
...

• Introduction
• Data Preparation
• Different Use Cases for Parquet Deduplication
• Using Parquet CDC feature with Pandas
• References
• Conclusion

Apache Parquet 는 데이터 엔지니어링 커뮤니티에서 널리 사용되는 열 기반 저장 형식입니다.

현재 Hugging Face 는 약 21 PB 의 데이터셋을 호스팅하며, 그 중 Parquet 파일만으로도 4 PB 이상의 저장 공간을 차지하고 있습니다. 따라서 Parquet 저장 최적화는 매우 높은 우선순위를 가집니다. Hugging Face 는 콘텐츠 정의 분할 (Content-Defined Chunking) 을 활용하여 데이터 블록의 중복 제거를 효율적으로 수행함으로써 저장 비용을 줄이고 다운로드/업로드 속도를 개선하는 새로운 스토리지 레이어인 Xet 를 도입했습니다.

Xet 은 형식 무관형이지만, Parquet 의 레이아웃과 열 분할 (데이터 페이지) 기반 압축은 데이터의 소소한 변화에도 불구하고 완전히 다른 바이트 수준의 표현을 생성할 수 있어 중복 제거 성능이 최적화되지 않을 수 있습니다. 이를 해결하기 위해 유사한 데이터 간의 바이트 수준의 차이를 최소화하는 방식으로 Parquet 파일을 작성해야 하는데, 이것이 콘텐츠 정의 분할 (CDC) 이 등장한 이유입니다.

Hugging Face 의 Xet 스토리지 레이어와 함께 사용되는 새로운 Parquet CDC 기능의 성능 혜택을 살펴보겠습니다.

시범을 위해 OpenOrca 데이터셋의 관리 가능한 크기의 부분 집합을 사용하겠습니다.

import numpy as np
import pyarrow as pa
import pyarrow.compute as pc
...

id	system_prompt	question_length	question	response_length	response
0	cot.64099	You are an AI assistant that helps people find...	241	Consider the question. What is the euphrates l...	1663
...

pyarrow>=21.0.0 에서는 pyarrow 함수를 사용하여 Hugging Face URI 를 직접 읽기 및 쓰기 (Parquet 및 기타 파일 형식) 파일을 Hub 로 사용할 수 있습니다.

>>> pq.write_table(table, "hf://datasets/kszucs/pq/orca.parquet")
New Data Upload: 100%|███████████████████████████████████████████████| 96.1MB / 96.1MB, 48.0kB/s
Total Bytes: 96.1M
...

테이블이 완전히 업로드되었음을 확인할 수 있습니다 (총 바이트 == 총 전송량) 는 새로운 데이터로 Xet 스토리지 레이어가 아직 이를 알지 못하기 때문입니다. 이제 pyarrow 테이블로 다시 읽으십시오:

downloaded_table = pq.read_table("hf://datasets/kszucs/pq/orca.parquet")
assert downloaded_table.equals(table)

모든 pyarrow 함수는 파일 경로를 받기도 하지만 Hugging Face URI 를도 받습니다. 예를 들어 pyarrow datasets, CSV 함수, 증분 Parquet 작성기 또는 Parquet 메타데이터만 읽기:

pq.read_metadata("hf://datasets/kszucs/pq/orca.parquet")

<pyarrow._parquet.FileMetaData object at 0x16ebfa980>
created_by: parquet-cpp-arrow version 21.0.0-SNAPSHOT
num_columns: 6
...

콘텐츠 정의 분할 기능의 효과를 입증하기 위해 다음과 같은 경우 어떻게 수행되는지 시도해 보겠습니다:

• Re-uploading exact copies of the table
• Adding/removing columns from the table
• Changing column types in the table
• Appending new rows and concatenating tables
• Inserting / deleting rows in the table
• Change row-group size of the table
• Using Varying File-Level Splits

이러한 사용 사례는 단순해 보이지만, 전통적인 파일 시스템은 파일을 중복 제거하지 않아 데이터의 완전한 재업로드 및 재다운로드를 유발합니다. 반면, 콘텐츠 정의된 chunking 을 사용하는 시스템은 파일 내용이 동일함을 인식하고 불필요한 데이터 전송을 피할 수 있습니다.

>>> pq.write_table(table, "hf://datasets/kszucs/pq/orca-copy.parquet")
New Data Upload: | | 0.00B / 0.00B, 0.00B/s
Total Bytes: 96.1M
...

새로운 데이터 업로드가 없음을 알 수 있으며, 작업은 순간적으로 완료되었습니다. 이제 동일한 파일을 다른 저장소에 업로드할 때 어떤 일이 일어나는지 확인해 보겠습니다:

>>> pq.write_table(table, "hf://datasets/kszucs/pq-copy/orca-copy-again.parquet")
New Data Upload: | | 0.00B / 0.00B, 0.00B/s
Total Bytes: 96.1M
...

중복 제거가 저장소 간에도 작동하므로 업로드는 다시 순간적으로 완료되었습니다. 이는 Xet 저장 레이어의 핵심 기능으로, 효율적인 데이터 공유 및 협업을 가능하게 합니다. Hub 블로그 포스트 "From Chunks to Blocks: Accelerating Uploads and Downloads" 에서 세부 사항과 확장성 과제를 더 자세히 읽을 수 있습니다.

먼저 원래 테이블과 변경된 테이블을 로컬 parquet 파일로 작성하여 크기를 확인해 보겠습니다:

table_with_new_columns = table.add_column(
table.schema.get_field_index("response"),
"response_short",
...

!ls -lah /tmp/*.parquet

-rw-r--r-- 1 kszucs wheel 92M Jul 22 14:47 /tmp/original.parquet
-rw-r--r-- 1 kszucs wheel 92M Jul 22 14:47 /tmp/with-new-columns.parquet
-rw-r--r-- 1 kszucs wheel 67M Jul 22 14:47 /tmp/with-removed-columns.parquet

이제 Hugging Face 에 업로드하여 실제로 전송되는 데이터 양을 확인해 보겠습니다:

>>> pq.write_table(table_with_new_columns, "hf://datasets/kszucs/pq/orca-added-columns.parquet")
New Data Upload: 100%|███████████████████████████████████████████████| 575kB / 575kB, 288kB/s
Total Bytes: 96.6M
...

새로운 열과 파일의 푸터에 배치된 새로운 parquet 메타데이터 만이 업로드되었으며, 원래 데이터는 다시 전송되지 않았음을 알 수 있습니다. 이는 Xet 저장 레이어의 큰 장점이며, 전체 데이터셋을 다시 전송하지 않고도 새로운 열을 효율적으로 추가할 수 있게 합니다.

열 제거에도 동일하게 적용되며, 아래에서 확인할 수 있습니다:

>>> pq.write_table(table_with_removed_columns, "hf://datasets/kszucs/pq/orca-removed-columns.parquet")
New Data Upload: 100%|███████████████████████████████████████████████| 37.7kB / 37.7kB, 27.0kB/s
Total Bytes: 70.6M
...

업로드된 내용을 더 잘 이해하기 위해 중복 제거 추정 도구를 사용하여 두 parquet 파일의 차이를 시각화해 보겠습니다:

from de import visualize
visualize(table, table_with_new_columns, title="With New Columns", prefix="orca")

Compression	Vanilla Parquet
None
...
두 개의 새로운 열을 추가하면 unseen 데이터 페이지가 생성되어 전송되어야 함을 알 수 있습니다 (빨간색으로 강조), 나머지 데이터는 변경되지 않았으므로 다시 전송되지 않습니다 (초록색으로 강조). parquet 파일을 수정할 때마다 거의 항상 변경되는 푸터 메타데이터의 작은 빨간색 영역을 주목하세요. 중복 통계는 <deduped size> / <total size> = <dedup ratio> 를 보여줍니다.

비율이 작을수록 더 높은 중복 제거 성능을 의미합니다.

또한 열을 제거한 후 차이를 시각화해 보겠습니다:

visualize(table, table_with_removed_columns, title="With Removed Columns", prefix="orca")

Compression	Vanilla Parquet
None
...

전체 열을 제거하기 때문에 변경 사항만 푸터 메타데이터에서만 확인할 수 있으며, 나머지 열은 저장 계층에 이미 존재하므로 다시 전송되지 않습니다.

또 다른 일반적인 사용 사례는 테이블의 열 타입을 변경하는 것입니다. 예를 들어 저장 공간을 줄이거나 특정 쿼리를 위해 데이터를 최적화하기 위해. question_length
열을 int64
data type 으로 int32
타입으로 변경하고 전송되는 데이터 양을 확인해 보겠습니다:

# 먼저 큰 문자열 열을 제거하여 테이블을 더 작게 만듭니다.
# 차이를 더 잘 강조하기 위해
table_without_text = table_with_new_columns.drop(["question", "response"])
...

>>> pq.write_table(table_with_casted_column, "hf://datasets/kszucs/pq/orca-casted-column.parquet")
New Data Upload: 100%|███████████████████████████████████████████████| 181kB / 181kB, 113kB/s
Total Bytes: 1.80M
...

다시, 새로운 열과 업데이트된 parquet 메타데이터만 업로드되었음을 확인할 수 있습니다. 이제 중복 제거 히트맵을 시각화해 보겠습니다:

visualize(table_without_text, table_with_casted_column, title="With Casted Column", prefix="orca")

Compression	Vanilla Parquet
None
...

첫 번째 빨간 영역은 추가된 새로운 열을 나타내며, 두 번째 빨간 영역은 푸터에 업데이트된 메타데이터를 나타냅니다. 나머지 데이터는 변경되지 않고 다시 전송되지 않습니다.

우리는 원본 데이터셋의 다른 슬라이스를 테이블에 연결하여 새로운 행을 추가할 것입니다.

table = orca[:100_000]
next_10k_rows = orca[100_000:110_000]
table_with_appended_rows = pa.concat_tables([table, next_10k_rows])
...

이제 원래 데이터가 Xet 저장 계층에 이미 알려져 있으므로 새로운 행만 업로드되는지 확인해 보겠습니다:

>>> pq.write_table(table_with_appended_rows, "hf://datasets/kszucs/pq/orca-appended-rows.parquet")
New Data Upload: 100%|███████████████████████████████████████████████| 10.3MB / 10.3MB, 1.36MB/s
Total Bytes: 106M
...

visualize(table, table_with_appended_rows, title="With Appended Rows", prefix="orca")

Compression	Vanilla Parquet
None
...

각 열이 새로운 데이터를 얻으므로 여러 개의 빨간 영역을 확인할 수 있습니다. 이는 실제 parquet 파일 사양에서 각 열이 서로 다음으로 배치됨 (각 행 그룹 내) 때문입니다.

여기서 insertions 과 deletions 은 기존 행을 이동시켜 parquet 명칭의 데이터 페이지 또는 열 블록을 다르게 만들며, 각 데이터 페이지는 별도로 압축되므로 단일 행 삽입 또는 삭제만으로도 편집된 행부터 parquet 파일 끝까지 완전히 다른 바이트 수준의 표현으로 이어질 수 있습니다.

이 parquet 특화 문제는 Xet 저장 계층만으로 해결할 수 없으며, parquet 파일 자체가 삽입되거나 삭제된 행이 있어도 데이터 페이지 차이를 최소화하는 방식으로 작성되어야 합니다.

기존 메커니즘을 사용해보고 성능을 확인해 보겠습니다.

table = orca[:100_000]
# 두 곳에서 4k 줄 제거
table_with_deleted_rows = pa.concat_tables([
...

pq.write_table(table_with_inserted_rows, "hf://datasets/kszucs/pq/orca-inserted-rows.parquet")
New Data Upload: 100%|███████████████████████████████████████████████| 89.8MB / 89.8MB, 42.7kB/s
Total Bytes: 99.1M
...

pq.write_table(table_with_deleted_rows, "hf://datasets/kszucs/pq/orca-deleted-rows.parquet")
New Data Upload: 100%|███████████████████████████████████████████████| 78.2MB / 78.2MB, 46.5kB/s
Total Bytes: 92.2M
...

또한 차이를 확인하기 위해 두 경우를 시각화합니다:

visualize(table, table_with_deleted_rows, title="Deleted Rows", prefix="orca")
visualize(table, table_with_inserted_rows, title="Inserted Rows", prefix="orca")

Compression	Vanilla Parquet
None
...
Compression	Vanilla Parquet
---	---
None
...
중복 제거 성능이 크게 저하됨 (비율이 더 높음)을 알 수 있으며, 중복 제거 히트맵은 압축된 parquet 파일이 서로 매우 다르다는 것을 보여줍니다. 이는 삽입 및 삭제된 행이 기존 행을 이동시켜 편집된 행부터 parquet 파일의 끝까지 다른 데이터 페이지로 인해 발생합니다.

이를 해결하기 위해 pyarrow 의 새로운 기능인 content-defined chunking (CDC) 을 사용하여 parquet 파일을 작성할 수 있습니다. 이 기능은 Xet 스토리지 레이어가 데이터를 중복 제거하는 방식과 유사하게, 직렬화 또는 압축이 발생하기 전에 컬럼의 논리적 값에 따라 데이터 페이지로 일관되게 분할되도록 보장합니다.

이 기능을 활성화하려면 write_parquet 함수에 use_content_defined_chunking=True 를 전달해야 합니다:

import pyarrow.parquet as pq
pq.write_table(table, "hf://user/repo/filename.parquet", use_content_defined_chunking=True)

Pandas 도 새로운 기능을 지원합니다:

df.to_parquet("hf://user/repo/filename.parquet", use_content_defined_chunking=True)

CDC 기능을 사용하여 before/after 의 중복 제거 차이를 시각화해 보겠습니다:

visualize(table, table_with_deleted_rows, title="With Deleted Rows", prefix="orca", with_cdc=True)
visualize(table, table_with_inserted_rows, title="With Inserted Rows", prefix="orca", with_cdc=True)

Compression	Vanilla Parquet	CDC Parquet
None
...
Compression	Vanilla Parquet	CDC Parquet
---	---
None
...
훨씬 더 좋아 보입니다! proof of the pudding is in the eating 이므로, 실제로 content-defined chunking parquet 기능을 사용하여 테이블을 업로드하고 데이터 전송량을 확인해 보겠습니다.

중요한 점은 content-defined chunking 을 활성화하여 원래 테이블을 먼저 업로드해야 한다는 것입니다:

>>> pq.write_table(table, "hf://datasets/kszucs/pq/orca-cdc.parquet", use_content_defined_chunking=True)
New Data Upload: 100%|███████████████████████████████████████████████| 94.5MB / 94.5MB, 46.5kB/s
Total Bytes: 96.4M
...

>>> pq.write_table(
... table_with_inserted_rows,
... "hf://datasets/kszucs/pq/orca-inserted-rows-cdc.parquet",
...

>>> pq.write_table(
... table_with_deleted_rows,
... "hf://datasets/kszucs/pq/orca-deleted-rows-cdc.parquet",
...

업로드된 데이터는 이전보다 훨씬 작아지며, 위에서 언급한 히트맵과 같이 중복 제거 성능이 크게 향상됨을 보여줍니다.

중요한 점은 huggingface_hub.hf_hub_download() 와 datasets.load_dataset() 함수를 사용하여 다운로드할 때도 동일한 성능 혜택이 적용된다는 것입니다.