时间戳

首页 > 解决方案 > Pandas / Dask - 分组和聚合一个大的 CSV 会破坏内存和/或需要相当长的时间

python - Pandas / Dask - 分组和聚合一个大的 CSV 会破坏内存和/或需要相当长的时间

问题描述

我正在尝试一个小型 POC 来尝试分组和聚合以减少 pandas 和 Dask 中大型 CSV 中的数据，并且我观察到内存使用率高和/或比我预期的处理时间慢......有人有吗python/pandas/dask noob 有什么技巧可以改善这一点吗？

背景

我有一个构建文件摄取工具的请求，该工具将：

  1. 能够接收几 GB 的文件，其中每行包含用户 ID 和其他一些信息
  2. 做一些转换
  3. 将数据减少到{ user -> [collection of info]}
  4. 将这些数据批量发送到我们的网络服务

根据我的研究，由于文件只有几 GB，我发现 Spark 等会过大，而 Pandas/Dask 可能很适合，因此 POC。

问题

我在这里做错了什么？

（注意：如果我不设置块大小，即使在输入 1GB 时也会崩溃）

我的代码

我无法共享 CSV，但它有 1 个整数列，后跟 8 个文本列。user_id下面引用的和列都是order_id文本列。

我用随机数据生成了这些 csv，user_id我从 10 个预先随机生成的值中随机选择了列，所以我希望最终输出是 10 个用户 ID，每个用户 ID 都有一个谁知道有多少订单 ID 的集合。

熊猫

#!/usr/bin/env python3
from pandas import DataFrame, read_csv
import pandas as pd
import sys

test_csv_location = '1gb.csv'
chunk_size = 100000
pieces = list()

for chunk in pd.read_csv(test_csv_location, chunksize=chunk_size, delimiter='|', iterator=True):
    df = chunk.groupby('user_id')['order_id'].agg(size= len,list= lambda x: list(x))
    pieces.append(df)
final = pd.concat(pieces).groupby('user_id')['list'].agg(size= len,list=sum)

final.to_csv('pandastest.csv', index=False)

达斯克

#!/usr/bin/env python3
from dask.distributed import Client
import dask.dataframe as ddf
import sys

test_csv_location = '1gb.csv'
df = ddf.read_csv(test_csv_location, blocksize=6400000, delimiter='|')

# For each user, reduce to a list of order ids
grouped = df.groupby('user_id')
collection = grouped['order_id'].apply(list, meta=('order_id', 'f8'))

collection.to_csv('./dasktest.csv', single_file=True)

标签： pythonpandasdataframedask

解决方案

该groupby操作很昂贵，因为dask将尝试在工作人员之间打乱数据以检查谁拥有哪些user_id值。如果user_id有很多独特的值（听起来像），则需要在工作人员/分区之间进行大量交叉检查。

至少有两种方法可以摆脱它：

  1. 设置user_id为索引。这在索引阶段会很昂贵，但后续操作会更快，因为现在 dask 不必检查每个分区的值user_id。
df = df.set_index('user_id')
collection = df.groupby('user_id')['order_id'].apply(list, meta=('order_id', 'f8'))
collection.to_csv('./dasktest.csv', single_file=True)
  1. 如果您的文件具有您知道的结构，例如，作为一个极端示例，如果user_id进行了某种排序，则首先 csv 文件仅包含user_id以 1 开头的值（或 A，或使用任何其他符号），然后是 2等，然后您可以使用该信息以groupby仅在这些“块”中需要的方式在“块”（松散术语）中形成分区。

推荐阅读