Организация конвейеров данных в библиотеке Pandas
Организация потоков данных в Pandas
При работе с табличными данными часто требуется выполнить последовательность преобразований: фильтрацию, добавление столбцов, группировку, сортировку. Потоковый подход (chaining, piping) позволяет записать эти шаги компактно и читаемо, избегая создания множества промежуточных переменных. Рассмотрим несколько способов построения таких цепочек.
Основной подход: метод .pipe() для создания модульных потоков
Как сделать цепочку преобразований, которую легко поддерживать и переиспользовать?
Метод .pipe() позволяет передать DataFrame в пользовательскую функцию и вернуть результат, встраивая её в цепочку вызовов. Это удобно, когда шаги обработки вынесены в отдельные функции.
import pandas as pd
df = pd.DataFrame({'x': [1,2,3], 'y': [4,5,6]})
def add_z(df, val):
df['z'] = df['x'] + df['y'] + val
return df
def filter_gt(df, col, thresh):
return df[df[col] > thresh]
result = (df
.pipe(add_z, val=10)
.pipe(filter_gt, col='z', thresh=12)
)
print(result)Column values python (значения столбца в python (pandas))
x y z 1 2 5 17
Python наборы данных (наборы данных в python)
Пояснение: каждая функция принимает DataFrame как первый аргумент, а дополнительные параметры передаются через .pipe(). Цепочка выполняется последовательно, результат последней функции становится итогом.
Типичные ошибки:
- Изменение исходного DataFrame внутри функции (используйте копию, если нужна сохранность оригинала).
- Забывание возвращать DataFrame – без return pipe вернёт None.
- Слишком длинные цепочки затрудняют отладку – разбивайте на логические блоки с комментариями.
Вариант 1: Простое chaining без пользовательских функций
Как объединить несколько встроенных методов в одну строку без pipe?
Многие методы Pandas возвращают DataFrame, поэтому их можно вызывать последовательно:
result = (df
.query('x > 1')
.assign(w = lambda d: d['x'] * 2)
.sort_values('w', ascending=False)
)обработка больших данных python (обработка больших данных в python)
Это просто и быстро для типовых операций. Проблема: если понадобится вставить отладочный вывод или логику, потребуется разорвать цепочку.
Частая ошибка – использование .assign() с мутирующей операцией (например, df['x'] = ...) внутри лямбды, что может привести к неожиданным результатам из-за копирования.
Вариант 2: Поток с .assign() и .query() для фильтрации и создания столбцов
Как одновременно создавать новые столбцы и фильтровать данные в одном потоке?
Методы .assign() (добавление столбцов) и .query() (фильтрация) идеально сочетаются:
result = (df
.assign(ratio = lambda d: d['x'] / d['y'])
.query('ratio > 0.5')
)работа с dataframe python (работа с dataframe в python)
Цель: быстрая обработка без создания временных переменных. Случаи использования – разведочный анализ, подготовка данных для визуализации.
Ошибка: .query() использует свой синтаксис (например, @variable для внешних переменных). Забывание @ приводит к ошибке NameError.
Вариант 3: Потоковая обработка больших данных по частям (chunking)
Как обрабатывать файлы, которые не помещаются в память, используя итеративный поток?
При чтении CSV с параметром chunksize возвращается итератор по частям DataFrame. Каждый chunk можно обработать в цепочке и накопить результат:
chunks = pd.read_csv('large.csv', chunksize=10000)
results = []
for chunk in chunks:
processed = (chunk
.query('value > 0')
.assign(doubled = lambda d: d['value'] * 2)
)
results.append(processed.sum())
final = pd.concat(results).to_frame().TЦель: работа с данными, не влезающими в RAM. Проблема: агрегация требует аккуратного слияния частичных результатов.
Типичная ошибка – забыть, что chunk – это копия, и попытка изменить оригинальный файл через него не сработает.
Расширенные примеры потоковой обработки с Pandas
Пример 1: Многоступенчатый пайплайн с .pipe() и передачей нескольких аргументов
import pandas as pd
import numpy as np
df = pd.DataFrame({'x': np.random.rand(5), 'y': np.random.rand(5)})
def scale(df, col, factor):
df[col] = df[col] * factor
return df
def clip_and_label(df, col, low, high, label):
df['label'] = np.where((df[col] >= low) & (df[col] <= high), label, 'other')
return df
result = (df
.pipe(scale, col='x', factor=100)
.pipe(clip_and_label, col='x', low=20, high=80, label='mid')
.query('label == "mid"')
)
print(result)x y label 3 51.278... 0.123... mid
Пояснение: каждая функция модифицирует DataFrame и возвращает его. Такой пайплайн легко тестировать по шагам.
Пример 2: Использование .pipe() с lambda для inline-преобразований
result = (df
.pipe(lambda d: d.assign(z = d['x'] + d['y']))
.pipe(lambda d: d[d['z'] > d['z'].median()])
)
print(result)Lambda позволяет выполнить короткое преобразование без объявления отдельной функции. Проблема: сложность отладки.
Пример 3: Цепочка с .groupby() и .agg() – потоковая агрегация
df = pd.DataFrame({'group': ['A','A','B'], 'val': [1,2,3]})
result = (df
.groupby('group')
.agg(mean_val=('val', 'mean'), sum_val=('val', 'sum'))
.reset_index()
.assign(ratio = lambda d: d['sum_val'] / d['mean_val'])
)
print(result)group mean_val sum_val ratio 0 A 1.5 3 2.0 1 B 3.0 3 1.0
Здесь groupby возвращает GroupBy объект, который не поддерживает chaining с assign напрямую, поэтому сначала агрегируем, затем сбрасываем индекс.
Пример 4: Комбинирование chunked чтения с агрегацией по ключу
import pandas as pd
from io import StringIO
# Имитация большого CSV
data = """key,val
A,10
B,20
A,30
B,40
"""
chunks = pd.read_csv(StringIO(data), chunksize=2)
total = {}
for chunk in chunks:
grouped = chunk.groupby('key')['val'].sum()
for key, val in grouped.items():
total[key] = total.get(key, 0) + val
print(total){'A': 40, 'B': 60}
Пояснение: каждый chunk группируется отдельно, результаты суммируются в словаре. Это эффективно при больших объёмах.
Пример 5: Поток с .eval() для быстрых вычислений без копирования
df = pd.DataFrame({'a': [1,2], 'b': [3,4]})
result = df.eval('c = a + b')
print(result)a b c 0 1 3 4 1 2 4 6
.eval() использует NumExpr и работает быстрее assign для сложных формул. Однако он не поддерживает прямую фильтрацию – для этого combine с .query().