Как быстрее прочитать огромный CSV?

Чтобы установить block_size, вам нужно будет напрямую использовать PyArrow:

from pyarrow import csv

# read CSV using PyArrow with ReadOptions
read_options = csv.ReadOptions(
    block_size=1024,  # <= define a block size here
)

table = csv.read_csv("file.csv", read_options=read_options)

# convert PyArrow Table to pandas DataFrame
df = table.to_pandas()

Вы также можете использовать более эффективные форматы файлов для хранения и загрузки больших наборов данных, например:

• Паркет : Паркет хранит данные по столбцам, а не по строкам, как CSV (полезно при загрузке части столбцов). Он предлагает сжатие и эффективное кодирование. В Pandas есть методы для создания и чтения файлов паркета в виде фреймов данных.

• Pickle : Pickle реализует двоичные протоколы для сериализации и десериализации структуры объектов Python. Файлы представляют собой сериализованные представления объектов Python и могут быть намного меньше по размеру по сравнению с текстовыми форматами, такими как CSV. В Pandas есть методы для создания и чтения Pickle файлов как фреймов данных.

Об исключении ArrowInvalid

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

data = 'value\nAB\nABC123456DEF\nAB'   # csv data

print('Blocks of 6 bytes:\n' 
      '------------------')
for p in range(0, len(data), 6):
    print(repr(data[p:p+6]))

Как вы можете видеть в последних трех строках ниже, третья запись "ABC123456DEF\n" данных занимает 3 блока по 6 байт:

Blocks of 6 bytes:
------------------
'value\n'
'AB\nABC'
'123456'
'DEF\nAB'

Итак, если вы прочитаете эти данные в виде файла csv с помощью block_size=6, вы столкнетесь с исключением ArrowInvalid:

from pyarrow import csv
from io import BytesIO

data = 'value\nAB\nABC123456DEF\nAB'
read_options = csv.ReadOptions(block_size=6)
csv.read_csv(BytesIO(data.encode()), read_options=read_options)   # ArrowInvalid expected

Чтобы избежать этого исключения, мы можем определить размер блока, достаточно большой, чтобы покрыть самую длинную запись. В приведенном выше случае самая длинная запись имеет размер 12, поэтому мы можем прочитать данные с помощью block_size=12 (может быть и меньше, но в общем случае без гарантии успеха).

Pandas<=2.2 придерживается значения по умолчанию block_size, которое, как мы видим в документации Arrow Apache, составляет 1 мегабайт. Поскольку в вашем случае этого недостаточно, вам придется попробовать block_size размером более 1048576 байт с самим pyarrow, как в примере выше.

Чтение больших файлов CSV с помощью pyarrow

Давайте создадим данные, похожие на ваш пример:

def gen_data(min_number=0, max_number=1_000, min_count=1<<10, max_count=1<<20, seed=0):
    from numpy.random import default_rng
    rng = default_rng(seed)
    while True:
        size = rng.integers(min_count, max_count)
        yield ','.join(map(str, rng.integers(min_number, max_number, size)))

from itertools import count

data_file = 'test.csv'
rec_id = count()
seq = gen_data()
headers = ['id', 'col1', 'col2']
file_size = int(1.22 * (1<<30))    # create a csv file of size >= 1.22 Gb
sep = ' '

with open(data_file, 'w') as file:
    size = file.write(sep.join(headers) + '\n')
    while size < file_size:
        size += file.write(str(next(rec_id)) + sep)
        size += file.write(next(seq) + sep)
        size += file.write(next(seq) + '\n')

Вряд ли вам хочется хранить все данные в памяти, скорее всего, вы хотите просто получить ответ о данных. Это можно сделать по частям с помощью pyarrow.csv.open_csv. Например, если первоначальный запрос «Собрать общие элементы по столбцам», мы можем сделать это так:

from pyarrow.csv import open_csv

# find the number of megabytes to cover the longest line of a data file
megabyte = 1<<20
max_len = max(len(line) for line in open(data_file, 'rb'))
block_size = megabyte * (1 + (max_len-1) // megabyte)       

read_options = csv.ReadOptions(block_size=block_size)
parse_options = csv.ParseOptions(delimiter=' ')
convert_options = csv.ConvertOptions(include_columns=['col1', 'col2'])

def reduce_intersect(table_column, sep=','):
    from functools import reduce
    return reduce(set.intersection, 
                  table_column.to_pandas().str.split(sep).map(set))

with open_csv(data_file, read_options, parse_options, convert_options) as file:
    common = {}    # collect here common items along columns
    batch = file.read_next_batch()
    for col in batch.schema.names:
        common[col] = reduce_intersect(batch[col])
    if any(common.values()):
        for batch in file:
            for col in (col for col, data in common.items() if data):
                common[col].intersection_update(reduce_intersect(batch[col]))
            if not any(common.values()):
                break