# 内存优化技巧（处理大型数据集）
orders = orders.astype({'customer_id':'uint32', 'amount':'float32'})
customers = customers.convert_dtypes()  # 自动选择合适类型

# 分布式处理模式（Dask集成示例）
import dask.dataframe as dd
ddf_orders = dd.from_pandas(orders, npartitions=4)
ddf_customers = dd.from_pandas(customers, npartitions=2)

merged_ddf = dd.merge(
    ddf_orders, 
    ddf_customers,
    on='customer_id',
    how='left',
    shuffle='tasks'  # 智能分区策略
)

merged_complex = pd.merge(
    orders,
    customer_levels,  # 客户等级表
    left_on=['customer_id', pd.Grouper(key='order_date', freq='M')],
    right_on=['client_id', 'month'],
    suffixes=('_order', '_level')
)

# 内存映射优化（处理超大数据文件）
with pd.HDFStore('transactions.h5') as store:
    chunks = [store[f'chunk_{i}'] for i in range(10)]
    concat_data = pd.concat(chunks, axis=0, ignore_index=True, copy=False)

# 高级填充示例（使用特征相关性）
corr_matrix = orders.corr()
high_corr_feature = corr_matrix['amount'].idxmax()
orders['amount'] = orders['amount'].fillna(
    orders.groupby(high_corr_feature)['amount'].transform('median')
)

def dynamic_threshold(df, col, sensitivity=0.2):
    q25 = df[col].quantile(0.25)
    q75 = df[col].quantile(0.75)
    iqr = q75 - q25
    return {
        'lower': q25 - (1.5 + sensitivity) * iqr,
        'upper': q75 + (1.5 + sensitivity) * iqr
    }

from sklearn.preprocessing import RobustScaler
from joblib import Parallel, delayed

def parallel_scale(df_chunk):
    scaler = RobustScaler()
    return pd.DataFrame(scaler.fit_transform(df_chunk),
                        columns=df_chunk.columns)

# 分块处理10GB级数据
scaled_data = Parallel(n_jobs=4)(
    delayed(parallel_scale)(chunk)
    for chunk in np.array_split(big_data, 8)
)

# 保留原始分布信息
orders['amount_scaled'] = orders['amount'].pipe(
    lambda x: (x - x.mean()) / x.std()
).add_prefix('zscore_')

orders['amount_original_ratio'] = orders['amount'] / orders['amount'].max()

class DataQualityMonitor:  
    def __init__(self, df):  
        self.df = df.copy()  
        self.metrics = {}  

    def generate_report(self):  
        self._check_completeness()  
        self._check_consistency()  
        self._check_anomalies()  
        return pd.DataFrame(self.metrics).T  

    def _check_completeness(self):  
        self.metrics['missing_rate'] = self.df.isna().mean()  
        self.metrics['zero_rate'] = (self.df == 0).mean()

    def _check_consistency(self):  
        self.metrics['id_duplicates'] = self.df.duplicated(subset='id').sum()  
        self.metrics['date_range'] = {  
            'start': self.df['date'].min(),  
            'end': self.df['date'].max()  
        }  

    def _check_anomalies(self):  
        for col in ['amount', 'quantity']:  
            stats = self.df[col].describe()  
            self.metrics[f'{col}_anomaly'] = {  
                'outliers': self.df[col].between(  
                    stats['25%'] - 1.5*(stats['75%']-stats['25%']),  
                    stats['75%'] + 1.5*(stats['75%']-stats['25%'])  
                ).sum()  
            }  

# 生成HTML可视化报告  
monitor = DataQualityMonitor(orders)  
report = monitor.generate_report()  
report.style.background_gradient(cmap='viridis').to_html('data_quality_report.html')

| 特性              | 1.x版本               | 2.x最佳实践           |
|-------------------|-----------------------|----------------------|
| 空值处理          | `NaN`                 | 强制使用`pd.NA`      |
| 字符串处理        | `object`类型          | 专用`StringDtype`    |
| 类型推断          | 手动指定              | `convert_dtypes()`   |
| 性能优化          | 单线程处理            | 支持PyArrow引擎      |

# 类型优化矩阵  
dtype_map = {  
    'int64': 'Int32',          # 空值兼容整型  
    'float64': 'Float32',  
    'object': 'string[pyarrow]' # 使用Arrow字符串  
}  

optimized_df = df.astype(dtype_map)  
print(f"内存节省：{(1 - optimized_df.memory_usage().sum() / df.memory_usage().sum()):.1%}")

# 使用Pandas实现漏斗分析  
funnel_steps = ['login', 'browse', 'add_cart', 'checkout']  
funnel_data = (  
    user_events  
    .assign(step=pd.Categorical(user_events['event_type'], categories=funnel_steps))  
    .groupby(['user_id', 'step'])['timestamp'].min()  
    .unstack()  
    .pipe(lambda df: df[funnel_steps])  # 强制顺序  
)  

# 计算转化率  
conversion_rates = {}  
for i in range(len(funnel_steps)-1):  
    rate = funnel_data[funnel_steps[i+1]].notna().sum() / funnel_data[funnel_steps[i]].notna().sum()  
    conversion_rates[f'{funnel_steps[i]}_to_{funnel_steps[i+1]}'] = f"{rate:.1%}"