Эффективная фильтрация строк, разделенных запятыми, в pandas/dask

Основываясь на вашем образце данных, я склоняюсь к использованию Regex во входном файле, прежде чем загружать его в фрейм данных pandas, чтобы изменить разделитель между столбцами (это просто, эффективно и отслеживает индексы).

ПРИМЕЧАНИЕ. Этот ответ отвечает на ваши первые два вопроса, третий, я думаю, можно было бы решить с помощью .loc или логического индексирования, но не уверен, так как я плохо его понял.

Опция 1

• Откройте входной файл с помощью текстовой программы Sublime или любого текстового редактора, поддерживающего механизм Regex.
• нажмите Ctrl+h, чтобы открыть поиск и замену, затем введите этот шаблон в раздел поиска (\w+),(.*),(\d{2}-\d{2}-\d{4}) и этот шаблон в раздел замены \1;\2;\3. Шаблон поиска будет соответствовать каждой строке ваших данных и изменит разделитель между данными ваших столбцов; строка ваших данных будет выглядеть так:
  MyName;1,2,3,4,5,6,7,8,9,10;19-04-2024

ПРИМЕЧАНИЕ. Обязательно вручную измените разделитель на ; между именами столбцов, а затем сохраните входной файл.
Ваши данные будут выглядеть следующим образом:

Name;Signal;Date
MyName;1,2,3,4,5,6,7,8,9,10;19-04-2024
MyName;1,2,3,4,5,6,7,8,9,10;19-04-2024

• с помощью простого кода pandas вы можете получить сумму сигналов, а затем отфильтровать их так, как хотите.

Вот пример кода:

temp = pd.read_csv('my_csv copy.csv', sep = ";")
df = (
    temp
    .assign(
        summation = lambda df_: df_.Signal.str.split(',').apply(lambda x: sum([int(i) for i in x]))
    )
)

у вас будет такой вывод:

  Name                Signal        Date     summation
0  MyName  1,2,3,4,5,6,7,8,9,10  19-04-2024         55
1  MyName  1,2,3,4,5,6,7,8,9,10  19-04-2024         55

Вариант 2

Вы можете открыть файл CSV, прочитать его построчно и добавить строки в фрейм данных pandas.

ПРИМЕЧАНИЕ. Это может быть неэффективно для больших файлов данных, но, поскольку я не знаю размера ваших данных, я подумал, что стоит протестировать.
Вот пример кода:

df = pd.read_csv('my_csv.csv', dtype = {"Signal" : "string"})
# read csv file line by line
output_df = pd.DataFrame(columns=['index','name','Signal', 'Time'])
i = 0
with open('my_csv.csv', 'r') as f:
    line = f.readline()
    for line in f:
        my_list = line.strip().split(',')
        singals =[int(x) for x in my_list[1:-1]]
        summation = sum(singals)
        output_df = pd.concat([
            output_df,
            pd.DataFrame([[i, my_list[0], singals,summation, my_list[-1]]], columns=['index','name','Signal', 'summation','Time'])
        ])
        i+=1
        
output_df = output_df.assign(Time = pd.to_datetime(output_df['Time'], format='%d-%m-%Y'))
output_df

вывод приведенного выше кода должен быть:

index    name                           Signal       Time  summation
0      MyName  [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 2024-04-19       55.0
1      MyName  [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] 2024-04-19       55.0

Надеюсь, это поможет!

Прочитав свой фрейм данных, вы можете преобразовать значения Signal в список целых чисел, используя этот код:

df['Signal'] = df['Signal'].apply(lambda s:list(map(int, s.split(','))))

Выход:

     Name                           Signal        Date
0  MyName  [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]  19-04-2024
1  MyName  [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]  19-04-2024

Затем вы можете поиграть со своими значениями, чтобы отфильтровать фрейм данных. Например, чтобы фильтровать по суммам:

sums = df['Signal'].apply(sum)
# 0    55
# 1    55
# Name: Signal, dtype: int64

mask = sums == 100
# 0    False
# 1    False
# Name: Signal, dtype: bool

df_filtered = df[mask]

Если вы хотите использовать суммы более одного раза, вероятно, лучше сохранить их в фрейме данных, например.

df['sum'] = df['Signal'].apply(sum)

Вариантом может быть расширение столбца «Сигнал» в фрейм данных и вычисление суммы по строкам. Затем используйте результат для фильтрации исходного фрейма данных.

sums = df['Signal'].str.split(',', expand=True).astype(float).sum(axis=1)
df = df[sums==100].reset_index(drop=True)

Поскольку целью является фильтрация, другой вариант — read_csv прочитать определенные столбцы и пропустить строки; это немного многословно, но ввод-вывод файлов Pandas невероятно быстр.

Идея состоит в том, чтобы прочитать только столбец «Сигнал», обработать его так, как если бы он находился в самом файле csv; затем прочитайте это в фрейме данных (signals_df ниже) и выполните суммирование по строкам. Затем снова прочитайте все столбцы CSV, только на этот раз пропустите строки, где сумма не равна 100, т. е. выберите только строки, где сумма равна 100.

signals = pd.read_csv('my_csv.csv', usecols=['Signal']).values.ravel().tolist()
signals_df = pd.read_csv(io.StringIO("\n".join(signals)), header=None)
total_signals = signals_df.sum(axis=1)
# +1 is to account for the header
rows_to_skip = total_signals.index[total_signals!=100] + 1
df = pd.read_csv('my_csv.csv', skiprows=rows_to_skip)

Вот мой первый подход. Я тестировал это на файле размером около 10 миллионов строк, который содержал от 1 до 25 случайных целых чисел со строкой значений в двойных кавычках, разделенных запятыми, в диапазоне от 0 до 99. Размер файла составлял ~ 650 МБ. Я использовал pandarallel, чтобы распараллелить приложение.

Возможно, у вас не установлен pandarallel. Это хороший пакет, который можно быстро установить с помощью pip install pandarallel (более подробную информацию см. на github или в этом посте stackoverflow).

На моей машине следующее заняло ~11 секунд с 10 ядрами.

import pandas as pd
from pandarallel import pandarallel
import numpy as np

df = pd.read_csv('data.csv',parse_dates=True)
pandarallel.initialize(progress_bar=True)

df['condition'] = df.parallel_apply(
  lambda x: np.array(x.Signal.split(','),dtype=np.int_).sum() == 100,
  axis=1,
)
df_where_true = df[df['condition']]

Новый столбец DataFrame condition позволяет быстро маскировать, отвечая на ваш первый вопрос.

Что касается вашего второго вопроса, то, вероятно, существуют способы дальнейшего ускорения подхода, но этот случай достаточно прост, и параллелизм может дать вам достаточную производительность. Одна из проблем, которая здесь рассматривается, заключается в том, что если ваши данные Signal потенциально имеют разную длину, т. е. некоторые строки имеют только 3 значения, тогда как другие имеют 10, то преобразование всех данных в двумерный массив numpy невозможно без заполнения нулями (массивы numpy требуют единообразных форм). Решение, представленное выше, обрабатывает сигналы разной длины.

Что касается вашего последнего вопроса, реализация dask может выглядеть так:

import dask.dataframe as dd
...
df['condition'] = df.apply(
  lambda x: np.array(x.Signal.split(',')).astype(int).sum() == 100,
  axis=1,
  meta=bool,
).compute()
...

Однако на моей рабочей станции это не было автоматически распараллелено, и в результате это заняло ~110 секунд.

В этом случае, я думаю, есть несколько вариантов, которые вы можете сделать:

1. Используя встроенные возможности Pandas, вы можете эффективно записывать индекс и удалять строки. Чтобы выбрать строки в зависимости от суммы массива в столбце «Сигнал», вы можете использовать логическое индексирование, а не перебирать их. После того как отфильтрованные строки будут идентифицированы, вы можете удалить их из DataFrame, записав их индекс.
2. Вы можете использовать массивы NumPy для работы непосредственно со столбцом «Сигнал» после его преобразования.
3. Массивы NumPy по-прежнему можно использовать для эффективной обработки при интеграции с Dask. С помощью массивов Dask вы можете быстро и эффективно анализировать большие наборы данных без необходимости использования согласованного интерфейса массивов NumPy.

Ниже приведен пример кода, которому вы можете следовать, чтобы реализовать вышеуказанный вариант:

import pandas as pd
import numpy as np

# Read CSV file
df = pd.read_csv("my_csv.csv", parse_dates=["Date"])

# Convert "Signal" column to arrays of numpy
df["Signal"] = df["Signal"].apply(lambda x: np.array([int(i) for i in x.split(",")]))

# this section will filter rows based on sum of "Signal" column inside the array
target = 100
mask_filter = df["Signal"].apply(lambda x: x.sum() == target)

# filtered data
filtered_datas = df.index[mask_filter]

# Drop filtered rows from DataFrame
filtered_df = df.drop(filtered_datas )

# If you want to keep only filtered rows:
# filtered_df = df.loc[filtered_datas ]

print(filtered_df)

Использование объектов Dask DataFrame и Array вместо той же логики будет необходимо при работе с большими наборами данных. Dask хорошо справится с распараллеливанием и распределенными вычислениями. Если ваш набор данных слишком велик и не помещается в памяти, Dask — отличный выбор.

Если вам просто нужен результирующий фрейм данных, содержащий только строки, в которых сумма Signal равна некоторому значению (например, 100), вы можете использовать поляры для чтения данных.

import polars as pl

df = (
    pl.read_csv('path/to/file.csv')
      .with_columns(pl.col('Signal').str.split(',').cast(pl.List(pl.Int64)))
      .filter(pl.col('Signal').list.sum() == 100)
)

Сравнивая это с решением pandas, использующим аргумент converters в read_csv:

import pandas as pd

df_all = pd.read_csv(
    'path/to/file.csv', 
    converters = {'Signal': lambda s: list( map(int, s.split(',')) )}
)
ix = df_all.Signal.apply(sum).eq(100)
df = df_all.loc[ix].reset_index(drop=True)

Вот время для 1 000 000 строк:

import random
import io

data = ['Name,Signal,Date']
for i in range(1_000_000):
    n = ''.join(random.choices('abcdefghijklmnopqrstuvwxyz', k=8))
    s = ','.join(map(str, (random.randint(0,10) for _ in range(15))))
    d = '19-04-2024'
    data.append(f'{n},"{s}",{d}')

data_str = '\n'.join(data)

In [22]: %%timeit
    ...: df = (
    ...:     pl.read_csv(io.StringIO(data_str))
    ...:       .with_columns(pl.col('Signal').str.split(',').cast(pl.List(pl.Int64)))
    ...:       .filter(pl.col('Signal').list.sum() == 100)
    ...: )
    ...:
    ...:
635 ms ± 12.3 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)

In [23]: %%timeit
    ...: df_all = pd.read_csv(
    ...:     io.StringIO(data_str),
    ...:     converters = {'Signal': lambda s: list( map(int, s.split(',')) )}
    ...: )
    ...: ix = df_all.Signal.apply(sum).eq(100)
    ...: df = df_all.loc[ix].reset_index(drop=True)
    ...:
    ...:
2.06 s ± 53.8 ms per loop (mean ± std. dev. of 7 runs, 1 loop each)