Python高效處理手機號數(shù)據(jù)缺失問題的方法

更新時間：2025年06月16日 08:13:54 作者：碼農阿豪@新空間

在大數(shù)據(jù)時代,完整準確的數(shù)據(jù)是業(yè)務發(fā)展的基石,本文將以處理手機號數(shù)據(jù)缺失問題為例,詳細介紹如何通過Python實現(xiàn)高效的數(shù)據(jù)校驗與補全方案,有需要的小伙伴可以了解下

引言

在大數(shù)據(jù)時代，完整準確的數(shù)據(jù)是業(yè)務發(fā)展的基石。本文將以處理手機號數(shù)據(jù)缺失問題為例，詳細介紹如何通過Python實現(xiàn)高效的數(shù)據(jù)校驗與補全方案。我們將從需求分析開始，逐步深入解決方案的設計與優(yōu)化，最終形成一個可應用于生產(chǎn)環(huán)境的高效腳本。

需求分析

業(yè)務背景

假設我們有一個包含數(shù)千萬手機號的數(shù)據(jù)庫表，其中手機號被拆分為以下幾個字段存儲：

前三位（prefix）：如157、185等
中間四位（middle）：0000-9999
后四位（suffix）：0000-9999

核心需求

完整性校驗：確保每個(prefix, suffix)組合下，中間四位0000-9999全部存在

缺失補全：自動識別缺失的中間四位并補全

高效執(zhí)行：優(yōu)化處理數(shù)百萬條數(shù)據(jù)的性能

可追蹤性：實時監(jiān)控處理進度和結果

技術方案設計

基礎實現(xiàn)

我們首先實現(xiàn)一個基礎版本，包含以下核心功能：

def generate_phone_prefix_suffix_pairs() -> List[Tuple[str, str]]:
    prefixes = ['157', '185', '178', '172']
    return [(prefix, f"{suffix:04d}") 
            for prefix in prefixes 
            for suffix in range(10000)]

???????def get_existing_middles(cursor, prefix: str, suffix: str) -> Set[str]:
    cursor.execute("""
        SELECT SUBSTRING(phone_number, 4, 4) 
        FROM phone_numbers 
        WHERE prefix=%s AND suffix=%s
    """, (prefix, suffix))
    return {row[0] for row in cursor.fetchall()}

def fill_missing_numbers_basic():
    conn = pymysql.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cursor:
            for prefix, suffix in generate_phone_prefix_suffix_pairs():
                existing = get_existing_middles(cursor, prefix, suffix)
                missing = {f"{i:04d}" for i in range(10000)} - existing
                
                for middle in missing:
                    phone = f"{prefix}{middle}{suffix}"
                    cursor.execute("""
                        INSERT INTO phone_numbers 
                        VALUES (%s, %s, %s, %s, %s)
                    """, (prefix, suffix, phone, "省", "市"))
        conn.commit()
    finally:
        conn.close()

問題分析

基礎版本存在幾個明顯問題：

性能低下：每條記錄單獨插入
無進度追蹤：無法知道處理進度
容錯性差：出錯后難以恢復
資源占用高：全量數(shù)據(jù)加載內存

優(yōu)化方案實現(xiàn)

1. 批量操作優(yōu)化

使用executemany實現(xiàn)批量插入，大幅提高性能：

def fill_missing_numbers_batch():
    batch_size = 1000  # 每批插入1000條
    conn = pymysql.connect(**DB_CONFIG)
    try:
        with conn.cursor() as cursor:
            for prefix, suffix in generate_phone_prefix_suffix_pairs():
                existing = get_existing_middles(cursor, prefix, suffix)
                missing = list({f"{i:04d}" for i in range(10000)} - existing)
                
                for i in range(0, len(missing), batch_size):
                    batch = missing[i:i + batch_size]
                    values = [(prefix, suffix, f"{prefix}{m}{suffix}", "省", "市") 
                             for m in batch]
                    cursor.executemany(INSERT_SQL, values)
                    conn.commit()
    finally:
        conn.close()

2. 進度監(jiān)控與日志

添加詳細的日志記錄和進度監(jiān)控：

def setup_logging():
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('fill_missing.log'),
            logging.StreamHandler()
        ]
    )

???????def log_progress(processed, total, start_time):
    if processed % 100 == 0:
        elapsed = time.time() - start_time
        remaining = (elapsed / processed) * (total - processed)
        logging.info(
            f"進度: {processed}/{total} "
            f"({processed/total:.1%}) | "
            f"預計剩余: {remaining/60:.1f}分鐘"
        )

3. 異常處理與恢復

增強異常處理和事務管理：

def fill_missing_numbers_safe():
    try:
        conn = pymysql.connect(**DB_CONFIG)
        with conn.cursor() as cursor:
            for prefix, suffix in generate_phone_prefix_suffix_pairs():
                try:
                    # 處理邏輯
                    conn.commit()
                except Exception as e:
                    conn.rollback()
                    logging.error(f"處理失敗: {prefix}{suffix}, 錯誤: {str(e)}")
                    continue
    except Exception as e:
        logging.error("程序異常終止", exc_info=True)
    finally:
        if conn: conn.close()

完整解決方案

將上述優(yōu)化整合后的完整實現(xiàn)：

import pymysql
import logging
import time
from typing import List, Tuple, Set

# 配置項
DB_CONFIG = {...}
BATCH_SIZE = 1000
LOG_INTERVAL = 100

def main():
    setup_logging()
    logging.info("開始執(zhí)行號碼補全任務")
    
    start_time = time.time()
    total = 4 * 10000  # 4前綴×10000后綴
    processed = 0
    
    try:
        conn = pymysql.connect(**DB_CONFIG)
        with conn.cursor() as cursor:
            for prefix, suffix in generate_phone_prefix_suffix_pairs():
                processed += 1
                log_progress(processed, total, start_time)
                
                try:
                    existing = get_existing_middles(cursor, prefix, suffix)
                    missing = calculate_missing(existing)
                    
                    if missing:
                        batch_insert(cursor, conn, prefix, suffix, missing)
                        
                except Exception as e:
                    handle_error(conn, prefix, suffix, e)
                    continue
                    
        log_completion(start_time, total)
        
    except Exception as e:
        logging.error("主程序異常", exc_info=True)
    finally:
        if conn: conn.close()

def batch_insert(cursor, conn, prefix, suffix, missing):
    for i in range(0, len(missing), BATCH_SIZE):
        batch = missing[i:i + BATCH_SIZE]
        values = [(prefix, suffix, f"{prefix}{m}{suffix}", "省", "市") 
                 for m in batch]
        try:
            cursor.executemany(INSERT_SQL, values)
            conn.commit()
            logging.debug(f"插入成功: {prefix}{suffix} 批次{i//BATCH_SIZE+1}")
        except Exception as e:
            conn.rollback()
            raise