analytics-data-processing

Analytics Data Processing

Task Objectives

This Skill is used to: master the complete processing workflow from data acquisition to modeling
Capabilities include: data acquisition, data cleaning, data transformation, data modeling
Trigger conditions: when analytical data needs to be prepared or data models need to be built

Data Acquisition

Data Source Types

Type	Source	Acquisition Method
Business Systems	CRM, ERP, e-commerce, etc.	Direct database connection, API
User Behavior	APP, Web, Mini Program	Tracking data, logs
External Data	Third-party, public data	File import, web scraping
Real-time Data	IoT, sensors	Streaming collection

Data Acquisition Modes

Batch Acquisition

import pandas as pd
from sqlalchemy import create_engine

engine = create_engine('postgresql://user:pass@host:5432/db')

df = pd.read_sql("""
    SELECT *
    FROM orders
    WHERE created_at >= '2024-01-01'
""", engine)

API Acquisition

import requests
import pandas as pd

response = requests.get(
    'https://api.example.com/data',
    params={
        'start_date': '2024-01-01',
        'end_date': '2024-12-31',
        'page': 1,
        'page_size': 1000
    },
    headers={'Authorization': 'Bearer xxx'}
)

data = response.json()['data']
df = pd.DataFrame(data)

File Import

import pandas as pd

df_csv = pd.read_csv('data.csv', encoding='utf-8')
df_excel = pd.read_excel('data.xlsx', sheet_name='Sheet1')
df_json = pd.read_json('data.json', orient='records')

Data Quality Assessment

def assess_data_quality(df):
    report = {
        'total_rows': len(df),
        'total_cols': len(df.columns),
        'missing_values': df.isnull().sum().to_dict(),
        'missing_rate': (df.isnull().sum() / len(df) * 100).to_dict(),
        'duplicates': df.duplicated().sum(),
        'dtypes': df.dtypes.to_dict()
    }
    return report

Data Cleaning

Missing Value Handling

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'A': [1, 2, np.nan, 4],
    'B': [5, np.nan, np.nan, 8],
    'C': ['a', 'b', 'c', 'd']
})

df_dropna = df.dropna()

df_fill_mean = df.fillna(df.mean())

df_fill_forward = df.fillna(method='ffill')

df_interpolate = df.interpolate()

Outlier Handling

import pandas as pd
import numpy as np

def handle_outliers(df, column, method='iqr'):
    if method == 'iqr':
        Q1 = df[column].quantile(0.25)
        Q3 = df[column].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
        return df[(df[column] >= lower) & (df[column] <= upper)]
    elif method == 'zscore':
        z_scores = np.abs((df[column] - df[column].mean()) / df[column].std())
        return df[z_scores < 3]

df_clean = handle_outliers(df, 'amount', method='iqr')

Data Type Conversion

df['date'] = pd.to_datetime(df['date'])
df['price'] = df['price'].astype(float)
df['category'] = df['category'].astype('category')
df['flag'] = df['flag'].map({'Y': 1, 'N': 0})

Duplicate Value Handling

df_dedup = df.drop_duplicates()
df_dedup_subset = df.drop_duplicates(subset=['order_id', 'product_id'])

Data Transformation

Aggregate Statistics

sales_summary = df.groupby(['year', 'month', 'category'])['amount'].agg([
    ('销量', 'sum'),
    ('笔数', 'count'),
    ('均价', 'mean'),
    ('最大值', 'max'),
    ('最小值', 'min')
]).reset_index()

Pivot Tables

pivot = pd.pivot_table(
    df,
    values='amount',
    index=['year', 'month'],
    columns='category',
    aggfunc='sum',
    fill_value=0
)

Time Series Transformation

df.set_index('date', inplace=True)

df_daily = df.resample('D').sum()
df_monthly = df.resample('M').sum()
df_quarterly = df.resample('Q').sum()

df_rolling = df['amount'].rolling(window=7).mean()
df_cumulative = df['amount'].expanding().sum()

Data Normalization

from sklearn.preprocessing import MinMaxScaler, StandardScaler

scaler = MinMaxScaler()
df_normalized = pd.DataFrame(
    scaler.fit_transform(df[['amount', 'quantity']]),
    columns=['amount_norm', 'quantity_norm']
)

scaler = StandardScaler()
df_standardized = pd.DataFrame(
    scaler.fit_transform(df[['amount', 'quantity']]),
    columns=['amount_std', 'quantity_std']
)

Data Modeling

Dimension Table Design

dim_user = pd.DataFrame({
    'user_id': [1, 2, 3],
    'user_name': ['张三', '李四', '王五'],
    'user_type': ['VIP', '普通', 'VIP'],
    'join_date': ['2023-01-01', '2023-06-15', '2023-03-20'],
    'age_group': ['25-30', '31-35', '26-30']
})

Fact Table Design

fact_orders = pd.DataFrame({
    'order_id': [1001, 1002, 1003],
    'user_id': [1, 2, 1],
    'product_id': [101, 102, 103],
    'order_date': ['2024-01-15', '2024-01-16', '2024-01-18'],
    'amount': [299, 599, 1299],
    'quantity': [1, 2, 1],
    'discount': [0, 50, 100]
})

fact_orders['order_date'] = pd.to_datetime(fact_orders['order_date'])

Metric Calculation

fact_orders['unit_price'] = fact_orders['amount'] / fact_orders['quantity']

fact_orders['is_vip_order'] = fact_orders['user_id'].map(
    dim_user.set_index('user_id')['user_type'] == 'VIP'
)

fact_orders['order_month'] = fact_orders['order_date'].dt.to_period('M')

Hierarchical Aggregation

daily_summary = fact_orders.groupby('order_date').agg({
    'order_id': 'count',
    'amount': 'sum',
    'quantity': 'sum'
}).rename(columns={
    'order_id': 'order_count',
    'amount': 'total_amount',
    'quantity': 'total_quantity'
})

monthly_summary = daily_summary.resample('M').sum()
monthly_summary['avg_order_amount'] = (
    monthly_summary['total_amount'] / monthly_summary['order_count']
)

Data Factory Architecture

Typical Layering

┌─────────────────────────────────────────────────────────────┐
│                    数据工厂分层架构                          │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────┐                                          │
│  │  数据应用层  │  ← 面向业务定制，数据产品和分析结果       │
│  └──────┬──────┘                                          │
│         │                                                  │
│  ┌──────┴──────┐                                          │
│  │  汇总明细层  │  ← 主题建模，衍生/复合指标                │
│  └──────┬──────┘                                          │
│         │                                                  │
│  ┌──────┴──────┐                                          │
│  │   整合层    │  ← 数据清洗、转换、统一编码                │
│  └──────┬──────┘                                          │
│         │                                                  │
│  ┌──────┴──────┐                                          │
│  │   贴源层    │  ← 原始数据集成，不做清洗转换             │
│  └─────────────┘                                          │
└─────────────────────────────────────────────────────────────┘

Operational Data Store (ODS)

CREATE TABLE ods_orders (
    order_id STRING,
    user_id STRING,
    product_id STRING,
    order_date STRING,
    amount STRING,
    quantity STRING,
    discount STRING,
    etl_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Data Warehouse Detail (DWD)

CREATE TABLE dwd_orders (
    order_id BIGINT,
    user_id BIGINT,
    product_id BIGINT,
    order_date DATE,
    amount DECIMAL(10,2),
    quantity INT,
    discount DECIMAL(10,2),
    is_vip STRING
);

Data Warehouse Summary (DWS)

CREATE TABLE dws_user_daily_summary (
    stat_date DATE,
    user_id BIGINT,
    order_count INT,
    order_amount DECIMAL(12,2),
    avg_order_amount DECIMAL(10,2)
);

Common Python Tools

Advanced Pandas Operations

import pandas as pd
import numpy as np

df = pd.read_csv('sales.csv')

df.query('amount > 100 and category == "电子产品"')

df.pipe(lambda x: x[x['amount'] > 0])

df.assign(
    amount_after_discount=lambda x: x['amount'] * (1 - x['discount_rate']),
    month=lambda x: pd.to_datetime(x['date']).dt.month
)

pd.get_dummies(df, columns=['category'], prefix='cat')

df.groupby('user_id').filter(lambda x: len(x) >= 3)

SQL Window Functions

SELECT
    order_date,
    user_id,
    amount,
    SUM(amount) OVER (PARTITION BY user_id ORDER BY order_date) as cumulative_amount,
    AVG(amount) OVER (PARTITION BY user_id ORDER BY order_date ROWS BETWEEN 6 PRECEDING AND CURRENT ROW) as rolling_7d_avg,
    LAG(amount, 1) OVER (PARTITION BY user_id ORDER BY order_date) as prev_amount,
    amount - LAG(amount, 1) OVER (PARTITION BY user_id ORDER BY order_date) as amount_change
FROM orders
ORDER BY user_id, order_date;

Data Quality Monitoring

Monitoring Metrics

def data_quality_monitor(df, date_col='order_date'):
    yesterday = pd.Timestamp.now() - pd.Timedelta(days=1)
    today_data = df[df[date_col] == yesterday]

    metrics = {
        'record_count': len(today_data),
        'null_rate': today_data.isnull().mean().to_dict(),
        'duplicate_rate': today_data.duplicated().mean(),
        'amount_sum': today_data['amount'].sum() if 'amount' in today_data else None,
        'amount_avg': today_data['amount'].mean() if 'amount' in today_data else None
    }

    return metrics

Anomaly Detection

def detect_anomaly(df, column, threshold=3):
    mean = df[column].mean()
    std = df[column].std()
    anomalies = df[np.abs(df[column] - mean) > threshold * std]
    return anomalies

def detect_trend_change(df, column, window=7):
    rolling_mean = df[column].rolling(window=window).mean()
    trend_change = np.abs(rolling_mean.diff()) > df[column].std() * 2
    return trend_change

Resource Index

Pandas Documentation:https://pandas.pydata.org/
Data Modeling Reference:Full-chain Data Visualization

Notes

Data cleaning is the foundation of data analysis; quality matters more than speed
Keep the original data and create new fields for processed data
Data definitions must be clearly recorded for subsequent traceability
Confirm metric definitions with business stakeholders to avoid misunderstandings

About this skill

Task Objectives

Data Acquisition

Data Source Types

Data Acquisition Modes

Batch Acquisition

API Acquisition

File Import

Data Quality Assessment

Data Cleaning

Missing Value Handling

Outlier Handling

Data Type Conversion

Duplicate Value Handling

Data Transformation

Aggregate Statistics

Pivot Tables

Time Series Transformation

Data Normalization

Data Modeling

Dimension Table Design

Fact Table Design

Metric Calculation

Hierarchical Aggregation

Data Factory Architecture

Typical Layering

Operational Data Store (ODS)

Data Warehouse Detail (DWD)

Data Warehouse Summary (DWS)

Common Python Tools

Advanced Pandas Operations

SQL Window Functions

Data Quality Monitoring

Monitoring Metrics

Anomaly Detection

Resource Index

Notes