Блог

Предположим, мы хотим отсортировать файл, содержащий 40000 строк вокруг a column=X . Давайте также предположим, что одни и те же значения широко распространены по всей таблице, так что строки с одинаковым значением in column=X встречаются не только в первых 1000 строках. Теперь, если мы прочитаем файл по частям и рассмотрим только 1000 строк, мы можем перепутать другие строки с тем же значением, что и в column=X , если мы снова отсортируем таблицу вокруг этого столбца. Итак, как мы можем решить эту проблему, пожалуйста? Код не требуется, так как данные недоступны, но, пожалуйста, я ищу ваше мнение по этому вопросу? Должны ли мы пойти ссортировка слиянием путем параллельного использования каждого фрагмента алгоритма сортировки слиянием, а затем рекомбинации результатов? Я не вижу способа сделать это с пандами, но я не уверен?

 import pandas as pd
chunk_size = 1000
batch_no = 1
for chunk in pd.read_csv('data.csv', chunksize=chunk_size):
    chunk.sort_values(by='X', inplace=True)
    chunk.to_csv('data'  str(batch_no)   '.csv', index=False)
    batch_no  =1
 

Вам нужно объединить отсортированные csv-файлы, к счастью, Python предоставляет для этого функцию. Используйте его, как показано ниже:

 from operator import itemgetter

import pandas as pd
import numpy as np
import csv
import heapq

# generate test data
test_data = pd.DataFrame(data=[[f"label{i}", val] for i, val in enumerate(np.random.uniform(size=40000))],
                         columns=["label", "X"])
test_data.to_csv("data.csv", index=False)

# read and sort each chunk
chunk_size = 1000
file_names = []
for batch_no, chunk in enumerate(pd.read_csv("data.csv", chunksize=chunk_size), 1):
    chunk.sort_values(by="X", inplace=True)
    file_name = f"data_{batch_no}.csv"
    chunk.to_csv(file_name, index=False)
    file_names.append(file_name)

# merge the chunks
chunks = [csv.DictReader(open(file_name)) for file_name in file_names]
with open("data_sorted.csv", "w") as outfile:
    field_names = ["label", "X"]
    writer = csv.DictWriter(outfile, fieldnames=field_names)
    writer.writeheader()
    for row in heapq.merge(*chunks, key=itemgetter("X")):
        writer.writerow(row)
 

Из документации по heapq.merge:

Объедините несколько отсортированных входных данных в один отсортированный выходной (например, объедините записи с отметками времени из нескольких файлов журнала). Возвращает итератор по отсортированным значениям.

Аналогично сортировке(itertools.chain(*iterables)), но возвращает итерацию, не извлекает данные в память все сразу и предполагает, что каждый из входных потоков уже отсортирован (от наименьшего до наибольшего).

Таким образом, использование, как вы можете прочитать в приведенной выше цитате (выделено мной), с помощью heapq.merge не загрузит все данные в память. Также стоит отметить, что сложность этой функции заключается O(n) в том, что n — это размер всех данных. Поэтому общий алгоритм сортировки является O(nlogn)