本文目录导读：
我来详细介绍Python操作数据库维度表的方法,包括常用的库和最佳实践。
常用数据库操作库
MySQL
import pymysql
from sqlalchemy import create_engine
# 方法1：使用pymysql
conn = pymysql.connect(
host='localhost',
user='root',
password='password',
database='dimension_db'
)
cursor = conn.cursor()
# 方法2：使用SQLAlchemy (推荐)
engine = create_engine('mysql+pymysql://root:password@localhost:3306/dimension_db')
PostgreSQL
import psycopg2
from sqlalchemy import create_engine
# 使用psycopg2
conn = psycopg2.connect(
host='localhost',
user='postgres',
password='password',
database='dimension_db'
)
SQL Server
import pyodbc
from sqlalchemy import create_engine
# 使用pyodbc
conn = pyodbc.connect(
'DRIVER={ODBC Driver 17 for SQL Server};'
'SERVER=localhost;DATABASE=dimension_db;'
'UID=sa;PWD=password'
)
完整的维度表操作示例
import pandas as pd
from sqlalchemy import create_engine, Table, MetaData, Column, Integer, String, DateTime
from datetime import datetime
import logging
# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DimensionTableManager:
def __init__(self, connection_string):
"""
初始化维度表管理器
"""
self.engine = create_engine(connection_string)
self.metadata = MetaData()
def create_date_dimension(self):
"""
创建日期维度表
"""
date_dim = Table('dim_date', self.metadata,
Column('date_key', Integer, primary_key=True),
Column('full_date', DateTime),
Column('year', Integer),
Column('month', Integer),
Column('day', Integer),
Column('quarter', Integer),
Column('day_of_week', Integer),
Column('is_weekend', Integer),
Column('month_name', String(20)),
Column('day_name', String(20))
)
self.metadata.create_all(self.engine)
logger.info("日期维度表创建成功")
return date_dim
def insert_date_dimension(self, start_date='2020-01-01', end_date='2030-12-31'):
"""
填充日期维度数据
"""
dates = pd.date_range(start=start_date, end=end_date, freq='D')
date_data = []
for date in dates:
date_data.append({
'date_key': int(date.strftime('%Y%m%d')),
'full_date': date,
'year': date.year,
'month': date.month,
'day': date.day,
'quarter': (date.month - 1) // 3 + 1,
'day_of_week': date.weekday(),
'is_weekend': 1 if date.weekday() >= 5 else 0,
'month_name': date.strftime('%B'),
'day_name': date.strftime('%A')
})
# 批量插入数据
df = pd.DataFrame(date_data)
df.to_sql('dim_date', self.engine, if_exists='replace', index=False)
logger.info(f"插入了 {len(date_data)} 条日期维度数据")
def create_product_dimension(self):
"""
创建产品维度表
"""
product_dim = Table('dim_product', self.metadata,
Column('product_key', Integer, primary_key=True, autoincrement=True),
Column('product_id', String(50)),
Column('product_name', String(200)),
Column('category', String(100)),
Column('subcategory', String(100)),
Column('brand', String(100)),
Column('unit_price', Integer),
Column('created_date', DateTime, default=datetime.now),
Column('effective_date', DateTime),
Column('expiry_date', DateTime),
Column('is_current', Integer, default=1)
)
self.metadata.create_all(self.engine)
logger.info("产品维度表创建成功")
return product_dim
def scd_type2_update(self, table_name, business_key, current_data, new_data):
"""
实现SCD Type 2缓慢变化维更新
"""
with self.engine.connect() as conn:
# 检查是否有当前有效记录
query = f"""
SELECT * FROM {table_name}
WHERE {business_key} = :business_key
AND is_current = 1
"""
result = conn.execute(query, {'business_key': current_data[business_key]})
existing_record = result.fetchone()
if existing_record:
# 比较数据是否变化
has_changed = False
for key, value in new_data.items():
if key not in ['effective_date', 'expiry_date', 'is_current']:
if getattr(existing_record, key) != value:
has_changed = True
break
if has_changed:
# 过期旧记录
update_query = f"""
UPDATE {table_name}
SET expiry_date = :expiry_date, is_current = 0
WHERE {business_key} = :business_key
AND is_current = 1
"""
conn.execute(update_query, {
'expiry_date': datetime.now(),
'business_key': current_data[business_key]
})
# 插入新记录
new_data['effective_date'] = datetime.now()
new_data['expiry_date'] = None
new_data['is_current'] = 1
insert_query = f"""
INSERT INTO {table_name} ({','.join(new_data.keys())})
VALUES ({','.join([':' + k for k in new_data.keys()])})
"""
conn.execute(insert_query, new_data)
logger.info(f"SCD Type 2更新: 记录{business_key}已变化")
else:
# 新记录，直接插入
conn.execute(f"""
INSERT INTO {table_name}
VALUES ({','.join([':' + k for k in new_data.keys()])})
""", new_data)
logger.info(f"新记录插入: {business_key}")
conn.commit()
# 使用示例
if __name__ == "__main__":
# 配置数据库连接
connection_string = 'mysql+pymysql://user:password@localhost:3306/dw_demo'
manager = DimensionTableManager(connection_string)
# 创建日期维度
manager.create_date_dimension()
manager.insert_date_dimension('2023-01-01', '2024-12-31')
# 创建产品维度
manager.create_product_dimension()
# SCD Type 2 更新示例
new_product_data = {
'product_id': 'PROD001',
'product_name': '新产品A',
'category': '电子产品',
'subcategory': '手机',
'brand': 'ABC手机',
'unit_price': 4999,
'effective_date': datetime.now(),
'expiry_date': None,
'is_current': 1
}
manager.scd_type2_update('dim_product', 'product_id',
{'product_id': 'PROD001'},
new_product_data)
使用pandas简化操作
import pandas as pd
from sqlalchemy import create_engine
class PandasDimensionManager:
def __init__(self, engine):
self.engine = engine
def load_csv_to_dimension(self, csv_file, table_name, if_exists='append'):
"""
从CSV文件加载数据到维度表
"""
df = pd.read_csv(csv_file)
# 数据清洗
df = self._clean_dimension_data(df)
# 写入数据库
df.to_sql(table_name, self.engine, if_exists=if_exists, index=False)
return len(df)
def _clean_dimension_data(self, df):
"""
清洗维度数据
"""
# 去除空值
df = df.dropna()
# 去除重复
df = df.drop_duplicates()
# 字符串标准化
string_columns = df.select_dtypes(include=['object']).columns
for col in string_columns:
df[col] = df[col].str.strip().str.upper()
return df
def merge_dimension_data(self, staging_data, dimension_table, key_columns):
"""
合并维度数据（UPSERT操作）
"""
# 读取现有维度数据
existing_data = pd.read_sql(f"SELECT * FROM {dimension_table}", self.engine)
# 找出新的记录
merged = pd.merge(staging_data, existing_data,
on=key_columns, how='left', indicator=True)
new_records = merged[merged['_merge'] == 'left_only']
if not new_records.empty:
# 只保留新数据的列
new_records = new_records[staging_data.columns]
new_records.to_sql(dimension_table, self.engine,
if_exists='append', index=False)
return len(new_records)
# 使用示例
engine = create_engine('postgresql://user:password@localhost:5432/dw')
manager = PandasDimensionManager(engine)
# 加载数据
count = manager.load_csv_to_dimension('dim_customer.csv', 'dim_customer')
print(f"已加载 {count} 条记录")
最佳实践和注意事项
错误处理和事务管理
import contextlib
from sqlalchemy.exc import SQLAlchemyError
@contextlib.contextmanager
def transaction(engine):
"""事务管理器"""
connection = engine.connect()
transaction = connection.begin()
try:
yield connection
transaction.commit()
except SQLAlchemyError as e:
transaction.rollback()
logger.error(f"事务回滚: {e}")
raise
finally:
connection.close()
# 使用示例
with transaction(engine) as conn:
# 执行维度表操作
conn.execute("INSERT INTO dim_product ...")
conn.execute("UPDATE dim_product ...")
性能优化
def batch_insert(engine, table_name, data_list, batch_size=1000):
"""
批量插入数据，优化性能
"""
from sqlalchemy import text
if not data_list:
return 0
with engine.connect() as conn:
for i in range(0, len(data_list), batch_size):
batch = data_list[i:i + batch_size]
# 构建批量插入语句
columns = batch[0].keys()
values_str = ', '.join([f"({','.join([':%s_%d' % (col, j) for col in columns])})"
for j in range(len(batch))])
params = {}
for j, record in enumerate(batch):
for col in columns:
params[f'{col}_{j}'] = record[col]
query = f"INSERT INTO {table_name} ({','.join(columns)}) VALUES {values_str}"
conn.execute(text(query), params)
conn.commit()
return len(data_list)
这些示例涵盖了Python操作数据库维度表的主要场景和最佳实践,根据你的具体需求选择合适的方案即可。
