高德API批量获取经纬度：从地理编码到线程池避坑实战

简介：面向需要批量处理地理位置的开发者，这套基于高德地图接口与Python语言的项目源码，能够自动将地址转换为对应经纬度，适用于数据采集、数据分析、地图可视化等位置服务场景。压缩包内共有9个文件，其中6个为Python脚本：分别负责调用高德接口发送地理编码请求、完成不同坐标系之间的转换、生成输入Excel表格以及读写输出结果；另附带样例数据、网页展示页面和多个可运行的演示脚本，整个包仅有15KB，结构简单清晰，方便按需摘取使用。目前已有135人学习。源码完整覆盖从读取地址列表、请求高德接口、处理返回结果到输出经纬度的主要流程，也对文件权限等常见运行问题给出了解决思路。用户替换自己的密钥后即可直接运行，也可以将相关模块嵌入更大型的数据处理项目中，从而免去逐条手工查询，降低出错风险，明显提升地理编码效率。整体上是一套适合初学者上手、也能满足日常批量坐标查询需求的轻量型工具集。

1. 高德API批量获取经纬度：这个需求藏着的四道坎，比接口本身更花时间

在位置相关的业务里，“手里有一批地址文本，想批量换成经纬度”几乎是所有可视化项目的第一道工序。高德开放平台的Web服务地理编码接口能完成地址到坐标的转换，但高德API批量获取经纬度真正让人头疼的不是请求本身，而是输入地址的脏数据、并发频控和中断后的恢复。这个需求我做过不止一次：有人用一个for循环就把三百条地址跑完了，也有人因为没做缓存和断点，项目上线后每天重复调用把配额烧光，第二天全员干等。这篇文章面向要把几百、几千条地址落成坐标的开发者，从单条请求开始，逐步讲清楚Key准备、Python批量脚本、多线程节流和五个高频踩坑，最后给一套坐标质量校验的思路。

2. 接口选型与Key准备：地理编码单次请求长什么样，配额边界在哪里

2.1 正向地理编码：把地址变成经纬度的请求参数与返回结构

所谓正向地理编码，就是传一个地址文本，接口返回对应的经纬度坐标。高德开放平台把这个能力封装在Web服务API里，接口路径是/v3/geocode/geo，请求方式为GET，单次调用就能拿到一个或多个候选坐标。

先看单条请求的完整结构。我用curl来验证，这样不依赖任何编程语言，浏览器也能直接打开：

# 用curl发起一次地理编码请求，注意替换为自己的Key # address 参数一定要URL编码，中文和特殊字符都要转义 curl -s -w "\nHTTP_CODE:%{http_code}\n" \ 'https://restapi.amap.com/v3/geocode/geo?address=%E6%9F%90%E5%B8%82%E6%9F%90%E5%8C%BA%E6%9F%90%E8%B7%AF1%E5%8F%B7&city=%E6%9F%90%E5%B8%82&key=你的Key&output=json'

这段命令里，-s表示静默模式，不输出下载进度；-w "\nHTTP_CODE:%{http_code}\n"会在响应末尾打印HTTP状态码，用来区分“接口正常返回”和“网络层失败”。address是必填参数，city选填但强烈建议带上，output默认就是json，不传也行。

返回的JSON结构是嵌套的，geocodes是一个数组，因为一个地址可能命中了多个级别不同的候选结果。下面是精简后的结构：

{ "status": "1", "info": "OK", "infocode": "10000", "count": "1", "geocodes": [ { "formatted_address": "某市某区某路1号", "country": "中国", "province": "某省", "citycode": "0001", "city": "某市", "district": "某区", "adcode": "110105", "location": "116.486409,39.921489", "level": "门牌号" } ] }

代码逻辑说明：status为1表示业务成功，为0表示失败，具体原因看info和infocode。真正有用的字段是geocodes[0].location，它是字符串形式的“经度,纬度”，用英文逗号分隔，绝大多数情况下直接split(",")就能拆出两列。level字段标注了这次匹配的精度，从“门牌号”到“道路”再到“区县”，精度逐级下降。

如果你看到count大于 1，说明存在多个候选结果。批量脚本里如果只拿第一个，就可能落到错误行政区，这个坑在第五章会重点展开。

2.2 Web服务Key的创建与权限边界：白名单和配额怎么看

走到批量这一步之前，先把Key的权限边界搞清楚，不然脚本跑一半会被各种鉴权错误打断。

常见流程是：在高德开放平台控制台里创建一个应用，然后在应用下添加Key，服务类型选择“Web服务”。这里最容易犯的错是，把用于网页JS API的Key直接拿来请求Web服务接口，Key类型不匹配，接口会返回鉴权失败。Web服务Key还需要设置IP白名单，只有白名单内的出口IP才能调用。

提示：如果跑脚本的机器是云服务器，IP白名单填服务器的公网IP；如果是本地电脑，填当前宽带的出口IP。出口IP不是固定的，换网络环境后要记得更新。

配额方面，控制台会明确展示每日调用量和并发QPS两个数字，个人开发者和企业认证后的差别不小。批量获取经纬度是一次性任务还是长期任务，决定你需不需要做企业认证。我的建议是：先拿个人Key跑通小批量，如果线上每天有几万条稳定需求，再去处理认证的事，不要一上来就追求高配额，毕竟Key的权限越大，泄露后的风险也越大。

2.3 先用curl打一次真实请求：确认链路再写程序

很多批量脚本跑出全量空坐标，问题不在代码，而在Key或参数从第一步就错了。在写Python之前，先手动执行一次curl，确认三件事：接口通不通、Key认不认、address参数解析正不正确。

# 第二次curl，换一个不带行政区前缀的地址，看city参数的作用 curl -s 'https://restapi.amap.com/v3/geocode/geo?address=%E6%9F%90%E8%B7%AF1%E5%8F%B7&city=%E6%9F%90%E5%B8%82&key=你的Key&output=json'

如果返回的count是0，说明地址文本和高德的道路库对不上。这时做两个实验：把完整地址带上行政区重新请求，或者把city参数去掉再请求，对比一下哪种返回的结果更准。这一步花五分钟，能省掉后面写代码和排错的两小时。

3. 最小批量脚本：用Python把CSV地址批量转成经纬度并落盘

3.1 地址预处理：空行、空格、冗余后缀与上下文拼接

批量场景第一步不是写请求，而是先看输入数据。实际项目里拿到的地址往往长这样：有全角空格、有换行符、有的带省市区分前缀、有的只写“某某路1号”、有的把“号”写成“.”，甚至还有不可见字符。这些脏数据直接喂给接口，匹配命中率会明显下降。

我通常先做一个清洗函数，把明显干扰项去掉，但不过度处理：

import re # clean_address：把脏地址收敛成比较规整的文本 # 核心思路：只处理空白和全半角，语义部分尽量保留交给高德去解析 def clean_address(raw: str) -> str: if not raw or not isinstance(raw, str): return "" text = raw.replace("，", ",").replace("（", "(").replace("）", ")") text = re.sub(r"\s+", "", text) # 去掉所有空白，包括全角空格和换行 text = text.replace("．", ".").replace(".", "号") # 兼容“x.”写成“x号” return text.strip()

逻辑说明：re.sub(r"\s+", "", text)把地址里的空格、全角空格、换行符一次性删干净，因为地址文本里本来就不该有空白。全角逗号替换成半角，是为了避免后面写CSV时字段被逗号干扰。把.替换成号不是通用规则，只有当你的数据源习惯把“某路1号”写成“某路1.”时才启用，默认别开。

参数说明：地址清洗没有万能规则。我一般只处理空白、全半角和个别符号，行政区前缀、道路名、门牌号这些语义信息尽量保留，让高德的解析引擎去做分词。你替接口做太多语义判断，反而容易引入错误。

3.2 单线程主流程：请求、解析、落盘的最小完整代码

清洗完地址，下面是最小可用的批量脚本。它不追求速度，先把链路跑通：

import csv import time import requests GEOCODE_URL = "https://restapi.amap.com/v3/geocode/geo" # 读取CSV里的地址列，逐条请求，结果写到一个新CSV def geocode_batch(in_csv: str, out_csv: str, key: str): with open(in_csv, "r", encoding="utf-8") as f: reader = csv.DictReader(f) fieldnames = reader.fieldnames + ["lng", "lat", "level", "geocode_status"] rows = list(reader) with open(out_csv, "w", encoding="utf-8-sig", newline="") as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for row in rows: address = row.get("address", "") row["geocode_status"] = "empty" row["lng"], row["lat"], row["level"] = "", "", "" if not address: writer.writerow(row) continue params = { "address": address, "city": row.get("city", ""), "key": key, "output": "json", } try: resp = requests.get(GEOCODE_URL, params=params, timeout=10) data = resp.json() except requests.exceptions.Timeout: row["geocode_status"] = "timeout" writer.writerow(row) continue if data.get("status") != "1" or not data.get("geocodes"): row["geocode_status"] = data.get("info", "unknown") writer.writerow(row) continue geo = data["geocodes"][0] loc = geo.get("location", "") if loc: row["lng"], row["lat"] = loc.split(",", 1) row["level"] = geo.get("level", "") row["geocode_status"] = "ok" writer.writerow(row) # 单线程限速：每请求间隔0.1秒，约10次/秒 time.sleep(0.1) if __name__ == "__main__": geocode_batch("addresses.csv", "addresses_geo.csv", key="你的Key")

代码逻辑说明：csv.DictWriter配合fieldnames可以在原CSV的每一行后面追加lng、lat、level和geocode_status四列。把geocode_status这种失败原因写进输出，而不是直接抛异常，是为了让批量任务跑完后可以针对失败行做二次处理。

参数说明：timeout=10表示单次请求最多等10秒，防止某条地址导致整个脚本卡死。time.sleep(0.1)是故意加的，把请求频率压到每秒10次以内。如果你的Key并发配额只有每秒3次，就把这个时间改成0.4秒，宁可慢一点，也不要触发频控。

3.3 location字段与level精度：拿到坐标后还要存什么

跑完上面的脚本，你手里会是这样一个CSV：原始地址、lng、lat、level、status。这里的level一定要保留，它是后面判断数据能不能用的关键。

常见level取值和对应含义我用一张表来说明：

注意：如果你拿到了level为“区县”的结果，说明高德没有解析出门牌号，location返回的是区县政府或区域中心点。这种坐标拿去算门店周边客流，结果会严重失真。处理办法是在后续分析里把level不是“门牌号”的行过滤掉，或者单独标记低精度，不要混在一张表里做空间计算。

4. 批量进阶：线程池并发、本地缓存与断点续跑

4.1 用ThreadPoolExecutor控制QPS：线程数不是越大越好

单线程脚本稳定，但处理几千条地址要跑几分钟到十几分钟，因为每条请求之间都sleep了0.1秒。批量场景要提速，常见做法是引入线程池，但线程数不能随便开。线程池的意义是减少网络IO等待，而不是无限制增加并发。

我一般会先写一个简单的限速器，再套线程池：

import threading import time # RateLimiter：固定速率限速器，保证每秒最多放行 max_calls 次 class RateLimiter: def __init__(self, max_calls: float): self.interval = 1.0 / max_calls self._lock = threading.Lock() self._last = 0.0 def wait(self): with self._lock: now = time.monotonic() gap = now - self._last if gap < self.interval: time.sleep(self.interval - gap) self._last = time.monotonic()

然后在每个线程的任务函数里调用limiter.wait()再发请求：

from concurrent.futures import ThreadPoolExecutor import requests limiter = RateLimiter(max_calls=5) # 每秒最多5次 # 单条地址的地理编码请求，带限速 def geocode_one(row, key): limiter.wait() params = { "address": row["address"], "city": row.get("city", ""), "key": key, "output": "json", } resp = requests.get(GEOCODE_URL, params=params, timeout=10) data = resp.json() # 这里继续复用上一章的解析逻辑，把结果写入row return row # 开5个线程，配合上面的5QPS限速器，实际上每秒就是5次 with ThreadPoolExecutor(max_workers=5) as pool: futures = [pool.submit(geocode_one, row, key) for row in rows] for future in futures: updated_row = future.result() # 把updated_row写入最终结果文件

逻辑说明：RateLimiter里的threading.Lock保证多线程环境下time.sleep的累计频率不会超限。max_workers=5和RateLimiter(max_calls=5)配合，就是“5个线程轮流跑，每秒总计最多5次请求”。如果你的并发配额是10，就都改成10。

参数说明：线程数不是越大越好。当线程数超过QPS限制时，多余的线程只是在排队，不会提高吞吐，反而会占用本地连接资源。经验做法是把max_workers设成不超过控制台并发配额的二分之一，留出余量给重试请求。

4.2 加一道本地缓存：相同地址不重复消耗配额

批量转坐标的项目通常不是跑一次就结束。地址库每周有新数据，如果把全部地址重新请求一遍，前两天刚跑过的几千条就白消耗配额了。我习惯在脚本里加一层本地缓存，以清洗后的地址文本为key，把坐标存到JSON文件里：

import json import os CACHE_FILE = "geo_cache.json" # 启动时加载已有缓存 def load_cache(): if os.path.exists(CACHE_FILE): with open(CACHE_FILE, "r", encoding="utf-8") as f: return json.load(f) return {} # 带缓存的查询入口 def geocode_with_cache(cache, address, city, key): cache_key = f"{address}|{city}" if cache_key in cache: return cache[cache_key] data = do_geocode(address, city, key) # 实际请求接口 if data: cache[cache_key] = data # 缓存仅在批量任务结束时统一写入，避免频繁IO return data # 批量任务跑完后调用save_cache def save_cache(cache): tmp_file = CACHE_FILE + ".tmp" with open(tmp_file, "w", encoding="utf-8") as f: json.dump(cache, f, ensure_ascii=False) os.replace(tmp_file, CACHE_FILE)

逻辑说明：缓存key把address和city拼在一起，是为了避免“某市某路1号”和“某县某路1号”互相覆盖。任务跑完再统一写盘，比每条请求都写一次JSON高效得多。用os.replace写临时文件，可以防止脚本中途崩溃把缓存文件写坏。

参数说明：这个缓存的粒度是整个地址，如果你传的地址每次都有细微差异，比如多了一个空格或者后缀，缓存命中率就会很低。所以在第3.1节的清洗函数一定要在进缓存前调用，保证同一地址的key始终一致。

4.3 断点续跑与幂等：中断后不要让结果归零

批量脚本最大的风险是跑到一半进程被杀、网络断开或者配额耗尽。如果没有进度记录，重跑一遍不仅浪费时间，还会重复扣掉已经成功过的那些请求配额。

我做断点续跑的方式很朴素：准备一个done.txt，每成功一条就把地址追加进去，重新运行时先加载这个集合，遇到已经存在的地址直接跳过。

# 已完成的地址集合，重复运行自动跳过 done_set = set() if os.path.exists("done.txt"): with open("done.txt", "r", encoding="utf-8") as f: done_set = set(line.strip() for line in f) # 在任务函数里，先判断再请求 def process_skip_done(row): if row["address"] in done_set: return None # 已处理过，跳过 updated = geocode_one(row, key) if updated and updated.get("geocode_status") == "ok": with open("done.txt", "a", encoding="utf-8") as f: f.write(row["address"] + "\n") return updated

逻辑说明：写done.txt用的是追加模式，即使脚本在第1000条崩溃，前999条记录都还在。重新运行时从第一条开始扫，遇到已在done_set里的直接跳过，不发起任何请求。这样脚本天然具备幂等性，跑多少遍都不会产生重复扣费。

提示：done.txt本质上也是缓存的一种，只不过它只记录“已完成”，不存储具体坐标。两者可以共存：缓存负责存坐标，进度文件负责去重。

5. 避坑实录：高德批量地理编码的五个高频翻车现场

5.1 现象：请求返回status=0，排查半天发现是Key的问题

现象：脚本刚启动就大面积失败，返回的info字段不是OK，而是各种鉴权或参数异常；也有一些情况是半小时后突然失败，前面正常。

原因：最常见的有三种。第一，Key类型用错了，拿的是Web端JS API的Key去请求Web服务接口；第二，IP白名单没配置或配置的不是当前出口IP；第三，Key本身写错或过期。

解决：先去控制台确认Key的“服务类型”是不是Web服务，再看“IP白名单”里有没有当前机器出口IP。可以用一个在线接口查出口IP，也可以直接在服务器上用系统命令看公网地址。改完白名单要等一两分钟生效，不要立刻重试。另外我习惯把Key放到环境变量里，不在代码里写死，方便不同环境切换，也不容易误提交到仓库。

5.2 现象：前期请求正常，跑到一半开始返回频控限制

现象：脚本开头几百条全部成功，跑到中段开始连续报错，info里出现类似“CUQPS_HAS_EXCEEDED_THE_LIMIT”的提示，重试几次还是一样。

原因：这是触发了配额限制。可能是单日总配额用完，也可能是并发QPS超了。日配额用完时，当天后续所有请求都会被拒，直到第二天的配额重置；并发超限则是瞬时的，等一小段时间能恢复。

解决：先区分是哪一种。看报错信息里的关键词，或者去控制台查看今日剩余配额。如果是日配额用完，只能等重置，或者升级认证后提高配额；如果是并发超限，说明你的线程数或请求频率超过了Key允许的QPS，调低max_workers和RateLimiter的速率即可。一个实用习惯是：批量任务开始前先记录当前配额剩余值，跑完后再看一眼，差值就是你这次实际消耗的量，这样能知道脚本离限制还有多少余量。

5.3 现象：地址明明存在，坐标却偏出几个路口，或者落在附近园区

现象：拿一条“某大厦B座”去请求，返回的status是1，坐标也是合法的经度纬度，但拉到地图上一看，位置偏到了旁边的路口甚至另一个园区。

原因：地理编码接口是按照结构化地址来解析的，它擅长处理“省+市+区+道路+门牌号”这种文本。但像“某大厦”“某产业园”这种名称型地址，本质上是POI，地理编码的词典里不一定有，只能靠相近的门牌号推断，所以坐标就会偏。

解决：对这种存量数据，不要死磕地理编码接口。把地址分成两类：结构化地址走地理编码，名称型地址走POI搜索接口。POI搜索的请求参数类似，但接口不同，返回的坐标精度要高得多。如果暂时不想换接口，可以在address参数前面拼上区级前缀，比如“某市某区某大厦”，让高德先锁定行政区，再在区内搜索，命中率会明显上升。

5.4 现象：一开线程池就大量超时和连接重置，比单线程还慢

现象：把max_workers从1改成10，预期速度翻几倍，结果大量请求超时，部分连接被重置，整个脚本跑下来耗时反而比单线程还长。

原因：这是一个典型的“并发超限雪崩”。线程数超过QPS配额后，超出的请求直接被限流，它们占着线程池里的槽位等待重试，后续请求进不来，整体吞吐反而下降。更糟的是，重试请求又叠加在已经超限的流量上，情况越来越糟。

解决：把线程数降回到配额以内，并给请求加指数退避重试。重试逻辑通常是这样的：第一次失败等2秒，第二次等4秒，第三次等8秒，最多重试两次就放弃，把这条地址标记为失败，不拖累整个队列。比例上说，宁可让线程池空着，也不要让它满载跑。

5.5 现象：同一地址两次运行结果不同，坐标相差不小

现象：同一份CSV，跑完第一次，再把结果删掉重跑，发现部分地址的经纬度变了，偏差甚至跨区。

原因：当geocodes返回多个候选时，我们默认取geocodes[0]，但这个顺序并不保证每次都是最优解。地址文本有歧义时，比如“某路1号”在多个区都存在，高德会返回多个候选，第一次可能排在前面的是A区，第二次排序权重变了，排到前面的成了B区，坐标自然就变了。

解决：批量落库时把count字段和level字段一起存下来。如果count大于1，把整条geocodes数组打印到日志里，人工抽查或写个小脚本对比几个候选的位置。同时，通过city参数把搜索范围限定在正确城市，绝大多数歧义都能消除。如果结果只是相差几百米且level都是“道路”级别，那属于匹配多义性，不一定要处理。

6. 数据落地前的最后一道工序：三条坐标质量校验规则

批量结果拿到手，先别急着建表或画图。我自己走过的弯路是：脚本看起来全跑通了，坐标格式也合法，结果落到地图上才发现有几十条落在了完全错误的方向。后来我给自己定了一套三道校验流程，每次批量结果落地前都走一遍。

第一道是范围合法性校验。虽然坐标格式是数字就能解析，但国内项目的坐标范围基本可以限定在一个合理边界内。我在代码里写死这个范围，超出直接标记异常：

def is_valid_lnglat(lng, lat): return 73 <= lng <= 135 and 18 <= lat <= 54

第二道是精度等级过滤。把所有level不是“门牌号”和“道路”的行单独提出来。这些低精度坐标不是不能用，但不能和精确坐标混在一张表里做距离计算。我一般会加一列precision_flag，标记为精确或模糊，下游分析时按需过滤。

第三道是随机抽检，用逆地理编码验证。随机抽20条结果，把坐标再拿给/v3/geocode/regeo接口，让它反查地址，再和原始地址的行政区做比对：

def reverse_check(lng, lat, key): url = "https://restapi.amap.com/v3/geocode/regeo" params = {"location": f"{lng},{lat}", "key": key, "output": "json"} resp = requests.get(url, params=params, timeout=10) data = resp.json() if data.get("status") == "1": regeo = data.get("regeocode", {}) return regeo.get("addressComponent", {}).get("district", "") return ""

抽出原始地址里的区名和反查结果里的区名做对比，偏差超过一定数量，就说明输入数据存在系统性问题，需要回到清洗环节排查，而不是继续往下游推。

这个三道校验的习惯帮我拦住过很多次“坐标数值正常但位置离谱”的脏数据，比出了问题再洗数据省事得多。希望这篇高德API批量获取经纬度的实战笔记能帮你少走几步弯路，希望帮到你。

本文还有配套的精品资源，点击获取