抱歉，您的问题中存在一个误解：**Python 并没有版本 2.68**。Python 的版本号通常是类似 `2.7.x` 或 `3.x.x` 的格式。目前官方支持的最低稳定版本是 Python 2.7（已于 2020 年停止维护），而 Python 2.6 及更早版本早已被淘汰。 --- 由于您提到的是 “Python 2.68”，我们假设您可能是想表达使用的是较老的 **Python 2.x 系列环境（如 Python 2.7）**，我们将提供一段兼容 **Python 2.7** 的拉链表处理脚本（基于 `pandas` 和 `MySQL`）。 > ⚠️ 提示：若您确实使用的是低于 2.7 的版本（例如 2.6 或更低），强烈建议升级，因为许多现代库（如 pandas、numpy）已不再支持 Python < 2.7。 --- ```python # -*- coding: utf-8 -*- """ 拉链表处理脚本（兼容 Python 2.7） 功能： 1. 删除重复跑批数据 2. 创建临时表并标注：新增、更新、不变 3. 关闭旧链（end_date 和 is_active） 4. 插入新链（新增和更新数据） """ import sys reload(sys) sys.setdefaultencoding('utf-8') # 兼容中文输出 import pandas as pd from sqlalchemy import create_engine # ===================== 配置区 ===================== # 数据库连接字符串：格式为 dialect+driver://user:pass@host:port/db db_url = "mysql+pymysql://username:password@localhost/dbname?charset=utf8" # 表名定义 source_table = "source_data" # 当前批次源表 history_table = "dim_table_his" # 拉链历史表 temp_table = "temp_delta_label" # 临时中间表 # ================================================= def execute_sql(engine, sql): """执行原始 SQL 语句""" with engine.begin() as conn: conn.execute(sql) def remove_duplicate_batches(engine): """1. 删除重复跑批数据（保留每个 id 在 batch_time 上最新的记录）""" print(u"正在删除重复跑批数据...") sql = """ DELETE s1 FROM {src} s1 USING {src} s2 WHERE s1.id = s2.id AND s1.batch_time < s2.batch_time; """.format(src=source_table) execute_sql(engine, sql) print(u"✅ 重复数据已清除") def create_temp_labeled_table(engine): """2. 创建临时表，标注每条记录的状态：新增、更新、不变""" print(u"正在生成变更标签...") drop_sql = "DROP TABLE IF EXISTS {tmp};".format(tmp=temp_table) create_sql = """ CREATE TABLE {tmp} AS SELECT s.*, CASE WHEN h.id IS NULL THEN '新增' WHEN COALESCE(s.col1, '') != COALESCE(h.col1, '') OR COALESCE(s.col2, '') != COALESCE(h.col2, '') OR COALESCE(s.col3, '') != COALESCE(h.col3, '') THEN '更新' ELSE '不变' END AS change_type, NOW() AS process_time FROM {src} s LEFT JOIN {his} h ON s.id = h.id AND h.is_active = 1; """.format(tmp=temp_table, src=source_table, his=history_table) execute_sql(engine, drop_sql) execute_sql(engine, create_sql) print(u"✅ 临时标签表已创建") def update_history_table(engine): """3. 更新原有效记录：关闭旧链（设置 end_date 和 is_active=0）""" print(u"正在关闭旧的有效链...") sql = """ UPDATE {his} SET end_date = NOW(), is_active = 0, update_time = NOW() WHERE is_active = 1 AND id IN (SELECT id FROM {tmp} WHERE change_type IN ('更新', '新增')); """.format(his=history_table, tmp=temp_table) execute_sql(engine, sql) print(u"✅ 旧链已关闭") def insert_new_and_updated_records(engine): """4. 插入新增和更新的数据作为新链""" print(u"正在插入新版本数据...") sql = """ INSERT INTO {his} (id, col1, col2, col3, start_date, end_date, is_active, source_batch, create_time) SELECT id, col1, col2, col3, NOW() AS start_date, '9999-12-31' AS end_date, 1 AS is_active, batch_time AS source_batch, NOW() AS create_time FROM {tmp} WHERE change_type IN ('新增', '更新'); """.format(his=history_table, tmp=temp_table) execute_sql(engine, sql) print(u"✅ 新链已插入") def check_count(engine): """统计最终表行数""" sql = "SELECT COUNT(*) AS cnt FROM {his};".format(his=history_table) df = pd.read_sql(sql, engine) count = df.iloc[0]['cnt'] print(u"📊 当前拉链表总记录数: %d" % count) # ===================== 主程序 ===================== if __name__ == "__main__": try: # 创建数据库引擎 engine = create_engine(db_url, echo=False) remove_duplicate_batches(engine) create_temp_labeled_table(engine) update_history_table(engine) insert_new_and_updated_records(engine) check_count(engine) print(u"✅ 拉链表处理完成！") except Exception as e: print(u"❌ 出错: %s" % str(e).decode('utf-8')) ``` --- ### ✅ 环境依赖安装（Python 2.7）： ```bash pip install pandas==0.24.2 pip install sqlalchemy pip install pymysql ``` > 📌 注意：`pandas>=1.0` 已不支持 Python 2，所以请使用 `pandas==0.24.2`。 --- ### 🔧 使用前请注意： 1. 替换数据库连接参数； 2. 修改字段名（如 `id`, `col1`, `col2`, `batch_time` 等）为真实字段； 3. 确保目标表结构包含： - `start_date`（生效开始） - `end_date`（生效结束） - `is_active`（是否当前有效） - `source_batch`（来源批次） --- 如果您能提供具体的数据库类型（MySQL / Oracle / Hive）、表结构或字段清单，我可以为您生成完全适配的脚本版本。