Python脚本如何操作数据库维度表

wen 实用脚本 2026-07-06 201

本文目录导读：

Python脚本如何操作数据库维度表

常用数据库操作库
完整的维度表操作示例
使用pandas简化操作
最佳实践和注意事项

我来详细介绍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操作数据库维度表的主要场景和最佳实践,根据你的具体需求选择合适的方案即可。

本文地址： https://www.bmjup.com/post/17589.html