Parquet-Coder

Patterns for efficient columnar data storage with Parquet.

Basic Operations

import pandas as pd import pyarrow as pa import pyarrow.parquet as pq

Write with compression

df.to_parquet('data.parquet', compression='snappy', index=False)

Common compression options:

- snappy: Fast, good compression (default)

- gzip: Slower, better compression

- zstd: Best balance of speed/compression

- None: No compression (fastest writes)

Read entire file

df = pd.read_parquet('data.parquet')

Read specific columns only (predicate pushdown)

df = pd.read_parquet('data.parquet', columns=['id', 'name', 'value'])

PyArrow for Large Files

Read as PyArrow Table (more memory efficient)

table = pq.read_table('data.parquet')

Convert to pandas when needed

df = table.to_pandas()

Filter while reading (row group filtering)

table = pq.read_table( 'data.parquet', filters=[ ('date', '>=', '2024-01-01'), ('status', '=', 'active') ] )

Read in batches for huge files

parquet_file = pq.ParquetFile('huge.parquet') for batch in parquet_file.iter_batches(batch_size=100_000): df_batch = batch.to_pandas() process(df_batch)

Partitioned Datasets

Write partitioned by columns

df.to_parquet( 'data/', partition_cols=['year', 'month'], compression='snappy' )

Creates: data/year=2024/month=01/part-0.parquet

Read partitioned dataset

df = pd.read_parquet('data/') # Reads all partitions

Read specific partitions only

df = pd.read_parquet('data/year=2024/')

With PyArrow dataset API (more control)

import pyarrow.dataset as ds

dataset = ds.dataset('data/', format='parquet', partitioning='hive')

Filter on partition columns (very fast)

table = dataset.to_table( filter=(ds.field('year') == 2024) & (ds.field('month') >= 6) )

Schema Definition

Explicit schema for consistency

schema = pa.schema([ ('id', pa.int64()), ('name', pa.string()), ('value', pa.float64()), ('date', pa.date32()), ('tags', pa.list_(pa.string())), ('metadata', pa.map_(pa.string(), pa.string())), ])

Write with schema

table = pa.Table.from_pandas(df, schema=schema) pq.write_table(table, 'data.parquet')

Read and validate schema

file_schema = pq.read_schema('data.parquet') assert file_schema.equals(schema), "Schema mismatch!"

Schema Evolution

def merge_schemas(old_schema: pa.Schema, new_schema: pa.Schema) -> pa.Schema: """Create unified schema from old and new.""" fields = {f.name: f for f in old_schema} for field in new_schema: if field.name not in fields: fields[field.name] = field elif fields[field.name].type != field.type: # Handle type conflicts (e.g., promote int to float) fields[field.name] = pa.field( field.name, promote_type(fields[field.name].type, field.type) ) return pa.schema(list(fields.values()))

def append_with_schema_evolution( existing_path: str, new_df: pd.DataFrame, output_path: str ) -> None: """Append data with automatic schema evolution.""" existing = pq.read_table(existing_path) new_table = pa.Table.from_pandas(new_df)

# Unify schemas
unified_schema = pa.unify_schemas([existing.schema, new_table.schema])

# Cast both to unified schema
existing = existing.cast(unified_schema)
new_table = new_table.cast(unified_schema)

# Concatenate and write
combined = pa.concat_tables([existing, new_table])
pq.write_table(combined, output_path)

Row Group Optimization

Control row group size (affects read performance)

pq.write_table( table, 'data.parquet', row_group_size=100_000, # Rows per group compression='snappy' )

Read metadata to see row groups

parquet_file = pq.ParquetFile('data.parquet') print(f"Num row groups: {parquet_file.metadata.num_row_groups}") print(f"Num rows: {parquet_file.metadata.num_rows}")

Read specific row groups

table = parquet_file.read_row_groups([0, 1]) # First two groups

Metadata and Statistics

Add custom metadata

custom_metadata = { b'created_by': b'etl_pipeline', b'version': b'1.0', b'source': b'api_export' } schema = table.schema.with_metadata(custom_metadata) table = table.cast(schema) pq.write_table(table, 'data.parquet')

Read metadata

parquet_file = pq.ParquetFile('data.parquet') print(parquet_file.schema.metadata)

Column statistics (min/max for filtering)

metadata = parquet_file.metadata for i in range(metadata.num_row_groups): rg = metadata.row_group(i) for j in range(rg.num_columns): col = rg.column(j) if col.statistics: print(f"{col.path_in_schema}: min={col.statistics.min}, max={col.statistics.max}")

Delta Lake Integration

from deltalake import DeltaTable, write_deltalake

Write as Delta table (versioned parquet)

write_deltalake('delta_table/', df, mode='overwrite')

Append data

write_deltalake('delta_table/', new_df, mode='append')

Read Delta table

dt = DeltaTable('delta_table/') df = dt.to_pandas()

Time travel

df_old = dt.load_version(0).to_pandas()

Compact small files

dt.optimize.compact()

Vacuum old versions

dt.vacuum(retention_hours=168) # Keep 7 days

Performance Tips

1. Use appropriate column types

schema = pa.schema([ ('category', pa.dictionary(pa.int8(), pa.string())), # For repeated strings ('count', pa.int32()), # Not int64 if values fit ])

2. Sort data before writing (improves predicate pushdown)

df = df.sort_values(['date', 'category']) df.to_parquet('sorted.parquet')

3. Use column selection

df = pd.read_parquet('data.parquet', columns=['needed', 'columns'])

4. Use filters for row pruning

df = pd.read_parquet('data/', filters=[('status', '==', 'active')])

5. Parallel reads

import pyarrow.dataset as ds dataset = ds.dataset('data/', format='parquet') table = dataset.to_table(use_threads=True)

Conversion Utilities

def csv_to_parquet( csv_path: str, parquet_path: str, chunk_size: int = 100_000 ) -> None: """Convert large CSV to Parquet efficiently.""" writer = None

for chunk in pd.read_csv(csv_path, chunksize=chunk_size):
    table = pa.Table.from_pandas(chunk)

    if writer is None:
        writer = pq.ParquetWriter(parquet_path, table.schema)

    writer.write_table(table)

writer.close()

def json_to_parquet(json_path: str, parquet_path: str) -> None: """Convert JSON lines to Parquet.""" df = pd.read_json(json_path, lines=True) df.to_parquet(parquet_path, compression='snappy')