SQL 数据分析

Pandas 很强大，但现实世界中大部分数据都存放在关系型数据库里。无论你是从数据仓库取数、查询业务系统，还是做报表，SQL 都是数据分析师最核心的技能之一。

这篇文章不讲 DBA 那一套（建表、索引优化、事务管理），而是聚焦于数据分析师最常用的 SQL 操作，并用 SQLite 作为实践环境，配合 Python 打通数据链路。

环境准备

import sqlite3
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns

# 创建内存数据库
conn = sqlite3.connect(':memory:')
cursor = conn.cursor()

我们创建三张模拟的业务表：

# 用户表
cursor.execute('''
CREATE TABLE users (
    user_id INTEGER PRIMARY KEY,
    name TEXT,
    age INTEGER,
    city TEXT,
    signup_date TEXT
)
''')

# 订单表
cursor.execute('''
CREATE TABLE orders (
    order_id INTEGER PRIMARY KEY,
    user_id INTEGER,
    product TEXT,
    amount REAL,
    quantity INTEGER,
    order_date TEXT,
    status TEXT
)
''')

# 产品表
cursor.execute('''
CREATE TABLE products (
    product_id INTEGER PRIMARY KEY,
    product_name TEXT,
    category TEXT,
    price REAL,
    stock INTEGER
)
''')

插入模拟数据：

import random
from datetime import datetime, timedelta

# 插入用户
users = []
for i in range(1, 101):
    users.append((
        i,
        f'用户{i}',
        random.randint(18, 65),
        random.choice(['北京', '上海', '广州', '深圳', '杭州', '成都']),
        f'2024-{random.randint(1,12):02d}-{random.randint(1,28):02d}'
    ))
cursor.executemany('INSERT INTO users VALUES (?,?,?,?,?)', users)

# 插入产品
products = [
    (1, '笔记本电脑', '电子产品', 5999, 100),
    (2, '手机', '电子产品', 3999, 200),
    (3, '耳机', '电子产品', 299, 500),
    (4, '运动鞋', '服饰', 599, 300),
    (5, 'T恤', '服饰', 99, 1000),
    (6, '咖啡机', '家居', 1299, 50),
    (7, '台灯', '家居', 199, 200),
]
cursor.executemany('INSERT INTO products VALUES (?,?,?,?,?)', products)

# 插入订单
orders = []
for i in range(1, 501):
    user_id = random.randint(1, 100)
    product_id = random.randint(1, 7)
    product_name = ['笔记本电脑', '手机', '耳机', '运动鞋', 'T恤', '咖啡机', '台灯'][product_id-1]
    price = [5999, 3999, 299, 599, 99, 1299, 199][product_id-1]
    qty = random.randint(1, 3)
    days_ago = random.randint(0, 365)
    order_date = (datetime(2024, 1, 1) + timedelta(days=days_ago)).strftime('%Y-%m-%d')
    
    orders.append((
        i, user_id, product_name, price * qty,
        qty, order_date,
        random.choice(['已完成', '已完成', '已完成', '已取消', '退款中'])
    ))
cursor.executemany('INSERT INTO orders VALUES (?,?,?,?,?,?,?)', orders)

conn.commit()

SQL 基础查询回顾

最常用的 SELECT 语句组合：

pd.read_sql('''
SELECT * FROM orders
WHERE status = '已完成'
  AND amount > 500
  AND order_date >= '2024-06-01'
ORDER BY amount DESC
LIMIT 10
''', conn)

聚合与分组

数据分析的核心是汇总和对比。

# 基础聚合
pd.read_sql('''
SELECT 
    COUNT(*) AS 订单总数,
    SUM(amount) AS 总金额,
    AVG(amount) AS 平均金额,
    MIN(amount) AS 最小订单,
    MAX(amount) AS 最大订单,
    COUNT(DISTINCT user_id) AS 下单用户数
FROM orders
WHERE status = '已完成'
''', conn)

分组聚合是 SQL 分析的精髓：

# 按产品分组
pd.read_sql('''
SELECT 
    product,
    COUNT(*) AS 订单数,
    SUM(amount) AS 销售额,
    ROUND(AVG(amount), 2) AS 均价,
    SUM(quantity) AS 销量
FROM orders
WHERE status = '已完成'
GROUP BY product
ORDER BY 销售额 DESC
''', conn)

# 多维度分组
pd.read_sql('''
SELECT 
    strftime('%m', order_date) AS 月份,
    product,
    COUNT(*) AS 订单数,
    SUM(amount) AS 销售额
FROM orders
WHERE status = '已完成'
GROUP BY 月份, product
ORDER BY 月份, 销售额 DESC
''', conn)

HAVING 过滤聚合结果：

pd.read_sql('''
SELECT 
    user_id,
    COUNT(*) AS 订单数,
    SUM(amount) AS 总消费
FROM orders
WHERE status = '已完成'
GROUP BY user_id
HAVING 总消费 > 10000
ORDER BY 总消费 DESC
''', conn)

JOIN 连接表

真实分析几乎从不只查一张表。

# INNER JOIN
pd.read_sql('''
SELECT 
    o.order_id,
    u.name,
    u.city,
    o.product,
    o.amount,
    o.order_date
FROM orders o
INNER JOIN users u ON o.user_id = u.user_id
WHERE o.status = '已完成'
LIMIT 10
''', conn)

不同 JOIN 的区别：

# 用户与其订单（包括从未下单的用户）
pd.read_sql('''
SELECT 
    u.user_id,
    u.name,
    COUNT(o.order_id) AS 订单数,
    COALESCE(SUM(o.amount), 0) AS 总消费
FROM users u
LEFT JOIN orders o ON u.user_id = o.user_id AND o.status = '已完成'
GROUP BY u.user_id
ORDER BY 总消费 DESC
LIMIT 10
''', conn)

# 每个用户最近一次订单（用 LEFT JOIN 的技巧）
pd.read_sql('''
SELECT 
    u.user_id,
    u.name,
    o.product,
    o.amount,
    o.order_date
FROM users u
LEFT JOIN orders o ON u.user_id = o.user_id
LEFT JOIN orders o2 ON u.user_id = o2.user_id AND o.order_date < o2.order_date
WHERE o2.order_id IS NULL
  AND o.status = '已完成'
ORDER BY u.user_id
LIMIT 15
''', conn)

子查询

# 查询消费高于平均水平的用户
pd.read_sql('''
SELECT 
    u.name,
    u.city,
    user_total.总消费
FROM (
    SELECT 
        user_id,
        SUM(amount) AS 总消费
    FROM orders
    WHERE status = '已完成'
    GROUP BY user_id
) user_total
JOIN users u ON user_total.user_id = u.user_id
WHERE user_total.总消费 > (
    SELECT AVG(总消费) FROM (
        SELECT SUM(amount) AS 总消费
        FROM orders
        WHERE status = '已完成'
        GROUP BY user_id
    )
)
ORDER BY user_total.总消费 DESC
''', conn)

# 每个城市消费最高的用户
pd.read_sql('''
SELECT 
    u.city,
    u.name,
    t.总消费
FROM (
    SELECT 
        user_id,
        SUM(amount) AS 总消费,
        RANK() OVER (PARTITION BY u2.city ORDER BY SUM(amount) DESC) AS rnk
    FROM orders o
    JOIN users u2 ON o.user_id = u2.user_id
    WHERE o.status = '已完成'
    GROUP BY o.user_id
) t
JOIN users u ON t.user_id = u.user_id
WHERE t.rnk = 1
ORDER BY t.总消费 DESC
''', conn)

WITH 子句（CTE）

CTE 让复杂查询变得像流水线一样清晰：

pd.read_sql('''
WITH 用户消费 AS (
    SELECT 
        user_id,
        COUNT(*) AS 订单数,
        SUM(amount) AS 总消费,
        AVG(amount) AS 客单价
    FROM orders
    WHERE status = '已完成'
    GROUP BY user_id
),
用户分层 AS (
    SELECT 
        user_id,
        CASE 
            WHEN 总消费 > 30000 THEN '高价值'
            WHEN 总消费 > 10000 THEN '中价值'
            ELSE '低价值'
        END AS 层级
    FROM 用户消费
)
SELECT 
    l.层级,
    COUNT(*) AS 用户数,
    ROUND(AVG(c.总消费), 2) AS 人均消费,
    ROUND(AVG(c.订单数), 2) AS 人均订单
FROM 用户分层 l
JOIN 用户消费 c ON l.user_id = c.user_id
GROUP BY l.层级
ORDER BY AVG(c.总消费) DESC
''', conn)

窗口函数

窗口函数是进阶 SQL 分析中最重要的技能。它能在一行上看到上下文中的其他行。

ROW_NUMBER / RANK / DENSE_RANK

# 每个产品类别的销量排名
pd.read_sql('''
SELECT 
    product,
    SUM(quantity) AS 销量,
    RANK() OVER (ORDER BY SUM(quantity) DESC) AS 总排名,
    ROW_NUMBER() OVER (PARTITION BY '整体' ORDER BY SUM(quantity) DESC) AS 序号
FROM orders
WHERE status = '已完成'
GROUP BY product
''', conn)

LAG / LEAD

# 月度销售额与环比增长
pd.read_sql('''
WITH 月度 AS (
    SELECT 
        strftime('%Y-%m', order_date) AS 月份,
        SUM(amount) AS 销售额
    FROM orders
    WHERE status = '已完成'
    GROUP BY 月份
)
SELECT 
    月份,
    销售额,
    LAG(销售额) OVER (ORDER BY 月份) AS 上月销售额,
    ROUND((销售额 - LAG(销售额) OVER (ORDER BY 月份)) / 
          LAG(销售额) OVER (ORDER BY 月份) * 100, 2) AS 环比增长率
FROM 月度
ORDER BY 月份
''', conn)

移动窗口函数

# 7 日移动平均销售额
pd.read_sql('''
WITH 每日 AS (
    SELECT 
        order_date,
        SUM(amount) AS 日销售额
    FROM orders
    WHERE status = '已完成'
    GROUP BY order_date
)
SELECT 
    order_date,
    日销售额,
    ROUND(AVG(日销售额) OVER (
        ORDER BY order_date 
        ROWS BETWEEN 6 PRECEDING AND CURRENT ROW
    ), 2) AS 七日移动平均
FROM 每日
ORDER BY order_date
LIMIT 20
''', conn)

SUM OVER PARTITION

# 累计销售额
pd.read_sql('''
WITH 每日 AS (
    SELECT 
        order_date,
        SUM(amount) AS 日销售额
    FROM orders
    WHERE status = '已完成'
    GROUP BY order_date
)
SELECT 
    order_date,
    日销售额,
    ROUND(SUM(日销售额) OVER (ORDER BY order_date), 2) AS 累计销售额
FROM 每日
ORDER BY order_date
LIMIT 15
''', conn)

日期与字符串处理

# 日期函数
pd.read_sql('''
SELECT 
    order_date,
    strftime('%Y', order_date) AS 年,
    strftime('%m', order_date) AS 月,
    strftime('%w', order_date) AS 星期几,  -- 0=周日
    CASE CAST(strftime('%w', order_date) AS INTEGER)
        WHEN 0 THEN '周日'
        WHEN 6 THEN '周六'
        ELSE '工作日'
    END AS 日类型,
    strftime('%Y-%m', order_date) AS 年月,
    julianday('2024-12-31') - julianday(order_date) AS 距年底天数
FROM orders
LIMIT 10
''', conn)

# 字符串函数
pd.read_sql('''
SELECT 
    name,
    LENGTH(name) AS 姓名长度,
    SUBSTR(name, 1, 1) AS 首字符,
    UPPER(name) AS 大写,
    REPLACE(city, '北京', 'Beijing') AS 城市双语,
    '【' || city || '】' || name AS 完整名称
FROM users
LIMIT 10
''', conn)

性能优化基础

当数据量达到百万级时，SQL 查询性能就变得重要了。

使用 EXPLAIN 分析查询计划

cursor.execute('EXPLAIN QUERY PLAN SELECT * FROM orders WHERE user_id = 42')
for row in cursor.fetchall():
    print(row)

输出会显示全表扫描还是索引查找。全表扫描 = 危险信号。

创建索引

# 为常用查询字段创建索引
cursor.execute('CREATE INDEX idx_orders_user_id ON orders(user_id)')
cursor.execute('CREATE INDEX idx_orders_date ON orders(order_date)')
cursor.execute('CREATE INDEX idx_orders_status ON orders(status)')
cursor.execute('CREATE INDEX idx_orders_user_status ON orders(user_id, status)')

# 再 EXPLAIN 一次，应该变成索引查找
cursor.execute('EXPLAIN QUERY PLAN SELECT * FROM orders WHERE user_id = 42')
for row in cursor.fetchall():
    print(row)

# 索引效果对比
import time

# 无索引查询（先删除索引做对比）
cursor.execute('DROP INDEX IF EXISTS idx_orders_user_id')
start = time.time()
cursor.execute('SELECT * FROM orders WHERE user_id = 42')
no_index = time.time() - start

# 重建索引
cursor.execute('CREATE INDEX idx_orders_user_id ON orders(user_id)')
start = time.time()
cursor.execute('SELECT * FROM orders WHERE user_id = 42')
with_index = time.time() - start

print(f'无索引: {no_index*1000:.2f}ms')
print(f'有索引: {with_index*1000:.2f}ms')
print(f'加速比: {no_index/with_index:.1f}x')

性能优化原则

优化建议 = '''
1. SELECT 只查需要的列，不要 SELECT *
2. WHERE 条件用索引列
3. JOIN 的关联列要有索引
4. 避免在 WHERE 中对列做函数运算（WHERE YEAR(date)=2024 不如 WHERE date>='2024-01-01'）
5. 大表分页用游标，不用 OFFSET
6. 聚合前先用 WHERE 过滤掉不需要的行
7. 小表驱动大表
'''
print(优化建议)

Python 与 SQL 的混合使用

真实项目中，SQL 和 Python 是互补的：

# SQL 负责取数和聚合，Python 负责分析和可视化
query = '''
SELECT 
    u.city,
    COUNT(DISTINCT o.user_id) AS 活跃用户数,
    SUM(o.amount) AS 总消费,
    AVG(o.amount) AS 客单价
FROM orders o
JOIN users u ON o.user_id = u.user_id
WHERE o.status = '已完成'
GROUP BY u.city
ORDER BY 总消费 DESC
'''

df_city = pd.read_sql(query, conn)

# 用 Python 做后续分析
df_city['人均消费'] = df_city['总消费'] / df_city['活跃用户数']

plt.figure(figsize=(10, 5))
sns.barplot(data=df_city, x='city', y='总消费')
plt.title('各城市消费情况')
plt.show()

print(df_city.sort_values('总消费', ascending=False))

用 SQLAlchemy 管理更复杂的连接：

from sqlalchemy import create_engine

# 创建 SQLAlchemy 引擎（支持 PostgreSQL, MySQL, SQLite 等）
engine = create_engine('sqlite:///analysis.db')

# 直接用 pandas 读写
df = pd.read_sql('SELECT * FROM orders', engine)

# 把 DataFrame 写入数据库
df_agg = df.groupby('product').agg(
    订单数=('order_id', 'count'),
    销售额=('amount', 'sum')
).reset_index()

df_agg.to_sql('product_summary', engine, if_exists='replace', index=False)

pd.read_sql('SELECT * FROM product_summary ORDER BY 销售额 DESC', engine)

实战：RFM 用户分析

RFM（Recency, Frequency, Monetary）是经典的用户分层模型：

rfm_query = '''
WITH user_stats AS (
    SELECT 
        o.user_id,
        u.name,
        u.city,
        julianday('2024-12-31') - julianday(MAX(o.order_date)) AS recency,
        COUNT(DISTINCT o.order_id) AS frequency,
        SUM(o.amount) AS monetary
    FROM orders o
    JOIN users u ON o.user_id = u.user_id
    WHERE o.status = '已完成'
    GROUP BY o.user_id
),
rfm_scores AS (
    SELECT *,
        NTILE(5) OVER (ORDER BY recency DESC) AS R_score,  -- 越近分越高
        NTILE(5) OVER (ORDER BY frequency) AS F_score,
        NTILE(5) OVER (ORDER BY monetary) AS M_score
    FROM user_stats
)
SELECT *,
    R_score || F_score || M_score AS rfm_code,
    CASE 
        WHEN R_score >= 4 AND F_score >= 4 THEN '重要价值'
        WHEN R_score >= 4 AND F_score >= 2 THEN '重要发展'
        WHEN R_score <= 2 AND F_score >= 4 THEN '重要保持'
        WHEN R_score <= 2 AND F_score <= 2 AND M_score <= 2 THEN '一般流失'
        ELSE '一般用户'
    END AS 用户分层
FROM rfm_scores
ORDER BY monetary DESC
LIMIT 20
'''

df_rfm = pd.read_sql(rfm_query, conn)

# 可视化
fig, axes = plt.subplots(1, 3, figsize=(15, 4))
axes[0].hist(df_rfm['recency'], bins=20)
axes[0].set_title('Recency 分布')
axes[1].hist(df_rfm['frequency'], bins=20)
axes[1].set_title('Frequency 分布')
axes[2].hist(df_rfm['monetary'], bins=20)
axes[2].set_title('Monetary 分布')
plt.tight_layout()
plt.show()

# 用户分层统计
print(df_rfm['用户分层'].value_counts())

SQL 学习路线图

初级（日常取数）
├── SELECT / WHERE / ORDER BY / LIMIT
├── COUNT / SUM / AVG / GROUP BY / HAVING
├── JOIN（INNER / LEFT）
└── 子查询

中级（分析工作）
├── 窗口函数（RANK / LAG / SUM OVER）
├── WITH（CTE）重构复杂查询
├── CASE WHEN 条件聚合
├── 日期与字符串函数
└── UNION / UNION ALL

高级（性能与优化）
├── 索引设计与查询计划
├── 分区表处理百万级数据
├── 窗口框架（ROWS/RANGE/GROUPS）
├── 递归 CTE
└── 物化视图

小结

SQL 是数据分析师的"母语"。不管业务需求多复杂，最终都要落地为一条 SQL 查询。掌握本文的内容，你已经能胜任绝大多数分析场景下的 SQL 需求：

用 JOIN 关联多张表，得到完整的业务视图
用 GROUP BY 做多维度聚合分析
用窗口函数计算排名、同环比、累计值
用 CTE 组织复杂查询，让逻辑清晰可读
用索引优化查询性能
用 Python + SQL 形成完整的工作流

最后一篇文章，我们来处理一类特殊但常见的数据——时间序列。

Summary: SQL 聚合、JOIN、窗口函数与 Python 集成实战。