Python2环境下MySQL多表百万行数据分块提取与查询优化方案咨询

核心优化思路 #

通过按主键分块切割查询范围的方式，把单次数百万行的大查询拆分为数十到数百个小批量查询，每块查询完成后直接写入磁盘释放内存，既避免一次性加载全量数据导致内存溢出，也能减少MySQL单次查询的关联、聚合压力，大幅缩短整体耗时。

分块处理具体实现（优先级最高） #

前置准备 #

首先调整MySQL会话参数，避免group_concat结果被截断：

SET SESSION group_concat_max_len = 1024000; -- 可根据实际拼接长度调整

分块逻辑选择 #

选择主表v的主键id作为分块键，所有关联逻辑都以v.id为核心，分块扫描性能最优，有两种实现可选：

• id连续场景：按固定id范围切割，WHERE v.id BETWEEN {start_id} AND {end_id}
• id不连续场景：按上一批次最大id分页，WHERE v.id > {last_max_id} ORDER BY v.id LIMIT {block_size}，避免空查询浪费资源

Python2 分块查询示例代码 #

import MySQLdb
# 可选流式游标，进一步降低内存占用
# import MySQLdb.cursors

# 初始化数据库连接
conn = MySQLdb.connect(host="你的数据库地址", user="账号", passwd="密码", db="库名", charset="utf8")
# 普通游标，单块数据量小的场景够用
cursor = conn.cursor()
# 内存压力大时用流式游标，结果集不会一次性加载到内存
# cursor = conn.cursor(MySQLdb.cursors.SSCursor)

# 配置参数
v1 = "你的主表名"
sq = "你的自定义HAVING条件"
block_size = 2000 # 可在1000-5000区间压测调整最优值
output_path = "./result.csv"

# 获取主表id边界
cursor.execute("SELECT MIN(id), MAX(id) FROM {}".format(v1))
min_id, max_id = cursor.fetchone()
current_id = min_id

# 逐块查询直接写入文件，内存只存单块数据
with open(output_path, "w") as f:
    # 先写表头，可根据实际返回字段调整
    headers = ["varid","chrom","vcf_pos","vcf_ref","vcf_alt","HPO_terms","HPO_names","family_label","AF_Pat","analysistypelist","AC","HAC","TAAC","AF_Assay","HCC","HCC1","TTCC","AF_Control","gene_name","tx_name","AllTranscriptAnnotations"]
    f.write(",".join(headers) + "\n")

    while current_id <= max_id:
        end_id = current_id + block_size - 1
        # 拼接分块查询SQL，其余关联逻辑保持不变
        query = """
        SELECT
            v.id AS varid,
            v.chrom AS chrom,
            v.vcf_pos AS vcf_pos,
            v.vcf_ref AS vcf_ref,
            v.vcf_alt AS vcf_alt,
            group_concat(distinct term.term) as HPO_terms,
            group_concat(distinct term.name) as HPO_names,
            group_concat(distinct pp.patientid,"-",pp.person_status,"-",pp.affected_status) as family_label,
            if(vcc.AF_Pat between 0 AND 1, vcc.AF_Pat, NULL) AS AF_Pat,
            replace(vcc.analysistypelist,',',';') AS analysistypelist,
            vcc2.HomPatCount AS AC,
            vcc2.HetPatCount AS HAC,
            vcc2.TotalPatCount AS TAAC,
            if(vcc2.AF_Pat between 0 AND 1, vcc2.AF_Pat, NULL) AS AF_Assay,
            vcc3.HomUnaffCount AS HCC,
            vcc3.HetUnaffCount AS HCC1,
            vcc3.TotalUnaffCount AS TTCC,
            if(vcc3.AF_healthy between 0 AND 1, vcc3.AF_healthy, NULL) AS AF_Control,
            g.gene_name as gene_name,
            t.tx_name as tx_name,
            group_concat(
                concat_ws(':', ifnull(g.gene_name,'.'), ifnull(t.tx_name,'.'), ifnull(ta.hgvsc,'.'), ifnull(ta.hgvsp,'.'))
                SEPARATOR '|'
            ) as `AllTranscriptAnnotations`
        FROM {} AS v
            LEFT JOIN table1 vcc ON vcc.variant_id=v.id
            LEFT JOIN table2 vcc3 ON vcc3.variant_id=v.id
            LEFT JOIN table3 va on v.id=va.variant_id and va.status='active'
            LEFT JOIN table4 vc on v.id=vc.variant_id
            LEFT JOIN table5 ta on v.id=ta.variant_id and ta.status='active'
            LEFT JOIN table6 t on t.id=ta.transcript_id and t.status='active'
            LEFT JOIN table7 g on g.id=t.gene_id and g.status='active'
            -- 剩余14张表关联逻辑保持不变
            LEFT JOIN table21 pt on pt.term_id=term.id
        WHERE v.id BETWEEN %s AND %s
        GROUP BY v.id, s.id
        HAVING 1 {}
        """.format(v1, sq)
        cursor.execute(query, (current_id, end_id))
        results = cursor.fetchall()
        # 逐行写入文件，处理逗号、空值等特殊情况
        for row in results:
            row_str = ",".join([str(item).replace(",", "，") if item is not None else "" for item in row])
            f.write(row_str + "\n")
        # 推进分块边界
        current_id = end_id + 1

cursor.close()
conn.close()

SQL层面辅助优化 #

• 所有关联字段加索引：给table1.variant_id、table3(variant_id, status)、table5(variant_id, status)、table6(id, status)、table7(id, status)等所有关联用到的外键、过滤字段建联合索引，避免全表扫描。
• 精简查询字段：原SQL中的ta.*、va.*等通配符会拉取大量无用字段，明确列出需要的字段即可，减少数据传输和内存占用。
• 小表提前过滤：对有状态过滤的小表，先做子查询过滤再关联，比如LEFT JOIN (SELECT id, tx_name FROM table6 WHERE status='active') t ON t.id=ta.transcript_id，减少关联的数据量。

进阶提速方案 #

如果需要进一步缩短耗时，可以把全量id范围切割为多个不重叠的区间，开多进程并行处理不同区间的查询，处理完成后合并结果文件即可，性能可以随核心数线性提升。

内容的提问来源于stack exchange，提问作者ARJ