如何分组并保留已排序文件上的组顺序
Posted
技术标签:
【中文标题】如何分组并保留已排序文件上的组顺序【英文标题】：How to groupby and preserve groups order on sorted file 【发布时间】：2020-06-28 14:01:51 【问题描述】：我有一个大型 CSV 文件，它按其中的几列排序，我们将这些列称为 sorted_columns。
我想对这些 sorted_columns 执行 groupby 并对这些组中的每一个应用一些逻辑。
该文件不完全适合内存，因此我想分块读取它并在每个块上执行groupby。
我注意到的是，即使文件已经按这些列排序，组的顺序也不会保留。
最终，这就是我想要做的：
import pandas as pd
def run_logic(key, group):
# some logic
pass
last_group = pd.DataFrame()
last_key = None
for chunk_df in df:
grouped_by_df = chunk_df.groupby(sorted_columns, sort=True)
for key, group in grouped_by_df:
if last_key is None or last_key == key:
last_key = key
last_group = pd.concat([last_group, group])
else: # last_key != key
run_logic(last_key, last_group)
last_key = key
last_group = group.copy()
run_logic(last_key, last_group)
但这不起作用，因为groupby 没有承诺保留组的顺序。如果相同的key 存在于两个连续的块中，则不能保证在第一个块中它将是最后一个组，而在下一个块中它将是第一个。
我尝试将 groupby 更改为使用 sort=False 并尝试更改列的顺序，但没有帮助。
如果键已经在原始文件中排序，是否有人知道如何保留组的顺序？
还有其他方法可以一次从文件中读取一个完整的组吗？
【问题讨论】：
【参考方案1】：我相信您的问题的本质是您试图在数据框中仅使用一次迭代来聚合每个组。您需要在内存中容纳多少组与需要读取数据帧的次数之间进行权衡
注意：我故意展示冗长的代码来传递这样一种想法，即有必要多次迭代 df。两种解决方案都变得相对复杂，但仍能达到预期的效果。代码有很多方面可以改进，感谢任何重构代码的帮助
我将使用这个虚拟的“data.csv”文件来举例说明我的解决方案。 将 data.csv 保存在与脚本相同的目录中，您可以复制并粘贴解决方案并运行它们。
sorted1,sorted2,sorted3,othe1,other2,other3,other4
1, 1, 1, 'a', 'a', 'a', 'a'
1, 1, 1, 'a', 'a', 'a', 'a'
1, 1, 1, 'a', 'a', 'a', 'a'
1, 1, 1, 'a', 'a', 'a', 'a'
2, 1, 1, 'a', 'a', 'a', 'a'
2, 1, 1, 'd', 'd', 'd', 'd'
2, 1, 1, 'd', 'd', 'd', 'a'
3, 1, 1, 'e', 'e', 'e', 'e'
3, 1, 1, 'b', 'b', 'b', 'b'
我们可以存储所有组密钥的场景中的初始解决方案：
先累积一个组的所有行，然后再处理。
基本上我会这样做： 对于 df(chunks) 中的每次迭代，选择一个组（如果内存允许，可以选择多个）。通过查找已处理组键的字典来检查它是否尚未处理，然后通过迭代每个块来累积每个块中选定的组行。当所有块迭代完成后处理组数据。
import pandas as pd
def run_logic(key, group):
# some logic
pass
def accumulate_nextGroup(alreadyProcessed_groups):
past_accumulated_group = pd.DataFrame()
pastChunk_groupKey = None
for chunk_index, chunk_df in enumerate(pd.read_csv("data.csv",iterator=True, chunksize=3)):
groupby_data = chunk_df.groupby(sorted_columns, sort=True)
for currentChunk_groupKey, currentChunk_group in groupby_data:
if (pastChunk_groupKey is None or pastChunk_groupKey == currentChunk_groupKey)\
and currentChunk_groupKey not in alreadyProcessed_groups.keys():
pastChunk_groupKey = currentChunk_groupKey
past_accumulated_group = pd.concat(
[past_accumulated_group, currentChunk_group]
)
print(f'I am the choosen group(currentChunk_groupKey) of the moment in the chunk chunk_index+1')
else:
if currentChunk_groupKey in alreadyProcessed_groups:
print(f'group(currentChunk_groupKey) is not the choosen group because it was already processed')
else:
print(f'group(currentChunk_groupKey) is not the choosen group(pastChunk_groupKey) yet :(')
return pastChunk_groupKey, past_accumulated_group
alreadyProcessed_groups =
sorted_columns = ["sorted1","sorted2","sorted3"]
number_of_unique_groups = 3 #
for iteration_in_df in range(number_of_unique_groups):
groupKey, groupData = accumulate_nextGroup(alreadyProcessed_groups)
run_logic(groupKey, groupData)
alreadyProcessed_groups[groupKey] = "Already Processed"
print(alreadyProcessed_groups)
print(f"end of iteration_in_df+1 iterations in df")
print("*"*50)
输出解决方案 1：
I am the choosen group((1, 1, 1)) of the moment in the chunk 1
I am the choosen group((1, 1, 1)) of the moment in the chunk 2
group((2, 1, 1)) is not the choosen group((1, 1, 1)) yet :(
group((2, 1, 1)) is not the choosen group((1, 1, 1)) yet :(
group((3, 1, 1)) is not the choosen group((1, 1, 1)) yet :(
(1, 1, 1): 'Already Processed'
end of 1 iterations in df
**************************************************
group((1, 1, 1)) is not the choosen group because it was already processed
group((1, 1, 1)) is not the choosen group because it was already processed
I am the choosen group((2, 1, 1)) of the moment in the chunk 2
I am the choosen group((2, 1, 1)) of the moment in the chunk 3
group((3, 1, 1)) is not the choosen group((2, 1, 1)) yet :(
(1, 1, 1): 'Already Processed', (2, 1, 1): 'Already Processed'
end of 2 iterations in df
**************************************************
group((1, 1, 1)) is not the choosen group because it was already processed
group((1, 1, 1)) is not the choosen group because it was already processed
group((2, 1, 1)) is not the choosen group because it was already processed
group((2, 1, 1)) is not the choosen group because it was already processed
I am the choosen group((3, 1, 1)) of the moment in the chunk 3
(1, 1, 1): 'Already Processed', (2, 1, 1): 'Already Processed', (3, 1, 1): 'Already Processed'
end of 3 iterations in df
**************************************************
更新解决方案 2：在我们无法将所有组键存储在字典中的情况下：
在我们无法将所有组键存储在字典中的情况下，我们需要使用在每个块中创建的每个组相对索引来为每个组创建一个全局引用索引。 （注意这个解决方案比前一个更密集）
这个解决方案的要点是我们不需要组键值来识别组。 更深入地说，您可以将每个块想象成反向链表中的一个节点，其中第一个块指向 null，第二个块指向第一个块，依此类推……数据帧上的一次迭代对应于该链表中的一次遍历.对于每个步骤（处理一个块），您每次需要保留的唯一信息是前一个块的头、尾和大小，只有有了这些信息，您才能为任何块中的组键分配一个唯一的索引标识符。
其他重要信息是，由于文件已排序，块的第一个元素的引用索引将是最后一个元素的前一个块的最后一个元素 + 1。这使得从块索引推断全局引用索引成为可能.
import pandas as pd
import pysnooper
def run_logic(key, group):
# some logic
pass
def generate_currentChunkGroups_globalReferenceIdx(groupby_data,
currentChunk_index, previousChunk_link):
if currentChunk_index == 0:
groupsIn_firstChunk=len(groupby_data.groups.keys())
currentGroups_globalReferenceIdx = [(i,groupKey)
for i,(groupKey,_) in enumerate(groupby_data)]
else:
lastChunk_firstGroup, lastChunk_lastGroup, lastChunk_nGroups \
= previousChunk_link
currentChunk_firstGroupKey = list(groupby_data.groups.keys())[0]
currentChunk_nGroups = len(groupby_data.groups.keys())
lastChunk_lastGroupGlobalIdx, lastChunk_lastGroupKey \
= lastChunk_lastGroup
if currentChunk_firstGroupKey == lastChunk_lastGroupKey:
currentChunk_firstGroupGlobalReferenceIdx = lastChunk_lastGroupGlobalIdx
else:
currentChunk_firstGroupGlobalReferenceIdx = lastChunk_lastGroupGlobalIdx + 1
currentGroups_globalReferenceIdx = [
(currentChunk_firstGroupGlobalReferenceIdx+i, groupKey)
for (i,groupKey) in enumerate(groupby_data.groups.keys())
]
next_previousChunk_link = (currentGroups_globalReferenceIdx[0],
currentGroups_globalReferenceIdx[-1],
len(currentGroups_globalReferenceIdx)
)
return currentGroups_globalReferenceIdx, next_previousChunk_link
def accumulate_nextGroup(countOf_alreadyProcessedGroups, lastChunk_index, dataframe_accumulator):
previousChunk_link = None
currentIdx_beingProcessed = countOf_alreadyProcessedGroups
for chunk_index, chunk_df in enumerate(pd.read_csv("data.csv",iterator=True, chunksize=3)):
print(f'ITER:iteration_in_df CHUNK:chunk_index InfoPrevChunk:previousChunk_link lastProcessed_chunk:lastChunk_index')
if (lastChunk_index != None) and (chunk_index < lastChunk_index):
groupby_data = chunk_df.groupby(sorted_columns, sort=True)
currentChunkGroups_globalReferenceIdx, next_previousChunk_link \
= generate_currentChunkGroups_globalReferenceIdx(
groupby_data, chunk_index, previousChunk_link
)
elif((lastChunk_index == None) or (chunk_index >= lastChunk_index)):
if (chunk_index == lastChunk_index):
groupby_data = chunk_df.groupby(sorted_columns, sort=True)
currentChunkGroups_globalReferenceIdx, next_previousChunk_link \
= generate_currentChunkGroups_globalReferenceIdx(
groupby_data, chunk_index, previousChunk_link
)
currentChunkGroupGlobalIndexes = [GlobalIndex \
for (GlobalIndex,_) in currentChunkGroups_globalReferenceIdx]
if((lastChunk_index is None) or (lastChunk_index <= chunk_index)):
lastChunk_index = chunk_index
if currentIdx_beingProcessed in currentChunkGroupGlobalIndexes:
currentGroupKey_beingProcessed = [tup
for tup in currentChunkGroups_globalReferenceIdx
if tup[0] == currentIdx_beingProcessed][0][1]
currentChunk_group = groupby_data.get_group(currentGroupKey_beingProcessed)
dataframe_accumulator = pd.concat(
[dataframe_accumulator, currentChunk_group]
)
else:
groupby_data = chunk_df.groupby(sorted_columns, sort=True)
currentChunkGroups_globalReferenceIdx, next_previousChunk_link \
= generate_currentChunkGroups_globalReferenceIdx(
groupby_data, chunk_index, previousChunk_link
)
currentChunkGroupGlobalIndexes = [GlobalIndex \
for (GlobalIndex,_) in currentChunkGroups_globalReferenceIdx]
if((lastChunk_index is None) or (lastChunk_index <= chunk_index)):
lastChunk_index = chunk_index
if currentIdx_beingProcessed in currentChunkGroupGlobalIndexes:
currentGroupKey_beingProcessed = [tup
for tup in currentChunkGroups_globalReferenceIdx
if tup[0] == currentIdx_beingProcessed][0][1]
currentChunk_group = groupby_data.get_group(currentGroupKey_beingProcessed)
dataframe_accumulator = pd.concat(
[dataframe_accumulator, currentChunk_group]
)
else:
countOf_alreadyProcessedGroups+=1
lastChunk_index = chunk_index-1
break
previousChunk_link = next_previousChunk_link
print(f'Done with chunks for group of global index:currentIdx_beingProcessed corresponding to groupKey:currentGroupKey_beingProcessed')
return countOf_alreadyProcessedGroups, lastChunk_index, dataframe_accumulator, currentGroupKey_beingProcessed
sorted_columns = ["sorted1","sorted2","sorted3"]
number_of_unique_groups = 3 #
lastChunk_index = None
for iteration_in_df in range(number_of_unique_groups):
dataframe_accumulator = pd.DataFrame()
countOf_alreadyProcessedGroups,lastChunk_index, group_data, currentGroupKey_Processed=\
accumulate_nextGroup(
iteration_in_df, lastChunk_index, dataframe_accumulator
)
run_logic(currentGroupKey_Processed, dataframe_accumulator)
print(f"end of iteration number iteration_in_df+1 in the df and processed currentGroupKey_Processed")
print(group_data)
print("*"*50)
输出解决方案 2：
ITER:0 CHUNK:0 InfoPrevChunk:None lastProcessed_chunk:None
ITER:0 CHUNK:1 InfoPrevChunk:((0, (1, 1, 1)), (0, (1, 1, 1)), 1) lastProcessed_chunk:0
ITER:0 CHUNK:2 InfoPrevChunk:((0, (1, 1, 1)), (1, (2, 1, 1)), 2) lastProcessed_chunk:1
Done with chunks for group of global index:0 corresponding to groupKey:(1, 1, 1)
end of iteration number 1 in the df and processed (1, 1, 1)
sorted1 sorted2 sorted3 othe1 other2 other3 other4
0 1 1 1 'a' 'a' 'a' 'a'
1 1 1 1 'a' 'a' 'a' 'a'
2 1 1 1 'a' 'a' 'a' 'a'
3 1 1 1 'a' 'a' 'a' 'a'
**************************************************
ITER:1 CHUNK:0 InfoPrevChunk:None lastProcessed_chunk:1
ITER:1 CHUNK:1 InfoPrevChunk:((0, (1, 1, 1)), (0, (1, 1, 1)), 1) lastProcessed_chunk:1
ITER:1 CHUNK:2 InfoPrevChunk:((0, (1, 1, 1)), (1, (2, 1, 1)), 2) lastProcessed_chunk:1
Done with chunks for group of global index:1 corresponding to groupKey:(2, 1, 1)
end of iteration number 2 in the df and processed (2, 1, 1)
sorted1 sorted2 sorted3 othe1 other2 other3 other4
4 2 1 1 'a' 'a' 'a' 'a'
5 2 1 1 'd' 'd' 'd' 'd'
6 2 1 1 'd' 'd' 'd' 'a'
**************************************************
ITER:2 CHUNK:0 InfoPrevChunk:None lastProcessed_chunk:2
ITER:2 CHUNK:1 InfoPrevChunk:((0, (1, 1, 1)), (0, (1, 1, 1)), 1) lastProcessed_chunk:2
ITER:2 CHUNK:2 InfoPrevChunk:((0, (1, 1, 1)), (1, (2, 1, 1)), 2) lastProcessed_chunk:2
Done with chunks for group of global index:2 corresponding to groupKey:(3, 1, 1)
end of iteration number 3 in the df and processed (3, 1, 1)
sorted1 sorted2 sorted3 othe1 other2 other3 other4
7 3 1 1 'e' 'e' 'e' 'e'
8 3 1 1 'b' 'b' 'b' 'b'
**************************************************
【讨论】：
谢谢。我有一个担忧——如果我也无法将所有组键保存在内存中怎么办？我认为您在回答中假设了这一点，对吗？ 我假设了两件重要的事情：1）一个存储已处理组的字典（以跟踪已处理的内容）2）对于所有组，组数据适合内存 我更新了解决方案，它更复杂。我会一点一点地编辑，尝试详细说明它的每个主要部分。 谢谢。在您的第一个解决方案中，可能存储键的哈希而不是键本身以节省一些内存，如果键的数量是一个问题，不确定它是否有帮助，但谢谢。【参考方案2】：itertools.groupby
将为按该键分组的所有值返回键和迭代器。 如果您的文件已经按您想要的键排序，那么您就可以开始了。 groupby 函数将为您处理几乎所有事情。
来自documentation：
groupby()的操作类似于 Unix 中的uniq过滤器。每次键函数的值发生变化时，它都会生成一个中断或新组（这就是为什么通常需要使用相同的键函数对数据进行排序的原因）。这种行为与 SQL 的 GROUP BY 不同，后者聚合公共元素而不管其输入顺序如何。
run_logic 是您要应用于记录组的任何业务逻辑。这个例子只是简单地计算迭代器中的观察次数。
data_iter 只需为每个 CSV 发出 1 行。只要您的文件按所需字段排序，您就不需要将整个文件读入内存。
chunks 使用 groupby 使用输入行的前 3 个字段对输入迭代器进行分组。它产生键和与该键关联的值的相应迭代器。
#!/usr/bin/env python3
import csv
from itertools import groupby
def run_logic(key, group):
cntr = 0
for rec in group:
cntr = cntr + 1
return (key, cntr)
def data_iter(filename):
with open(filename, "r") as fin:
csvin = csv.reader(fin)
for row in csvin:
yield row
def chunks(diter):
for chunk, iter_ in groupby(diter, key=lambda x: x[0:3]):
yield (chunk, iter_)
if __name__ == "__main__":
csviter = data_iter("test.csv")
chunk_iter = chunks(csviter)
for chunk, iter_ in chunk_iter:
print(run_logic(chunk, iter_))
输入数据
['1', '1', '1', 'a', 'a', 'a', 'a']
['1', '1', '1', 'b', 'b', 'b', 'b']
['1', '1', '1', 'c', 'c', 'c', 'c']
['1', '1', '1', 'd', 'd', 'd', 'd']
['1', '1', '1', 'e', 'e', 'e', 'e']
['2', '1', '1', 'a', 'a', 'a', 'a']
['2', '1', '1', 'd', 'd', 'd', 'd']
['2', '1', '1', 'e', 'e', 'e', 'e']
['2', '1', '1', 'b', 'b', 'b', 'b']
['2', '1', '1', 'c', 'c', 'c', 'c']
['3', '1', '1', 'e', 'e', 'e', 'e']
['3', '1', '1', 'b', 'b', 'b', 'b']
['3', '1', '1', 'c', 'c', 'c', 'c']
['3', '1', '1', 'a', 'a', 'a', 'a']
['3', '1', '1', 'd', 'd', 'd', 'd']
分组数据
组：['1', '1', '1']
['1', '1', '1', 'a', 'a', 'a', 'a']
['1', '1', '1', 'b', 'b', 'b', 'b']
['1', '1', '1', 'c', 'c', 'c', 'c']
['1', '1', '1', 'd', 'd', 'd', 'd']
['1', '1', '1', 'e', 'e', 'e', 'e']
组：['2', '1', '1']
['2', '1', '1', 'a', 'a', 'a', 'a']
['2', '1', '1', 'd', 'd', 'd', 'd']
['2', '1', '1', 'e', 'e', 'e', 'e']
['2', '1', '1', 'b', 'b', 'b', 'b']
['2', '1', '1', 'c', 'c', 'c', 'c']
组：['3', '1', '1']
['3', '1', '1', 'e', 'e', 'e', 'e']
['3', '1', '1', 'b', 'b', 'b', 'b']
['3', '1', '1', 'c', 'c', 'c', 'c']
['3', '1', '1', 'a', 'a', 'a', 'a']
['3', '1', '1', 'd', 'd', 'd', 'd']
应用业务逻辑
组：['1', '1', '1']
(['1', '1', '1'], 5)
组：['2', '1', '1']
(['2', '1', '1'], 5)
组：['3', '1', '1']
(['3', '1', '1'], 5)
【讨论】：
谢谢你。编辑了您的答案以包含来自文档的引用:)以上是关于如何分组并保留已排序文件上的组顺序的主要内容，如果未能解决你的问题，请参考以下文章