pip install smallpond

import smallpond

# 默认配置：自动检测可用节点
sp = smallpond.init()

# 自定义配置
sp = smallpond.init(
    num_nodes=10,           # 使用 10 个节点
    duckdb_memory="8GB",    # 每个节点内存限制
    data_dir="/smallpond/data",  # 3FS 数据路径
)

# 读 Parquet（自动分片）
df = sp.read_parquet("huge_dataset/*.parquet")

# 读 CSV
df = sp.read_csv("logs/*.csv")

# 读 JSON
df = sp.read_json("events/*.jsonl")

# 按某列哈希重分区（类似 Spark 的 repartition）
df = df.repartition(10, hash_by="user_id")

# 随机重分区
df = df.repartition(20)

# 注意：{0} 是占位符，代表 DataFrame
df_result = sp.partial_sql(
    "SELECT user_id, COUNT(*), AVG(amount) "
    "FROM {0} "
    "WHERE event_type = 'purchase' "
    "GROUP BY user_id",
    df
)

# 写回 Parquet
df.write_parquet("output/")

# 转为 Pandas DataFrame（适合小结果集）
pandas_df = df.to_pandas()

# 查看行数
print(f"Total rows: {df.count()}")

import smallpond

# 1. 初始化
sp = smallpond.init()

# 2. 读取 1TB 用户事件数据
events = sp.read_parquet("s3://data/events/*.parquet")
users = sp.read_parquet("s3://data/users/*.parquet")

# 3. 重分区（按用户 ID 分布）
events = events.repartition(50, hash_by="user_id")

# 4. 分布式 JOIN + 聚合
result = sp.partial_sql("""
    SELECT
        u.country,
        u.tier,
        COUNT(DISTINCT e.user_id) AS active_users,
        SUM(e.revenue) AS total_revenue,
        AVG(e.revenue) AS avg_revenue_per_user
    FROM {0} e
    JOIN users u ON e.user_id = u.user_id
    WHERE e.event_date >= '2026-01-01'
    GROUP BY u.country, u.tier
""", events)

# 5. 写结果
result.write_parquet("output/daily_report/")

# 6. 打印预览
print(result.to_pandas().head(20))

import smallpond
import pandas as pd
import numpy as np
from datetime import datetime, timedelta

# 初始化 Smallpond
sp = smallpond.init()

# 模拟用户数据（1000 万用户）
num_users = 10_000_000
users_df = pd.DataFrame({
    "user_id": range(1, num_users + 1),
    "country": np.random.choice(["CN", "US", "JP", "DE", "BR"], num_users),
    "tier": np.random.choice(["bronze", "silver", "gold", "platinum"], num_users,
                              p=[0.5, 0.3, 0.15, 0.05]),
    "registration_date": (
        datetime.now() - pd.to_timedelta(np.random.randint(1, 365*3, num_users), unit="D")
    ).strftime("%Y-%m-%d"),
})
users_df.to_parquet("/tmp/sample/users.parquet")
print(f"用户数据已生成: {len(users_df):,} 条")

# 模拟事件数据（每日约 5000 万条，模拟 30 天 = 15 亿条）
num_days = 3  # 演示用 3 天，实际可以全量
events_per_day = 50_000_000

for day in range(num_days):
    date_str = (datetime.now() - timedelta(days=day)).strftime("%Y-%m-%d")
    n = events_per_day
    events_df = pd.DataFrame({
        "event_id": range(day * n + 1, (day + 1) * n + 1),
        "user_id": np.random.randint(1, num_users + 1, n),
        "event_type": np.random.choice(
            ["page_view", "click", "add_cart", "purchase", "favorite"],
            n, p=[0.6, 0.2, 0.1, 0.07, 0.03]
        ),
        "revenue": np.where(
            np.random.random(n) < 0.07,  # 7% 的购买行为
            np.random.uniform(10, 500, n).round(2),
            0.0
        ),
        "event_date": date_str,
        "timestamp": [
            f"{date_str} {np.random.randint(0,24):02d}:{np.random.randint(0,60):02d}:{np.random.randint(0,60):02d}"
            for _ in range(n)
        ],
    })
    events_df.to_parquet(f"/tmp/sample/events/{date_str}.parquet")
    print(f"事件数据已生成: {date_str} ({n:,} 条)")

import smallpond

sp = smallpond.init()

# 1. 读取数据
print("读取数据...")
events = sp.read_parquet("/tmp/sample/events/*.parquet")
users = sp.read_parquet("/tmp/sample/users.parquet")

# 2. 按 user_id 重分区（确保 JOIN 在本地完成）
events = events.repartition(10, hash_by="user_id")

# 3. 执行分布式 SQL 分析
print("执行分布式查询...")
result = sp.partial_sql("""
    SELECT
        u.country,
        u.tier,
        e.event_date,
        COUNT(DISTINCT e.user_id) AS active_users,
        COUNT(*) AS total_events,
        SUM(CASE WHEN e.event_type = 'purchase' THEN 1 ELSE 0 END) AS purchases,
        SUM(e.revenue) AS total_revenue,
        SUM(e.revenue) / NULLIF(COUNT(DISTINCT e.user_id), 0) AS revenue_per_user,
        SUM(CASE WHEN e.event_type = 'add_cart' THEN 1 ELSE 0 END) AS cart_adds,
        SUM(CASE WHEN e.event_type = 'purchase' THEN 1 ELSE 0 END) * 1.0
            / NULLIF(SUM(CASE WHEN e.event_type = 'add_cart' THEN 1 ELSE 0 END), 0)
            AS cart_to_purchase_rate
    FROM {0} e
    JOIN users u ON e.user_id = u.user_id
    WHERE e.event_date >= '2026-04-01'
    GROUP BY u.country, u.tier, e.event_date
""", events)

# 4. 查看结果
pandas_result = result.to_pandas()
print(f"\n结果行数: {len(pandas_result):,}")
print(f"\nTop 20 结果预览:")
print(pandas_result.head(20))

# 5. 写入结果
result.write_parquet("/tmp/sample/output/daily_stats/")
print("\n结果已写入 /tmp/sample/output/daily_stats/")

# 1. 所有节点安装 3FS
# 参考: https://github.com/deepseek-ai/3FS

# 2. 所有节点安装 Smallpond
pip install smallpond

# 3. 配置 3FS 挂载点（所有节点相同路径）
# /smallpond/data ← 所有节点通过 3FS 共享

# 4. 将数据复制到 3FS
cp /local/data/*.parquet /smallpond/data/

# 5. 在任何节点上提交任务
python my_etl_script.py