适用于高性能数据管道的 5 种强大 Python 装饰器

import pandas as pd import numpy as np # 加载数据集 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.")

from numba import njit import time # 提取一个数值列作为 NumPy 数组以进行快速处理 incomes = df_pipeline['median_income'].fillna(0).values @njit def compute_complex_metric(income_array): result = np.zeros_like(income_array) # 在纯 Python 中，这样的循环通常会拖慢速度 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!")

from joblib import Memory import time # 为管道构件创建一个本地缓存目录 memory = Memory(".pipeline_cache", verbose=0) @memory.cache def expensive_aggregation(df): print("Running heavy grouping operation...") time.sleep(1.5) # 模拟长时间运行的管道步骤 # 按 ocean_proximity 对数据点进行分组并计算属性级均值 return df.groupby('ocean_proximity', as_index=False).mean(numeric_only=True) # 第一次运行执行代码；第二次从磁盘加载以实现即时加载 agg_df = expensive_aggregation(df_pipeline) agg_df_cached = expensive_aggregation(df_pipeline)

import pandera as pa import pandas as pd import numpy as np from dask import delayed, compute # 定义一个模式以强制执行数据类型和有效范围 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: """ 验证数据帧块是否符合定义的模式。如果数据损坏，Pandera 会引发 SchemaError。 """ return housing_schema.validate(df) # 将管道数据分成 4 个块以进行并行验证 chunks = np.array_split(df_pipeline, 4) lazy_validations = [validate_and_process(chunk) for chunk in chunks] print("Starting parallel schema validation...") try: # 触发 Dask 图以并行验证块 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}")

from dask import delayed, compute @delayed def process_chunk(df_chunk): # 模拟独立的转换任务 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 # 将数据集分成 4 个并行处理的块 chunks = np.array_split(df_pipeline, 4) # 懒惰计算图（Dask 的工作方式！） lazy_results = [process_chunk(chunk) for chunk in chunks] # 同时跨多个 CPU 触发执行 processed_chunks = compute(*lazy_results) df_parallel = pd.concat(processed_chunks) print(f"Parallelized output shape: {df_parallel.shape}")

from memory_profiler import profile # 一个装饰函数，它会向控制台打印逐行内存细分 @profile(precision=2) def memory_intensive_step(df): print("Running memory diagnostics...") # 创建一个巨大的临时副本以引起故意内存峰值 df_temp = df.copy() df_temp['new_col'] = df_temp['total_bedrooms'] * 100 # 删除临时数据帧会释放 RAM del df_temp return df.dropna(subset=['total_bedrooms']) # 运行管道步骤：您可能会在终端中看到内存报告 final_df = memory_intensive_step(df_pipeline)