按多列对大型 CSV 文件进行排序/压缩答案

【问题标题】：Sorting/condensing large CSV file by multiple columns按多列对大型 CSV 文件进行排序/压缩
【发布时间】：2017-05-17 10:47:12
【问题描述】：

所以我正在使用的 csv 文件看起来像：

Date Time,            SegmentID,   indicatorvalue
2016-12-01T00:00:00Z  147649       1
2016-12-01T00:01:00Z  147649       0
2016-12-01T00:02:00Z  147649       1
...                   ...          ...
2016-12-01T00:23:00Z  2938733      0

等等。

我想要的是将其浓缩为所有细分的列表，然后是该细分的多少条目具有指标值的百分比，按一天中的时间（AM/IP/PM/OP）

例如：

segmentID,  Time Period,  Percentage
147649      AM            78
147649      IP            100
147649      PM            60
147649      OP            30
243546      AM            79
243546      IP            98
...         ...           ...

我的尝试是设置一个 for 循环，以便对于每个段，它将保持一天中 4 次的指标百分比的运行总数，然后一旦检测到它已更改为新的段 ID，它将追加将这些值转换为字符串。

问题是segmentID 是无序的，导致同一个segmentID 有多个条目。我尝试按 SegmentID 对文件进行排序，但文件太大了。关于替代方法的任何建议？

编辑：有人删除了他们的评论，说我应该使用 Pandas，所以经过一番研究，我的代码目前是：

import numpy as np
import pandas as pd

df=pd.read_csv("data.csv",sep=",",index_col="Segment ID",usecols=["Segment ID","Date Time","indicatorvalue"])
df['Date Time'] = ['AM' if '06' <= x[11:13] < '10' 
    else 'IP' if '10' <= x[11:13] < '16' 
    else 'PM' if '16' <= x[11:13] < '19'
    else 'OP' if '19' <= x[11:13] or x[11:13] < '06'
    else 'Error' for x in df['Date Time']]

现在我只需要弄清楚如何压缩包含“日期时间”和“段 ID”重复条目的行，同时平均它们的“指标值”。

【问题讨论】：

质量有多大？
大约 3.5GB，不知道有多少行，但以百万计
还有多少内存可用，预计有多少段？

标签： python python-3.x sorting csv

【解决方案1】：

我假设输入文件的结构是每分钟一行，给出开始时间、段 id 和指标值。

如果段的数量与可用内存兼容，我将一次只在线读取输入文件，并在每个段的 8 个计数器中添加分钟，即每天的时间和指标值。这意味着初始文件只读取一次，不排序，唯一关键的数量是段数 - 如果它变得太高，我会使用 sqlite3 或 dbm 数据库而不是 dict。

对于您当前的示例（使用 not csv），代码可能是：

class Segment:
    labels = ['AM', 'IP', 'PM', 'OP']
    def __init__(self, segid):
        self.id = segid
        self.values = [ [ 0, 0 ] for i in range(4) ]
    def add(self, hour, indic):
        ix = 3
        if hour >= 6 and hour < 10: ix=0
        elif hour >= 10 and hour < 16: ix=1
        elif hour >= 16 and hour < 19: ix = 2
        self.values[ix][indic] += 1
    def percent(self, ix):
        try:
            return int(.5 + (100 * self.values[ix][1] /
                 (self.values[ix][0] + self.values[ix][1])))
        except ZeroDivisionError:
            return 0

dummy = next(fd)
splitter = re.compile(' +')
segments = dict()
for line in fd:  # read and store
    d, seg, indic = splitter.split(line.strip())  # could be replaced with a csv reader
    hour = int(d[11:13])
    if not seg in segments:
        segments[seg] = Segment(seg)
    segments[seg].add(hour, int(indic))
for seg in sorted(segments.keys()):  # output the stats
    for ix in range(4):
        print(seg, Segment.labels[ix], segments[seg].percent(ix))

以上代码缺少对错误或异常情况的测试

【讨论】：

【解决方案2】：

我设法用 pandas 找到了解决方案。我使用的代码是：

import pandas as pd

df=pd.read_csv("data.csv",sep=",",usecols=["Segment ID","Date Time","indicator value"])

df['Date Time'] = ['AM' if '06' <= x[11:13] < '10' 
   else 'IP' if '10' <= x[11:13] < '16' 
   else 'PM' if '16' <= x[11:13] < '19'
   else 'OP' if '19' <= x[11:13] or x[11:13] < '06'
   else 'Error' for x in df['Date Time']]

grouped = df.groupby(['Segment ID','Date Time']).mean() 

grouped.to_csv('output.csv', sep =',')

我不能推荐 Pandas。

【讨论】：