如何在 Python 中基于非唯一 ID 和最近年份匹配连接两个 SQL 表-Python教程-PHP中文网

如何在 Python 中基于非唯一 ID 和最近年份匹配连接两个 SQL 表

本文介绍如何使用 sqlite + pandas 在内存中高效连接超大规模出生与婚姻数据表，通过非唯一姓名 id 关联，并为每个出生记录精确匹配时间上最接近的婚姻记录，最终计算年份差并统计分布。

本文介绍如何使用 sqlite + pandas 在内存中高效连接超大规模出生与婚姻数据表，通过非唯一姓名 id 关联，并为每个出生记录精确匹配时间上最接近的婚姻记录，最终计算年份差并统计分布。

在构建家谱自动化系统时，一个核心挑战是：当多个婚姻证书包含完全相同的父母姓名（即 PARENTS_ID = MARRIAGE_ID）时，如何从这些候选婚姻中选出最可能对应当前出生事件的一条？由于姓名重复率高（尤其在历史户籍数据中），仅靠字符串匹配无法保证准确性——而引入时间维度（即“婚姻年份最接近出生年份”）是强效的业务约束。本文提供一种内存可控、可扩展、SQL 原生实现的解决方案，适用于千万级数据场景。

核心思路：窗口函数 + 分块处理 + 索引优化

关键不在于“先算差再取最小”，而是在 JOIN 后对每个 PARENTS_ID 分组，按 |birth_year − marriage_year| 升序排序，并仅保留排名第一的记录。这需借助 ROW_NUMBER() OVER (...) 窗口函数——它必须在子查询（或 CTE）中定义，不能在 SELECT 列表中直接引用别名（原报错 AGE_DIFF does not exist 正源于此：SQL 不允许在同级 SELECT 中复用刚定义的列别名）。

以下是完整、可运行的优化代码：

python学习

Python 编程学习助手 skill，面向零基础用户。当用户表达以下意图时触发： - 学习 Python 基础概念（变量、数据类型、循环、函数、类等） - 请求 Python 练习题或测验 - 请求审查或调试 Python 代码 - 请求 Python 项目实战指导（爬虫、数据分析、Web 开发等入门项目） - 泛化表述如"教我 Python""我想学编程""Python 怎么写""帮我学 Python" 此 skill

下载

import sqlite3
import pandas as pd
import time

def generate_chunks(df, chunk_size):
    for i in range(0, len(df), chunk_size):
        yield df.iloc[i:i+chunk_size]

def freq(dic, arr):
    for i in arr:
        dic[i] = dic.get(i, 0) + 1
    return dic

# ✅ 假设 births 和 marriages 已加载为 DataFrame，且含以下列：
# births: ['PR_BIR_YEAR', 'PARENTS_ID', ...]
# marriages: ['EVENT_YEAR', 'MARRIAGE_ID', ...]

start = time.time()
conn = sqlite3.connect(':memory:')
conn.execute('PRAGMA synchronous = OFF;')
conn.execute('PRAGMA journal_mode = MEMORY;')
conn.execute('PRAGMA temp_store = MEMORY;')

# ✅ 一次性加载婚姻表（相对较小：500万行），并建索引
marriages.to_sql('sqlmarriages', conn, index=False, if_exists='replace')
conn.execute("CREATE INDEX idx_marriage_id ON sqlmarriages (MARRIAGE_ID);")

chunk_size = 2000
age_diff_freq = {}  # 推荐用 dict 而非全局变量

for i, chunk in enumerate(generate_chunks(births, chunk_size)):
    print(f"Processing chunk {i} at {time.time() - start:.2f}s")

    # ✅ 每次用 'replace' 覆盖临时表，避免重复建表错误
    chunk.to_sql('sqlchunk', conn, index=False, if_exists='replace')
    conn.execute("CREATE INDEX IF NOT EXISTS idx_parents_id ON sqlchunk (PARENTS_ID);")

    # ✅ 核心查询：CTE + ROW_NUMBER() 实现“每组取最近”
    query = """
    WITH ranked_matches AS (
      SELECT 
        b.PARENTS_ID,
        b.PR_BIR_YEAR,
        m.EVENT_YEAR,
        m.EVENT_YEAR - b.PR_BIR_YEAR AS AGE_DIFF,
        ROW_NUMBER() OVER (
          PARTITION BY b.PARENTS_ID 
          ORDER BY ABS(b.PR_BIR_YEAR - m.EVENT_YEAR)
        ) AS rn
      FROM sqlchunk b
      INNER JOIN sqlmarriages m
        ON b.PARENTS_ID = m.MARRIAGE_ID
    )
    SELECT PARENTS_ID, PR_BIR_YEAR, EVENT_YEAR, AGE_DIFF
    FROM ranked_matches
    WHERE rn = 1;
    """

    matches = pd.read_sql_query(query, conn)

    # ✅ 提取并累加年份差频次
    age_differences = matches['AGE_DIFF'].dropna().astype(int).tolist()
    age_diff_freq = freq(age_diff_freq, age_differences)

print("✅ Frequency distribution of marriage-birth year gaps:")
print({k: v for k, v in sorted(age_diff_freq.items()) if -5 <h3>⚠️ 关键注意事项</h3>

索引必须存在：MARRIAGE_ID 和 PARENTS_ID 上的 B-tree 索引对 INNER JOIN 性能影响巨大；缺失索引可能导致查询从秒级退化至小时级。
if_exists='replace' 是必须的：避免多次写入 sqlchunk 导致主键冲突或重复索引错误。
ABS(...) 确保“最近”是双向的：既支持婚前生育（负值），也支持婚后多年生育（正值），符合真实人口规律。
内存安全边界：chunk_size=2000 是经验平衡值；若 OOM，可降至 500；若 CPU 成瓶颈，可适度提高（如 5000），但需同步监控 SQLite 内存占用。
数据质量前置校验：确保 PR_BIR_YEAR 和 EVENT_YEAR 为整数且无异常值（如 0, 9999），否则 ABS() 会引入噪声。

✅ 最终产出与延伸

运行结束后，age_diff_freq 字典即为所有成功匹配对的年份差频次统计——例如 {0: 245812, 1: 189304, -1: 76231} 表明：约 24.6 万对父母在孩子出生同年结婚，18.9 万对在次年结婚，7.6 万对在出生前一年结婚。这一分布可直接用于：