Pandas高级数据处理：数据报告生成实战指南

# 安全列名检查方案
valid_columns = [col for col in ['销售日期', '销售额'] if col in df.columns]
clean_df = df[valid_columns].copy()

# 分块处理大型数据集
chunk_size = 10**6
aggregator = defaultdict(lambda: {'sum': 0, 'count': 0})

for chunk in pd.read_csv('big_data.csv', chunksize=chunk_size):
    chunk['category'] = chunk['category'].astype('category')  # 内存占用减少70%
    for key, grp in chunk.groupby('category'):
        aggregator[key]['sum'] += grp['value'].sum()
        aggregator[key]['count'] += len(grp)

# 安全绘图模板
def safe_plot(df, x_col, y_col):
    if not all(col in df.columns for col in [x_col, y_col]):
        print(f"缺少{x_col}或{y_col}列")
        return
    
    valid_data = df[[x_col, y_col]].dropna()
    if len(valid_data) < 2:
        print("有效数据不足")
        return
    
    plt.figure(figsize=(12,6))
    sns.lineplot(data=valid_data, x=x_col, y=y_col)
    plt.xticks(rotation=45)
    plt.tight_layout()

class ReportGenerator:
    def __init__(self, data_source):
        self.raw_data = self._load_data(data_source)
        self.clean_data = None
        self.analysis_results = {}
        
    def _load_data(self, source):
        # 支持多种数据源加载
        if source.endswith('.parquet'):
            return pd.read_parquet(source)
        elif source.endswith('.csv'):
            return pd.read_csv(source, low_memory=False)
        
    def data_pipeline(self):
        # 数据清洗流水线
        self.clean_data = (
            self.raw_data
            .pipe(self._handle_missing)
            .pipe(self._convert_types)
            .pipe(self._filter_outliers)
        )
        
    def generate_report(self, output_format='html'):
        # 多格式输出支持
        if output_format == 'html':
            return self._generate_html()
        elif output_format == 'pdf':
            return self._generate_pdf()

def safe_aggregation(df, group_col, agg_col):
    try:
        return df.groupby(group_col)[agg_col].agg(['mean', 'sum'])
    except KeyError as e:
        print(f"关键列缺失: {e}")
        return pd.DataFrame()
    except TypeError as e:
        print(f"类型错误: {e}")
        return df.groupby(group_col).apply(lambda x: x.select_dtypes(include='number').sum())

from pandarallel import pandarallel
pandarallel.initialize()

df.groupby('category').parallel_apply(complex_calculation)  # 提速4-8倍