Python实现将pandas经纬度样本数据按分类追加存入数据库

你的场景总共有固定10000种经纬度组合，核心需求是按经纬度键追加存储savings值，不需要上重型数据库，两个落地成本极低的方案完全覆盖需求：

• 本地单机使用：选SQLite，零服务部署，单文件存储，Python原生支持，和pandas适配度极高
• 高频写入/多服务共享：选Redis，天然支持列表结构，追加操作原子性好，内存读写延迟极低

表结构设计 #

直接建单表带联合索引即可，预留sample_id字段满足追溯需求，不用做额外分表：

CREATE TABLE IF NOT EXISTS savings_records (
    lat REAL NOT NULL,
    lng REAL NOT NULL,
    savings REAL NOT NULL,
    sample_id TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- 给经纬度建联合索引，查询时直接命中，10万级数据量下查询延迟毫秒级
CREATE INDEX IF NOT EXISTS idx_lat_lng ON savings_records(lat, lng);

对接pandas的读写代码 #

import sqlite3
import pandas as pd
from typing import Optional, List, Tuple

def write_sample(df: pd.DataFrame, sample_id: Optional[str] = None, db_path: str = "location_savings.db") -> None:
    """
    写入单份pandas样本数据
    df要求包含lat、lng、savings三列，同一份样本内的重复经纬度组合会按顺序追加存储
    """
    write_df = df.copy()
    if sample_id is not None:
        write_df["sample_id"] = sample_id
    conn = sqlite3.connect(db_path)
    try:
        # 批量写入比循环单条插入快10倍以上，40行数据写入无感知
        write_df.to_sql("savings_records", conn, if_exists="append", index=False)
    finally:
        conn.close()

def get_savings_by_location(lat: float, lng: float, with_trace: bool = False, db_path: str = "location_savings.db") -> List:
    """
    查询指定经纬度对应的savings列表
    with_trace设为True时返回结果同时带对应样本编号
    """
    conn = sqlite3.connect(db_path)
    try:
        if with_trace:
            cursor = conn.execute(
                "SELECT savings, sample_id FROM savings_records WHERE lat = ? AND lng = ? ORDER BY created_at",
                (lat, lng)
            )
            return cursor.fetchall()
        else:
            cursor = conn.execute(
                "SELECT savings FROM savings_records WHERE lat = ? AND lng = ? ORDER BY created_at",
                (lat, lng)
            )
            return [row[0] for row in cursor.fetchall()]
    finally:
        conn.close()

方案说明 #

• 完全兼容同一样本内重复经纬度的场景，重复记录不会被覆盖，严格按写入顺序追加
• 后续需要做聚合统计（比如单点位的savings均值、极值、分位数）直接写SQL即可，不需要全量拉取数据到内存计算
• 不要用本地JSON、Python字典持久化的方式存，数据量上来之后读写性能会快速下降，还容易因为进程异常退出损坏数据

如果你的样本生成频率很高，或者需要多个进程/服务同时读写数据，换Redis即可，核心是用Redis的List结构存每个经纬度对应的savings序列：

import redis
import pandas as pd
from typing import Optional, List

# 初始化redis连接，记得开持久化避免数据丢失
r = redis.Redis(host="127.0.0.1", port=6379, db=0, decode_responses=True)

def write_sample_redis(df: pd.DataFrame, sample_id: Optional[str] = None) -> None:
    pipe = r.pipeline()
    for _, row in df.iterrows():
        redis_key = f"savings:{row['lat']}:{row['lng']}"
        if sample_id:
            # 需要追溯的话把savings和sample_id拼为字符串存储
            val = f"{row['savings']}|{sample_id}"
        else:
            val = str(row["savings"])
        # rpush是原子操作，多进程同时写不会乱序
        pipe.rpush(redis_key, val)
    pipe.execute()

def get_savings_by_location_redis(lat: float, lng: float, with_trace: bool = False) -> List:
    redis_key = f"savings:{lat}:{lng}"
    raw_data = r.lrange(redis_key, 0, -1)
    if with_trace:
        return [(float(item.split("|")[0]), item.split("|")[1]) for item in raw_data]
    else:
        return [float(item.split("|")[0]) for item in raw_data]

方案说明 #

• 纯内存操作，写入和查询性能比SQLite高一个量级
• 缺点是需要单独部署Redis服务，必须开启RDB/AOF持久化配置，否则服务重启会丢失内存数据

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