parquet-analysis

使用 Python 和 Ibis 分析 Parquet

使用 Ibis 分析 Parquet 文件。Ibis 是一种与数据库无关的 Python DataFrame API。Ibis 会将 Python 操作转换为针对底层后端优化的查询（默认使用 DuckDB 处理 Parquet 文件）。

快速入门

基本工作流

import ibis

# Connect to DuckDB (optimized for parquet)
con = ibis.duckdb.connect()

# Read parquet file
table = con.read_parquet("data.parquet")

# Explore
print(table.schema())      # View schema
print(table.head(10))      # Preview data
print(table.describe())    # Summary statistics

# Filter and aggregate
summary = (
    table
    .filter(table.amount > 100)
    .group_by("category")
    .aggregate(
        total=table.amount.sum(),
        avg=table.amount.mean(),
        count=table.count()
    )
)
print(summary.execute())

# Export
con.to_parquet(summary, "output.parquet")

核心操作

选择和筛选

# Select columns
selected = table.select("id", "amount", "date")

# Filter rows
filtered = table.filter(
    (table.amount > 100) &
    (table.date >= "2024-01-01")
)

# Sort
sorted_data = table.order_by(table.amount.desc())

转换

# Add computed columns
enriched = table.mutate(
    revenue=table.quantity * table.unit_price,
    year=table.date.year(),
    size=ibis.case()
        .when(table.amount < 100, "small")
        .when(table.amount < 1000, "medium")
        .else_("large")
        .end()
)

聚合

# Group by and summarize
by_category = (
    table.group_by("category")
    .aggregate(
        total=table.amount.sum(),
        avg=table.amount.mean(),
        count=table.count()
    )
)

连接

# Read and join multiple files
customers = con.read_parquet("customers.parquet")
orders = con.read_parquet("orders.parquet")

joined = (
    orders
    .join(customers, orders.customer_id == customers.id, how="left")
    .select(
        orders.order_id,
        orders.amount,
        customers.name
    )
)

导出

# To parquet
con.to_parquet(result, "output.parquet")

# To CSV (via pandas)
df = result.execute()
df.to_csv("output.csv", index=False)

常见模式

数据质量检查

# Row count
row_count = table.count().execute()

# Check for nulls
null_counts = table.select([
    col.isnull().sum().name(f"{col}_nulls")
    for col in table.columns
]).execute()

# Value distribution
table.group_by("category").aggregate(count=table.count()).execute()

基于时间的分析

# Monthly aggregation
monthly = (
    table.mutate(month=table.date.truncate("M"))
    .group_by("month")
    .aggregate(total=table.amount.sum())
    .order_by("month")
)

# Extract date components
dated = table.mutate(
    year=table.date.year(),
    month=table.date.month(),
    quarter=table.date.quarter()
)

窗口函数

# Rank within groups
ranked = table.mutate(
    rank=table.amount.rank().over(
        ibis.window(group_by="category", order_by=table.amount.desc())
    )
)

# Running total
with_cumsum = table.mutate(
    cumulative=table.amount.sum().over(
        ibis.window(order_by="date", rows=(None, 0))
    )
)

最佳实践

1. 尽早筛选：在聚合前应用筛选条件，以减少数据量
2. 使用惰性求值：Ibis 操作不会立即执行，只有调用 .execute() 时才会执行；先串联操作，再执行
3. 处理空值：显式检查并处理空值
4. 利用选择器进行列操作（参见 REFERENCE.md）

详细资源

• EXAMPLES.md - 所有常见任务的完整示例
• REFERENCE.md - Ibis 操作的详细 API 参考
• scripts/analyze.py - 完整的可运行示例脚本
• scripts/template.py - 用于分析的快速入门模板
• ../ibis-data/SKILL.md - 更多 Ibis 背景信息和模式

脚本

完整分析工作流

运行完整示例脚本：

python scripts/analyze.py

此脚本演示：

• 数据探索和质量检查
• 转换和丰富数据
• 聚合和筛选
• 连接（如果有多个文件）
• 多种导出格式
• 可选的可视化

快速入门模板

复制并自定义模板：

cp scripts/template.py my_analysis.py
# Edit with your file paths and logic
python my_analysis.py

安装

安装所需的软件包：

uv add "ibis-framework[duckdb]"

# Optional for visualization
uv add matplotlib

故障排除

导入错误：确保已安装 ibis-framework[duckdb]

大型文件：DuckDB 能够高效处理大型 Parquet 文件。对于非常大的数据集：

• 尽早并积极地筛选
• 选择性地读取列：con.read_parquet("file.parquet", columns=["id", "amount"])
• 分块处理或使用筛选条件更严格的查询

连接时架构不匹配：使用 .cast() 确保列类型一致：

table = table.mutate(id=table.id.cast("int64"))

性能：对于复杂查询，请使用 ibis.to_sql(table) 查看生成的 SQL，了解实际执行的内容