Data Engineering Topic
Apache Parquet & Apache Arrow: Modern Data Formats
Overview & Evolution
The Problem with Traditional Formats
text
Traditional Formats Issues: ┌─────────────────┬─────────────────────────────────────┐ │ Format │ Issues │ ├─────────────────┼─────────────────────────────────────┤ │ CSV/TSV │ • No schema │ │ │ • Poor compression │ │ │ • Slow parsing │ │ │ • No column types │ ├─────────────────┼─────────────────────────────────────┤ │ JSON │ • Verbose │ │ │ • Slow parsing │ │ │ • Inefficient storage │ │ │ • Schema ambiguity │ ├─────────────────┼─────────────────────────────────────┤ │ Avro │ • Row-based (slow for analytics) │ │ │ • Requires schema │ │ │ • Java-centric │ └─────────────────┴─────────────────────────────────────┘
Enter Parquet & Arrow
text
Modern Analytics Stack: ┌─────────────────────────────────────────────────────┐ │ Application Layer │ ├─────────────────────────────────────────────────────┤ │ Arrow (In-memory) - Zero-copy sharing │ │ • Columnar memory format │ │ • Language-agnostic │ │ • SIMD optimized │ ├─────────────────────────────────────────────────────┤ │ Parquet (On-disk) - Efficient storage │ │ • Columnar storage │ │ • Compression │ │ • Predicate pushdown │ └─────────────────────────────────────────────────────┘
Apache Parquet: Columnar Storage Format
Parquet Architecture
text
┌─────────────────────────────────────────────────────┐ │ Parquet File Structure │ ├─────────────────────────────────────────────────────┤ │ File Metadata │ │ • Schema │ │ • Row groups │ │ • Column chunks │ │ • Page statistics │ ├─────────────────────────────────────────────────────┤ │ Row Group 1 (128MB) │ │ ┌─────────────┬─────────────┬─────────────┐ │ │ │ Column Chunk│ Column Chunk│ Column Chunk│ │ │ │ (Col1) │ (Col2) │ (Col3) │ │ │ └─────────────┴─────────────┴─────────────┘ │ │ Pages Pages Pages Pages │ │ ┌──┐ ┌──┐ ┌──┐ ┌──┐ │ │ │██│ │██│ │██│ │██│ │ │ └──┘ └──┘ └──┘ └──┘ │ ├─────────────────────────────────────────────────────┤ │ Row Group 2 │ │ ┌─────────────┬─────────────┬─────────────┐ │ │ │ Column Chunk│ Column Chunk│ Column Chunk│ │ │ │ (Col1) │ (Col2) │ (Col3) │ │ │ └─────────────┴─────────────┴─────────────┘ │ └─────────────────────────────────────────────────────┘
Key Features of Parquet
-
Columnar Storage
python
# Traditional row-based storage (CSV/JSON):
# Row1: [id:1, name:"Alice", age:30, salary:50000]
# Row2: [id:2, name:"Bob", age:25, salary:60000]
# Row3: [id:3, name:"Charlie", age:35, salary:70000]
# Parquet columnar storage:
# Column1 (id): [1, 2, 3]
# Column2 (name): ["Alice", "Bob", "Charlie"]
# Column3 (age): [30, 25, 35]
# Column4 (salary): [50000, 60000, 70000]
# Benefits for analytics:
# - Read only needed columns (projection pushdown)
# - Better compression (similar values together)
# - Vectorized processing
-
Nested Data Support
python
# Complex nested structures
import pyarrow as pa
import pyarrow.parquet as pq
# Define schema with nested types
schema = pa.schema([
pa.field('id', pa.int64()),
pa.field('name', pa.string()),
pa.field('address', pa.struct([
pa.field('street', pa.string()),
pa.field('city', pa.string()),
pa.field('zip', pa.string())
])),
pa.field('tags', pa.list_(pa.string())),
pa.field('metadata', pa.map_(pa.string(), pa.string()))
])
# Create data
data = [
{
'id': 1,
'name': 'Alice',
'address': {'street': '123 Main', 'city': 'NYC', 'zip': '10001'},
'tags': ['premium', 'active'],
'metadata': {'source': 'web', 'campaign': 'spring2024'}
}
]
table = pa.table(data, schema=schema)
pq.write_table(table, 'complex_data.parquet')
# Query nested data efficiently
# Parquet stores nested data in columnar format too!
-
Predicate Pushdown & Statistics
python
import pyarrow.parquet as pq
import pyarrow as pa
import numpy as np
# Create sample data with statistics
data = pa.table({
'id': pa.array(range(1_000_000)),
'value': pa.array(np.random.randn(1_000_000)),
'category': pa.array(['A', 'B', 'C', 'D'] * 250_000)
})
# Write with statistics
pq.write_table(
data,
'data.parquet',
row_group_size=100_000, # 100K rows per row group
compression='snappy',
write_statistics=True, # Store min/max/null counts
dictionary_columns=['category'] # Dictionary encoding
)
# Reading with predicate pushdown
# Parquet reads ONLY necessary row groups using statistics
result = pq.read_table(
'data.parquet',
columns=['id', 'value'], # Projection pushdown
filters=[ # Predicate pushdown
('category', '=', 'A'),
('value', '>', 1.0)
]
)
# Without predicate pushdown: Read 1M rows, filter
# With predicate pushdown: Read ~250K rows (only category A)
# Even better: May read only relevant row groups using min/max
Parquet File Internals
python
import pyarrow.parquet as pq
import json
# Inspect Parquet file metadata
parquet_file = pq.ParquetFile('data.parquet')
# File metadata
metadata = parquet_file.metadata
print(f"Number of rows: {metadata.num_rows}")
print(f"Number of row groups: {metadata.num_row_groups}")
print(f"Created by: {metadata.created_by}")
print(f"Schema: {metadata.schema}")
# Row group details
for i in range(metadata.num_row_groups):
row_group = metadata.row_group(i)
print(f"\nRow Group {i}:")
print(f" Total bytes: {row_group.total_byte_size}")
print(f" Number of rows: {row_group.num_rows}")
# Column chunk details
for j in range(row_group.num_columns):
col = row_group.column(j)
print(f" Column {j} ({col.path_in_schema}):")
print(f" Compressed size: {col.total_compressed_size}")
print(f" Uncompressed size: {col.total_uncompressed_size}")
# Statistics
stats = col.statistics
if stats is not None:
print(f" Min: {stats.min}")
print(f" Max: {stats.max}")
print(f" Null count: {stats.null_count}")
# Schema details
schema = parquet_file.schema_arrow
print("\nSchema details:")
for field in schema:
print(f" {field.name}: {field.type}")
# Physical storage info
print("\nPhysical storage:")
print(f"File size: {parquet_file.size} bytes")
print(f"Compression: {metadata.row_group(0).column(0).compression}")
Parquet Compression & Encoding
python
import pyarrow as pa
import pyarrow.parquet as pq
import pandas as pd
import numpy as np
# Test different compression algorithms
data_size = 1_000_000
df = pd.DataFrame({
'id': range(data_size),
'value': np.random.randn(data_size),
'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], data_size),
'description': ['Product_' + str(i) for i in range(data_size)]
})
table = pa.Table.from_pandas(df)
compression_methods = ['none', 'snappy', 'gzip', 'brotli', 'lz4', 'zstd']
results = []
for compression in compression_methods:
# Write with different compression
pq.write_table(
table,
f'test_{compression}.parquet',
compression=compression,
use_dictionary=True, # Dictionary encoding for categorical
compression_level=9 if compression == 'zstd' else None
)
# Read back and measure
import os
file_size = os.path.getsize(f'test_{compression}.parquet')
# Read performance
import time
start = time.time()
result = pq.read_table(f'test_{compression}.parquet')
read_time = time.time() - start
results.append({
'compression': compression,
'file_size_mb': file_size / (1024 * 1024),
'compression_ratio': (table.nbytes / file_size),
'read_time_ms': read_time * 1000
})
# Results analysis
results_df = pd.DataFrame(results)
print("\nCompression Comparison:")
print(results_df.to_string())
"""
Typical Results:
┌───────────┬─────────────┬──────────────────┬─────────────┐
│ Compression │ File Size MB │ Compression Ratio │ Read Time ms │
├───────────┼─────────────┼──────────────────┼─────────────┤
│ none │ 50.2 │ 1.0 │ 120 │
│ snappy │ 18.7 │ 2.7 │ 150 │
│ gzip │ 12.3 │ 4.1 │ 220 │
│ brotli │ 10.8 │ 4.6 │ 250 │
│ lz4 │ 20.1 │ 2.5 │ 140 │
│ zstd │ 11.5 │ 4.4 │ 180 │
└───────────┴─────────────┴──────────────────┴─────────────┘
Recommendations:
- Snappy: Fast read/write, good compression
- Zstd: Best compression/read speed balance
- Gzip: Good compression, slower reads
- None: Fastest, no compression
"""
Advanced Parquet Features
-
Partitioned Datasets
python
import pyarrow as pa
import pyarrow.parquet as pq
import pyarrow.dataset as ds
import pandas as pd
import os
# Create partitioned dataset
data = pd.DataFrame({
'date': pd.date_range('2024-01-01', '2024-01-10').repeat(100),
'region': ['US', 'EU', 'APAC'] * 333 + ['US'], # 1000 rows
'product': ['A', 'B', 'C', 'D', 'E'] * 200,
'sales': np.random.randint(100, 1000, 1000),
'cost': np.random.randint(50, 500, 1000)
})
# Add derived columns for partitioning
data['year'] = data['date'].dt.year
data['month'] = data['date'].dt.month
data['day'] = data['date'].dt.day
# Write as partitioned dataset
table = pa.Table.from_pandas(data)
pq.write_to_dataset(
table,
root_path='sales_partitioned',
partition_cols=['year', 'month', 'day', 'region'],
compression='snappy',
use_dictionary=['product']
)
# Directory structure created:
# sales_partitioned/
# ├── year=2024/
# │ ├── month=1/
# │ │ ├── day=1/
# │ │ │ ├── region=US/
# │ │ │ │ └── part-0.parquet
# │ │ │ ├── region=EU/
# │ │ │ │ └── part-0.parquet
# │ │ │ └── region=APAC/
# │ │ │ └── part-0.parquet
# │ │ ├── day=2/
# │ │ └── ...
# Query partitioned data efficiently
dataset = ds.dataset(
'sales_partitioned',
format='parquet',
partitioning=ds.partitioning(
pa.schema([
pa.field('year', pa.int64()),
pa.field('month', pa.int64()),
pa.field('day', pa.int64()),
pa.field('region', pa.string())
])
)
)
# Filter using partition columns (pruning)
result = dataset.to_table(
filter=(
(ds.field('year') == 2024) &
(ds.field('month') == 1) &
(ds.field('day').isin([1, 2, 3])) &
(ds.field('region') == 'US')
),
columns=['product', 'sales', 'cost'] # Projection pushdown
)
# Only reads partitions: year=2024/month=1/day=1,2,3/region=US
# Skips all other partitions entirely!
-
Delta Layout & Sorting
python
import pyarrow as pa
import pyarrow.parquet as pq
import numpy as np
# Create data with natural sorting
data = {
'customer_id': np.repeat(range(1000), 10), # Each customer 10 times
'transaction_id': range(10000),
'amount': np.random.randn(10000) * 100 + 500,
'category': np.random.choice(['A', 'B', 'C', 'D'], 10000)
}
# Sort by customer_id for better compression
df = pd.DataFrame(data).sort_values('customer_id')
table = pa.Table.from_pandas(df)
# Write with sorting benefits
pq.write_table(
table,
'sorted_data.parquet',
row_group_size=1000, # 1000 rows per group
sorting_columns=['customer_id'], # Data is pre-sorted
use_dictionary=['category'],
compression='zstd'
)
# Benefits:
# 1. Better compression: Similar customer_id values together
# 2. Faster filtering: Row groups have min/max customer_id
# 3. Efficient joins: Co-located data
# Z-order (multi-dimensional sorting)
# Sort by (customer_id, category) for multi-dimensional queries
df_multi = df.sort_values(['customer_id', 'category'])
table_multi = pa.Table.from_pandas(df_multi)
pq.write_table(
table_multi,
'zorder_data.parquet',
sorting_columns=['customer_id', 'category']
)
# Now queries on (customer_id, category) are efficient
# Example: WHERE customer_id BETWEEN 100 AND 200 AND category = 'A'
-
Schema Evolution
python
import pyarrow as pa
import pyarrow.parquet as pq
from datetime import datetime
# Initial schema v1
schema_v1 = pa.schema([
pa.field('id', pa.int64()),
pa.field('name', pa.string()),
pa.field('age', pa.int32())
])
data_v1 = [
{'id': 1, 'name': 'Alice', 'age': 30},
{'id': 2, 'name': 'Bob', 'age': 25}
]
table_v1 = pa.table(data_v1, schema=schema_v1)
pq.write_table(table_v1, 'data_v1.parquet')
# Schema evolution v2: Add new column, change type
schema_v2 = pa.schema([
pa.field('id', pa.int64()),
pa.field('name', pa.string()),
pa.field('age', pa.float64()), # Changed from int32 to float64
pa.field('email', pa.string()) # New field
])
data_v2 = [
{'id': 3, 'name': 'Charlie', 'age': 35.5, 'email': 'charlie@email.com'},
{'id': 4, 'name': 'Diana', 'age': 28.0, 'email': 'diana@email.com'}
]
table_v2 = pa.table(data_v2, schema=schema_v2)
pq.write_table(table_v2, 'data_v2.parquet')
# Reading with schema evolution
# Option 1: Read each file separately
table1 = pq.read_table('data_v1.parquet')
table2 = pq.read_table('data_v2.parquet')
# Option 2: Use dataset with schema unification
dataset = ds.dataset(['data_v1.parquet', 'data_v2.parquet'])
combined = dataset.to_table()
# Schema is unified:
# - age: float64 (promoted from int32)
# - email: string (nullable, null for v1 data)
print("Unified schema:")
print(combined.schema)
print("\nData:")
print(combined.to_pandas())
# Writing with schema evolution options
pq.write_table(
combined,
'merged.parquet',
version='2.6', # Parquet format version
data_page_version='2.0',
store_schema=True # Store Arrow schema for better compatibility
)
Apache Arrow: In-Memory Columnar Format
Arrow Architecture
text
┌─────────────────────────────────────────────────────┐ │ Arrow Memory Layout │ ├─────────────────────────────────────────────────────┤ │ Array (Column) │ │ ┌─────────────┬─────────────┬─────────────┐ │ │ │ Validity │ Data │ Buffers │ │ │ │ Bitmap │ Buffer │ (offsets) │ │ │ │ (nulls) │ (values) │ │ │ │ └─────────────┴─────────────┴─────────────┘ │ │ Fixed-size or variable-size │ ├─────────────────────────────────────────────────────┤ │ Record Batch (Row Group) │ │ ┌─────────────┬─────────────┬─────────────┐ │ │ │ Array1 │ Array2 │ Array3 │ │ │ │ (Col1) │ (Col2) │ (Col3) │ │ │ └─────────────┴─────────────┴─────────────┘ │ │ Same number of rows across arrays │ ├─────────────────────────────────────────────────────┤ │ Table (Collection of Record Batches) │ │ ┌─────────────────────────────────────────────┐ │ │ │ Record Batch 1 │ │ │ ├─────────────────────────────────────────────┤ │ │ │ Record Batch 2 │ │ │ ├─────────────────────────────────────────────┤ │ │ │ ... │ │ │ └─────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────┘
Key Features of Arrow
-
Zero-Copy Data Sharing
python
import pyarrow as pa
import numpy as np
import pandas as pd
# Create Arrow array from numpy (zero-copy)
np_array = np.array([1, 2, 3, 4, 5], dtype=np.int64)
arrow_array = pa.array(np_array) # Zero-copy: shares memory
# Verify zero-copy
print(f"NumPy array memory id: {np_array.__array_interface__['data'][0]}")
print(f"Arrow array buffer address: {arrow_array.buffers()[1].address}")
# Create pandas DataFrame from Arrow (zero-copy)
table = pa.table({
'id': [1, 2, 3],
'value': [10.5, 20.3, 30.7]
})
# Convert to pandas with zero-copy for numeric columns
df = table.to_pandas(zero_copy_only=True)
# For string columns, might need copy
# Arrow to NumPy (zero-copy for numeric types)
arrow_array = pa.array([1.5, 2.5, 3.5], type=pa.float64())
np_array = arrow_array.to_numpy(zero_copy_only=True)
# Cross-language zero-copy (C++ ↔ Python ↔ R ↔ Java)
# Same memory layout across languages
-
SIMD Optimizations
python
import pyarrow.compute as pc
import pyarrow as pa
import numpy as np
import time
# Create large dataset
size = 10_000_000
data = pa.array(np.random.randn(size), type=pa.float64())
# Arrow compute uses SIMD optimizations
start = time.time()
result = pc.sum(data) # SIMD-optimized summation
arrow_time = time.time() - start
# Compare with NumPy
np_data = data.to_numpy()
start = time.time()
np_result = np.sum(np_data)
numpy_time = time.time() - start
print(f"Arrow SIMD sum: {arrow_time:.4f}s")
print(f"NumPy sum: {numpy_time:.4f}s")
print(f"Speedup: {numpy_time/arrow_time:.2f}x")
# More SIMD-optimized operations
operations = {
'sum': pc.sum,
'mean': pc.mean,
'stddev': pc.stddev,
'min': pc.min,
'max': pc.max,
'add': lambda x: pc.add(x, pa.scalar(10.0)),
'multiply': lambda x: pc.multiply(x, pa.scalar(2.0))
}
for name, op in operations.items():
start = time.time()
result = op(data)
elapsed = time.time() - start
print(f"{name}: {elapsed:.6f}s")
-
Complex Data Types
python
import pyarrow as pa
# Struct types
struct_type = pa.struct([
pa.field('x', pa.float64()),
pa.field('y', pa.float64()),
pa.field('z', pa.float64())
])
struct_data = [
{'x': 1.0, 'y': 2.0, 'z': 3.0},
{'x': 4.0, 'y': 5.0, 'z': 6.0},
None, # Null struct
{'x': 7.0, 'y': 8.0, 'z': 9.0}
]
struct_array = pa.array(struct_data, type=struct_type)
# List types
list_type = pa.list_(pa.int64())
list_data = [
[1, 2, 3],
[4, 5],
None,
[6, 7, 8, 9]
]
list_array = pa.array(list_data, type=list_type)
# Fixed-size list
fixed_list_type = pa.list_(pa.int64(), list_size=3)
fixed_list_data = [
[1, 2, 3],
[4, 5, 6],
[7, 8, 9]
]
fixed_list_array = pa.array(fixed_list_data, type=fixed_list_type)
# Map types (key-value pairs)
map_type = pa.map_(pa.string(), pa.int64())
map_data = [
[('a', 1), ('b', 2)],
[('c', 3)],
None,
[('d', 4), ('e', 5), ('f', 6)]
]
map_array = pa.array(map_data, type=map_type)
# Union types
union_type = pa.union([
pa.field('int_field', pa.int64()),
pa.field('str_field', pa.string()),
pa.field('bool_field', pa.bool_())
], mode=pa.lib.UnionMode_DENSE)
union_data = [
(0, 42), # int_field: 42
(1, "hello"), # str_field: "hello"
(2, True), # bool_field: True
(0, 100) # int_field: 100
]
union_array = pa.array(union_data, type=union_type)
# Dictionary/encoding types
dict_type = pa.dictionary(pa.int32(), pa.string())
dict_data = pa.array(['apple', 'banana', 'apple', 'cherry', 'banana'])
dict_array = dict_data.dictionary_encode()
print(f"Original size: {dict_data.nbytes} bytes")
print(f"Dictionary encoded size: {dict_array.nbytes} bytes")
Arrow Compute & Kernels
python
import pyarrow.compute as pc
import pyarrow as pa
import numpy as np
# Create sample data
table = pa.table({
'id': pa.array(range(1000)),
'value': pa.array(np.random.randn(1000) * 100 + 500),
'category': pa.array(['A', 'B', 'C', 'D'] * 250)
})
# String operations
strings = pa.array(['hello', 'world', 'foo', 'bar', 'baz'])
upper = pc.utf8_upper(strings)
lower = pc.utf8_lower(strings)
length = pc.utf8_length(strings)
substring = pc.utf8_slice(strings, start=0, stop=2)
# Mathematical operations
values = table['value']
add_result = pc.add(values, pa.scalar(10.0))
multiply_result = pc.multiply(values, pa.scalar(2.0))
sqrt_result = pc.sqrt(values.abs())
# Aggregations with grouping
grouped = pc.hash_sum(
table['value'],
table['category']
)
# Returns struct array with category and sum
# Filtering
filtered = pc.filter(
table,
pc.greater(table['value'], 600.0)
)
# Sorting
sorted_table = pc.sort_indices(
table,
sort_keys=[
('category', 'ascending'),
('value', 'descending')
]
)
# Window functions
window_spec = pc.WindowSpec(
grouping_keys=['category'],
ordering_keys=['value'],
frame=pc.WindowFrame(
type='rows',
start=pc.WindowFrameBound(type='preceding', value=2),
end=pc.WindowFrameBound(type='following', value=2)
)
)
# Running average
running_avg = pc.window_mean(table['value'], window_spec)
# Time-based operations
timestamps = pa.array([
'2024-01-01 10:00:00',
'2024-01-01 11:00:00',
'2024-01-01 12:00:00'
], type=pa.timestamp('s'))
# Extract date parts
years = pc.year(timestamps)
months = pc.month(timestamps)
days = pc.day(timestamps)
hours = pc.hour(timestamps)
# Date arithmetic
one_day = pa.scalar(pa.duration('s', 24 * 60 * 60))
tomorrow = pc.add(timestamps, one_day)
# Complex: Filter + Transform + Aggregate
result = pc.sum(
pc.multiply(
pc.filter(table['value'], pc.greater(table['value'], 100.0)),
pa.scalar(1.1) # Add 10% tax
)
)
Arrow Flight RPC (High-Performance Data Transfer)
python
# Server side
import pyarrow.flight as flight
import pyarrow as pa
class FlightServer(flight.FlightServerBase):
def __init__(self, location="grpc://localhost:8815"):
super().__init__(location)
self.data_store = {}
def do_put(self, context, descriptor, reader, writer):
"""Handle data upload"""
table = reader.read_all()
self.data_store[descriptor.path[0].decode()] = table
print(f"Stored data with key: {descriptor.path[0].decode()}")
def do_get(self, context, ticket):
"""Handle data download"""
key = ticket.ticket.decode()
if key in self.data_store:
return flight.RecordBatchStream(self.data_store[key])
raise flight.FlightError(f"Key {key} not found")
def list_flights(self, context, criteria):
"""List available datasets"""
for key in self.data_store:
descriptor = flight.FlightDescriptor.for_path(key)
yield flight.FlightInfo(
schema=self.data_store[key].schema,
descriptor=descriptor,
endpoints=[],
total_records=self.data_store[key].num_rows,
total_bytes=self.data_store[key].nbytes
)
# Client side
def arrow_flight_example():
# Create client
client = flight.FlightClient("grpc://localhost:8815")
# Create data
table = pa.table({
'id': [1, 2, 3],
'value': [10.5, 20.3, 30.7]
})
# Upload data
descriptor = flight.FlightDescriptor.for_path("dataset1")
writer, reader = client.do_put(descriptor, table.schema)
writer.write_table(table)
writer.close()
# Download data
ticket = flight.Ticket("dataset1")
reader = client.do_get(ticket)
downloaded_table = reader.read_all()
print(f"Downloaded {downloaded_table.num_rows} rows")
return downloaded_table
# Benefits:
# - Zero-copy data transfer
# - Streaming capabilities
# - Authentication/encryption
# - Metadata exchange
# - 10-100x faster than REST APIs for large datasets
Parquet + Arrow Integration
Efficient Analytics Pipeline
python
class ParquetArrowPipeline:
"""Complete pipeline using Parquet and Arrow"""
def __init__(self):
self.arrow_table = None
self.parquet_path = None
def read_parquet_efficiently(self, path, columns=None, filters=None):
"""Read Parquet with Arrow for maximum efficiency"""
import pyarrow.dataset as ds
import pyarrow.compute as pc
# Use dataset API for advanced features
dataset = ds.dataset(path, format='parquet')
# Apply filters for predicate pushdown
if filters:
# Convert filters to Arrow expressions
arrow_filters = self._convert_filters(filters)
dataset = dataset.filter(arrow_filters)
# Scan only needed columns (projection pushdown)
if columns:
dataset = dataset.select(columns)
# Convert to Arrow Table
self.arrow_table = dataset.to_table()
return self.arrow_table
def _convert_filters(self, filters):
"""Convert simple filters to Arrow expressions"""
import pyarrow.compute as pc
import pyarrow.dataset as ds
expressions = []
for filter_expr in filters:
if isinstance(filter_expr, tuple) and len(filter_expr) == 3:
column, op, value = filter_expr
field = ds.field(column)
if op == '==':
expressions.append(field == value)
elif op == '!=':
expressions.append(field != value)
elif op == '>':
expressions.append(field > value)
elif op == '>=':
expressions.append(field >= value)
elif op == '<':
expressions.append(field < value)
elif op == '<=':
expressions.append(field <= value)
elif op == 'in':
expressions.append(field.isin(value))
elif op == 'not in':
expressions.append(~field.isin(value))
# Combine with AND
if expressions:
result = expressions[0]
for expr in expressions[1:]:
result = result & expr
return result
return None
def transform_with_arrow(self):
"""Perform transformations using Arrow compute"""
import pyarrow.compute as pc
if self.arrow_table is None:
raise ValueError("No data loaded")
# Example transformations
transformed = self.arrow_table
# Add computed columns
if 'value' in transformed.column_names:
# Vectorized operations
value_col = transformed['value']
transformed = transformed.append_column(
'value_scaled',
pc.multiply(value_col, pa.scalar(1.1))
)
transformed = transformed.append_column(
'value_category',
pc.if_else(
pc.greater(value_col, 100.0),
pa.scalar('high'),
pa.scalar('low')
)
)
# Filter rows
if 'category' in transformed.column_names:
transformed = transformed.filter(
pc.not_equal(transformed['category'], pa.scalar('exclude'))
)
# Sort
if 'timestamp' in transformed.column_names:
indices = pc.sort_indices(
transformed['timestamp'],
sort_order='descending'
)
transformed = transformed.take(indices)
self.arrow_table = transformed
return self.arrow_table
def write_parquet_optimized(self, path, **kwargs):
"""Write Arrow Table to Parquet with optimizations"""
import pyarrow.parquet as pq
default_kwargs = {
'compression': 'zstd',
'use_dictionary': True,
'write_statistics': True,
'data_page_size': 1024 * 1024, # 1MB pages
'version': '2.6',
'row_group_size': 100_000
}
# Merge with user kwargs
write_kwargs = {**default_kwargs, **kwargs}
# Determine optimal row group size
if self.arrow_table.num_rows > 1_000_000:
write_kwargs['row_group_size'] = 1_000_000
# Write to Parquet
pq.write_table(self.arrow_table, path, **write_kwargs)
self.parquet_path = path
# Verify write
parquet_file = pq.ParquetFile(path)
metadata = parquet_file.metadata
print(f"Written {metadata.num_rows} rows")
print(f"File size: {parquet_file.size / (1024*1024):.2f} MB")
print(f"Compression ratio: {self.arrow_table.nbytes / parquet_file.size:.2f}x")
return path
def analyze_performance(self, original_path, optimized_path):
"""Compare performance of different storage strategies"""
import time
# Test 1: Full scan
print("\n=== Full Scan Performance ===")
start = time.time()
original = pq.read_table(original_path)
original_time = time.time() - start
start = time.time()
optimized = pq.read_table(optimized_path)
optimized_time = time.time() - start
print(f"Original: {original_time:.3f}s")
print(f"Optimized: {optimized_time:.3f}s")
print(f"Speedup: {original_time/optimized_time:.2f}x")
# Test 2: Filtered read
print("\n=== Filtered Read Performance ===")
start = time.time()
original_filtered = pq.read_table(
original_path,
filters=[('value', '>', 100.0)]
)
original_filter_time = time.time() - start
start = time.time()
optimized_filtered = pq.read_table(
optimized_path,
filters=[('value', '>', 100.0)]
)
Comments (0)
No comments yet. Be the first to comment!