Введение
Пайплайны данных позволяют автоматизировать обработку в проектах data science и машинного обучения. Код иногда перегружает основную логику, но декораторы Python упрощают задачу. Здесь описаны пять эффективных декораторов для построения и улучшения быстрых пайплайнов.
Перед демонстрацией примеров загрузим версию датасета California Housing из публичного репозитория на GitHub:
import pandas as pd
import numpy as np
# Loading the dataset
DATA_URL = "https://raw.githubusercontent.com/gakudo-ai/open-datasets/main/housing.csv"
print("Downloading data pipeline source...")
df_pipeline = pd.read_csv(DATA_URL)
print(f"Loaded {df_pipeline.shape[0]} rows and {df_pipeline.shape[1]} columns.")
1. Компиляция JIT
Циклы в Python славились медлительностью при сложных математических преобразованиях на датасетах, но есть простой выход. Декоратор @njit из библиотеки Numba преобразует функции Python в оптимизированный машинный код, похожий на C, прямо во время выполнения. На больших объемах данных и в сложных пайплайнах это дает серьезный прирост скорости.
from numba import njit
import time
# Extracting a numeric column as a NumPy array for fast processing
incomes = df_pipeline['median_income'].fillna(0).values
@njit
def compute_complex_metric(income_array):
result = np.zeros_like(income_array)
# In pure Python, a loop like this would normally drag
for i in range(len(income_array)):
result[i] = np.log1p(income_array[i] * 2.5) ** 1.5
return result
start = time.time()
df_pipeline['income_metric'] = compute_complex_metric(incomes)
print(f"Processed array in {time.time() - start:.5f} seconds!")
2. Кэширование промежуточных результатов
Если пайплайн включает тяжелые агрегации или объединения данных, занимающие минуты или часы, декоратор memory.cache сохраняет выводы функций на диск. При перезапуске скрипта или после сбоя он загружает готовые массивы, пропуская вычисления и экономя ресурсы с временем.
from joblib import Memory
import time
# Creating a local cache directory for pipeline artifacts
memory = Memory(".pipeline_cache", verbose=0)
@memory.cache
def expensive_aggregation(df):
print("Running heavy grouping operation...")
time.sleep(1.5) # Long-running pipeline step simulation
# Grouping data points by ocean_proximity and calculating attribute-level means
return df.groupby('ocean_proximity', as_index=False).mean(numeric_only=True)
# The first run executes the code; the second resorts to disk for instant loading
agg_df = expensive_aggregation(df_pipeline)
agg_df_cached = expensive_aggregation(df_pipeline)
3. Проверка схемы данных
Библиотека Pandera занимается проверкой схем для предотвращения скрытого ухудшения качества данных, которое портит модели машинного обучения или дашборды. В примере ниже она сочетается с параллельной библиотекой Dask, чтобы убедиться, что входные данные пайплайна соответствуют схеме. Несоответствие вызовет ошибку на раннем этапе.
import pandera as pa
import pandas as pd
import numpy as np
from dask import delayed, compute
# Define a schema to enforce data types and valid ranges
housing_schema = pa.DataFrameSchema({
"median_income": pa.Column(float, pa.Check.greater_than(0)),
"total_rooms": pa.Column(float, pa.Check.gt(0)),
"ocean_proximity": pa.Column(str, pa.Check.isin(['NEAR BAY', '<1H OCEAN', 'INLAND', 'NEAR OCEAN', 'ISLAND']))
})
@delayed
@pa.check_types
def validate_and_process(df: pa.typing.DataFrame) -> pa.typing.DataFrame:
"""
Validates the dataframe chunk against the defined schema.
If the data is corrupt, Pandera raises a SchemaError.
"""
return housing_schema.validate(df)
# Splitting the pipeline data into 4 chunks for parallel validation
chunks = np.array_split(df_pipeline, 4)
lazy_validations = [validate_and_process(chunk) for chunk in chunks]
print("Starting parallel schema validation...")
try:
# Triggering the Dask graph to validate chunks in parallel
validated_chunks = compute(*lazy_validations)
df_parallel = pd.concat(validated_chunks)
print(f"Validation successful. Processed {len(df_parallel)} rows.")
except pa.errors.SchemaError as e:
print(f"Data Integrity Error: {e}")
4. Ленивая параллелизация
Последовательное выполнение независимых шагов пайплайна не использует процессоры по максимуму. Декоратор @delayed строит граф зависимостей для таких преобразований, а потом запускает задачи параллельно оптимальным образом, сокращая общее время.
from dask import delayed, compute
@delayed
def process_chunk(df_chunk):
# Simulating an isolated transformation task
df_chunk_copy = df_chunk.copy()
df_chunk_copy['value_per_room'] = df_chunk_copy['median_house_value'] / df_chunk_copy['total_rooms']
return df_chunk_copy
# Splitting the dataset into 4 chunks processed in parallel
chunks = np.array_split(df_pipeline, 4)
# Lazy computation graph (the way Dask works!)
lazy_results = [process_chunk(chunk) for chunk in chunks]
# Trigger execution across multiple CPUs simultaneously
processed_chunks = compute(*lazy_results)
df_parallel = pd.concat(processed_chunks)
print(f"Parallelized output shape: {df_parallel.shape}")
5. Профилирование памяти
Декоратор @profile выявляет утечки памяти, которые крашат серверы на огромных файлах. Он отслеживает функцию поэтапно, показывая потребление RAM на каждом шаге. Так проще находить неэффективности и оптимизировать использование памяти.
from memory_profiler import profile
# A decorated function that prints a line-by-line memory breakdown to the console
@profile(precision=2)
def memory_intensive_step(df):
print("Running memory diagnostics...")
# Creation of a massive temporary copy to cause an intentional memory spike
df_temp = df.copy()
df_temp['new_col'] = df_temp['total_bedrooms'] * 100
# Dropping the temporary dataframe frees up the RAM
del df_temp
return df.dropna(subset=['total_bedrooms'])
# Running the pipeline step: you may observe the memory report in your terminal
final_df = memory_intensive_step(df_pipeline)
Итоги
Здесь представлены пять полезных декораторов Python для оптимизации ресурсоемких пайплайнов данных. Благодаря библиотекам вроде Dask и Numba они ускоряют преобразования данных и повышают устойчивость к сбоям.