Modin：让数据处理飞起来的Python并行计算库

# 通过PyPI安装（推荐）
pip install modin

# 安装指定引擎（如Ray）
pip install modin[ray]

# 源码安装（适用于开发版本）
git clone https://github.com/modin-project/modin.git
cd modin
pip install .

import pandas as pd

# 读取CSV文件
df_pd = pd.read_csv("large_data.csv")

# 数据清洗：过滤缺失值
df_pd_clean = df_pd.dropna()

# 统计分析：计算数值列均值
mean_values = df_pd_clean.mean()

# 保存结果
mean_values.to_csv("pandas_result.csv")

import modin.pandas as pd  # 仅需修改导入语句

# 读取CSV文件（自动分块）
df_modin = pd.read_csv("large_data.csv")

# 数据清洗：并行执行缺失值过滤
df_modin_clean = df_modin.dropna()

# 统计分析：并行计算均值
mean_values_modin = df_modin_clean.mean()

# 保存结果（自动合并分块数据）
mean_values_modin.to_csv("modin_result.csv", index=False)

import modin.pandas as pd
import matplotlib.pyplot as plt

# 生成示例数据（Modin DataFrame）
data = pd.DataFrame({
    "category": pd.Categorical(["A", "B", "C"] * 10000),
    "value": pd.np.random.randn(30000)
})

# 计算每个类别的均值
grouped = data.groupby("category").mean()

# 转换为Pandas DataFrame进行可视化（Modin对象需先转换）
plt.bar(grouped.index.to_pandas(), grouped["value"].to_pandas())
plt.title("Category Mean Comparison")
plt.show()

# 设置默认引擎为Dask
export MODIN_ENGINE=dask

# 设置每个核心的内存限制（单位：MB）
export MODIN_MEMORY=4096

# 设置数据分块大小（单位：MB，默认100MB）
export MODIN_CHUNK_SIZE=200

import modin.config as config

# 设置引擎为Ray
config.Engine.put("ray")

# 动态调整分块大小（适用于小文件场景）
config.ChunkSize.put(50)  # 50MB per partition

import modin.pandas as pd

# 读取包含文本的数据集
df = pd.read_csv("reviews.csv")

# 并行处理：转换为小写、去除停用词
stopwords = {"the", "and", "a", "an"}

def clean_text(text):
    words = text.lower().split()
    return " ".join([word for word in words if word not in stopwords])

# 应用并行UDF（用户定义函数）
df["clean_review"] = df["review_text"].apply(clean_text, engine="ray")

import modin.pandas as pd
from datetime import datetime

# 生成时间序列数据
dates = pd.date_range(start="2020-01-01", periods=1000000, freq="D")
df = pd.DataFrame({"date": dates, "value": pd.np.random.randn(1000000)})

# 转换为datetime类型（并行处理）
df["date"] = pd.to_datetime(df["date"])

# 按周聚合统计
weekly_stats = df.groupby(pd.Grouper(key="date", freq="W")).agg({
    "value": ["mean", "std", "count"]
})

import modin.numpy as np

# 生成大规模矩阵（分块存储）
a = np.random.rand(10000, 10000)
b = np.random.rand(10000, 10000)

# 并行矩阵乘法
c = np.dot(a, b)

# 计算每行均值（并行归约）
row_means = c.mean(axis=1)

import modin.pandas as pd
from datetime import timedelta

# 读取CSV文件（自动分块，假设文件按category分区存储）
df = pd.read_csv(
    "click_logs/*.csv",
    parse_dates=["click_time"],
    dtype={"user_id": "string", "item_id": "string"}
)

# 处理缺失值：填充默认停留时间为0秒
df["duration"].fillna(0, inplace=True)

# 提取时间特征：小时、分钟
df["hour"] = df["click_time"].dt.hour
df["minute"] = df["click_time"].dt.minute

# 按类别分组统计点击次数和平均停留时间
category_stats = df.groupby("category").agg({
    "user_id": "count",        # 点击次数
    "duration": "mean"         # 平均停留时间
}).rename(columns={
    "user_id": "click_count",
    "duration": "avg_duration"
})

# 异常检测：同一用户30分钟内点击同一商品超过3次
def detect_abnormal(group):
    # 按用户和商品分组，按时间排序
    group_sorted = group.sort_values("click_time")
    # 计算相邻点击的时间差
    group_sorted["time_diff"] = group_sorted["click_time"].diff()
    # 统计连续短时间差的次数
    abnormal_mask = group_sorted["time_diff"] < timedelta(minutes=30)
    group_sorted["abnormal_count"] = abnormal_mask.cumsum()
    # 标记异常记录（次数>3）
    return group_sorted[group_sorted["abnormal_count"] > 3]

# 并行应用异常检测函数
abnormal_clicks = df.groupby(["user_id", "item_id"]).apply(
    detect_abnormal,
    engine="ray",        # 使用Ray引擎处理复杂分组操作
    result_type="expand"
)

# 保存类别统计结果
category_stats.to_csv("category_analysis.csv", index=True)

# 保存异常记录（前100条）
abnormal_clicks.head(100).to_pandas().to_csv("abnormal_clicks.csv", index=False)

# 可视化：各品类点击次数TOP5
top_categories = category_stats.sort_values("click_count", ascending=False).head(5)
top_categories["click_count"].to_pandas().plot(kind="barh", title="Top 5 Categories by Clicks")