2026年のカラム型データ: Apache Arrowメモリレイアウト、Parquetストレージ、DuckDBベクタライゼーション

Parquet File
├── File Metadata (footer)
│   └── Schema, Row Group Metadata, Column Metadata
├── Row Group 1
│   ├── Column 1 Chunk
│   │   ├── Page 1 (Data, Dictionary, Index)
│   │   ├── Page 2
│   │   └── ...
│   ├── Column 2 Chunk
│   └── ...
├── Row Group 2
│   ├── Column 1 Chunk
│   └── ...
└── ...

import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import numpy as np
import os

# Generate sample data
data = {
    'id': np.arange(100000),
    'timestamp': pd.to_datetime(pd.date_range('2022-01-01', periods=100000, freq='H')),
    'category': np.random.choice(['A', 'B', 'C', 'D'], 100000),
    'value': np.random.rand(100000) * 100
}
df = pd.DataFrame(data)

# Convert to PyArrow Table
table = pa.Table.from_pandas(df)

# Define Parquet file path
parquet_file = 'sample_data.parquet'

# Write to Parquet with specific options
# row_group_size: Controls the number of rows in each row group.
#                 Smaller row groups can improve pruning but increase metadata overhead.
#                 Larger row groups reduce metadata but might read more data than needed.
# compression: Zstd offers good balance.
# write_statistics: Essential for predicate pushdown.
pq.write_table(
    table,
    parquet_file,
    row_group_size=10000, # 10,000 rows per row group
    compression='zstd',
    write_statistics=True,
    data_page_size=8 * 1024 * 1024 # 8MB data page size
)

print(f"Parquet file '{parquet_file}' created.")

# Inspect Parquet file metadata
parquet_schema = pq.read_schema(parquet_file)
print("\nParquet Schema:")
print(parquet_schema)

parquet_metadata = pq.read_metadata(parquet_file)
print("\nParquet Metadata:")
print(parquet_metadata)

# Example of reading with predicate pushdown
# This query will only read row groups where 'timestamp' is after '2023-01-01'
# and only the 'timestamp' and 'value' columns.
start_time = pd.Timestamp('2023-01-01')
filtered_table = pq.read_table(
    parquet_file,
    columns=['timestamp', 'value'],
    filters=[('timestamp', '>', start_time)]
)

print(f"\nFiltered table (first 5 rows) after predicate pushdown and column projection:")
print(filtered_table.to_pandas().head())

# Clean up
os.remove(parquet_file)

import pyarrow as pa
import numpy as np

# Example: Integer Array
int_array = pa.array([1, 2, None, 4, 5], type=pa.int32())
print("Integer Array:")
print(int_array)
print(f"  Type: {int_array.type}")
print(f"  Buffers: {[b.to_pybytes() for b in int_array.buffers()]}") # Validity bitmap, Data buffer
print(f"  Is Null: {int_array.is_null()}")
print(f"  Values: {int_array.to_numpy()}")

# Example: String Array
string_array = pa.array(["apple", "banana", None, "cherry"], type=pa.string())
print("\nString Array:")
print(string_array)
print(f"  Type: {string_array.type}")
print(f"  Buffers: {[b.to_pybytes() for b in string_array.buffers()]}") # Validity bitmap, Offset buffer, Data buffer
print(f"  Is Null: {string_array.is_null()}")
print(f"  Values: {string_array.to_numpy()}")

# Example: RecordBatch (collection of arrays/columns)
schema = pa.schema([
    pa.field('id', pa.int64()),
    pa.field('name', pa.string()),
    pa.field('score', pa.float32())
])

record_batch = pa.RecordBatch.from_arrays(
    [
        pa.array([1, 2, 3, 4]),
        pa.array(["Alice", "Bob", "Charlie", "David"]),
        pa.array([90.5, 88.0, 92.1, 78.9])
    ],
    schema=schema
)

print("\nRecordBatch:")
print(record_batch)
print(f"  Number of columns: {record_batch.num_columns}")
print(f"  Number of rows: {record_batch.num_rows}")
print(f"  Column 'name' data buffer (first 10 bytes): {record_batch.column('name').buffers()[2].to_pybytes()[:10]}")

import duckdb
import pyarrow as pa
import pyarrow.parquet as pq
import pandas as pd
import numpy as np
import os
import time

# 1. Create a large Parquet file for demonstration
num_rows = 10_000_000
data = {
    'id': np.arange(num_rows),
    'timestamp': pd.to_datetime(pd.date_range('2020-01-01', periods=num_rows, freq='10s')),
    'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], num_rows),
    'value': np.random.rand(num_rows) * 1000,
    'is_active': np.random.choice([True, False], num_rows, p=[0.7, 0.3])
}
df_large = pd.DataFrame(data)
table_large = pa.Table.from_pandas(df_large)
large_parquet_file = 'large_sample_data.parquet'
pq.write_table(table_large, large_parquet_file, row_group_size=100000, compression='zstd', write_statistics=True)
print(f"Created large Parquet file: {large_parquet_file} with {num_rows} rows.")

# 2. Initialize DuckDB
con = duckdb.connect(database=':memory:', read_only=False)

# 3. Query Parquet directly
print("\n--- Querying Parquet file directly with DuckDB ---")
query = f"""
SELECT
    category,
    COUNT(*) AS count,
    AVG(value) AS avg_value,
    MAX(timestamp) AS max_timestamp
FROM '{large_parquet_file}'
WHERE timestamp >= '2023-01-01' AND is_active = TRUE
GROUP BY category
ORDER BY count DESC;
"""

start_time = time.time()
result_df = con.execute(query).fetchdf()
end_time = time.time()

print(f"Query executed in {end_time - start_time:.4f} seconds.")
print(result_df)

# 4. Demonstrate Arrow integration: Create an Arrow Table and query it
print("\n--- Querying an in-memory PyArrow Table with DuckDB ---")
# Create a smaller Arrow table for in-memory demo
arrow_data = pa.table({
    'city': pa.array(['NYC', 'LA', 'Chicago', 'NYC', 'LA']),
    'population': pa.array([8.4, 3.9, 2.7, 8.4, 3.9]),
    'area_sq_mi': pa.array([302, 469, 234, 302, 469])
})

# Register the Arrow table as a view in DuckDB
con.register('my_arrow_table', arrow_data)

arrow_query = """
SELECT
    city,
    SUM(population) AS total_population
FROM my_arrow_table
WHERE area_sq_mi > 300
GROUP BY city
ORDER BY total_population DESC;
"""

start_time = time.time()
arrow_result_df = con.execute(arrow_query).fetchdf()
end_time = time.time()

print(f"Arrow Table query executed in {end_time - start_time:.4f} seconds.")
print(arrow_result_df)

# 5. Clean up
con.close()
os.remove(large_parquet_file)