基于Pandas数据框用Google Maps API获取路口经纬度遇NaN问题求助

问题核心 #

缺失N/E/W/S方位标识的路口会让Google Geocoding API无法准确定位（比如芝加哥Lake St分东西段），要么返回NaN，要么返回其他城市的错误坐标；数十万条数据必须通过自动化方式批量处理。

解决方案步骤 #

1. 先优化地理编码请求：限定城市范围 #

首先在请求中明确添加目标城市（比如Chicago, IL），减少API的歧义判断，优先返回目标城市内的结果，避免出现跨城市的错误坐标。

修改地理编码函数：

import pandas as pd
import numpy as np
import requests
import json
import re

def geocode_intersection(api_key, intersection, city="Chicago, IL"):
    base_url = "https://maps.googleapis.com/maps/api/geocode/json?"
    # 拼接地址+城市，缩小搜索范围
    full_address = f"{intersection}, {city}"
    params = {"address": full_address, "key": api_key}
    response = requests.get(base_url, params=params)
    
    if response.status_code == 200:
        data = json.loads(response.text)
        if data["status"] == "OK":
            location = data["results"][0]["geometry"]["location"]
            # 返回格式化地址，用于后续提取方位
            formatted_addr = data["results"][0]["formatted_address"]
            return location["lat"], location["lng"], formatted_addr
        else:
            print(f"API状态错误: {data['status']} | 路口: {intersection}")
            return None, None, None
    else:
        print(f"HTTP错误 {response.status_code} | 路口: {intersection}")
        return None, None, None

2. 批量补全缺失的方位标识 #

针对首次请求失败的路口，通过两种方式自动化补全方位：

方式A：利用API返回的格式化地址反向补全

当API成功返回结果时，从格式化地址中提取方位信息，更新原始路口名称，后续遇到相同路口可直接使用补全后的名称：

# 初始化数据列
busiest_stations_subs["latitude"] = np.nan
busiest_stations_subs["longitude"] = np.nan
busiest_stations_subs["formatted_address"] = ""

api_key = "YOUR_API_KEY"

# 第一轮：批量请求所有路口，记录有效结果和格式化地址
for index, row in busiest_stations_subs.iterrows():
    if pd.isna(row["latitude"]):
        lat, lng, formatted_addr = geocode_intersection(api_key, row["from_station_name"])
        if lat is not None:
            busiest_stations_subs.at[index, "latitude"] = lat
            busiest_stations_subs.at[index, "longitude"] = lng
            busiest_stations_subs.at[index, "formatted_address"] = formatted_addr

# 补全原始路口的方位信息
def extract_and_update_direction(original_name, formatted_addr):
    streets = original_name.split(" & ")
    for street in streets:
        # 匹配格式化地址中的带方位街道（比如W Lake St）
        match = re.search(rf"(N|E|W|S) {street}", formatted_addr)
        if match:
            original_name = original_name.replace(street, match.group(0))
    return original_name

# 对有格式化地址但原始名称缺方位的行进行更新
mask = (busiest_stations_subs["formatted_address"] != "") & (~busiest_stations_subs["from_station_name"].str.contains(r"N|E|W|S"))
busiest_stations_subs.loc[mask, "from_station_name"] = busiest_stations_subs.apply(
    lambda x: extract_and_update_direction(x["from_station_name"], x["formatted_address"]), axis=1
)

方式B：枚举方位重试请求

针对首次请求返回NaN的路口，尝试给街道添加N/E/W/S方位的所有组合，重试请求直到找到有效结果：

def retry_with_directions(api_key, intersection, city="Chicago, IL"):
    streets = intersection.split(" & ")
    directions = ["N", "E", "W", "S"]
    
    # 尝试两条街道都加方位的组合
    for dir1 in directions:
        for dir2 in directions:
            modified = f"{dir1} {streets[0]} & {dir2} {streets[1]}"
            lat, lng, _ = geocode_intersection(api_key, modified, city)
            if lat is not None:
                return lat, lng, modified
        # 尝试只给第一条街道加方位
        modified = f"{dir1} {streets[0]} & {streets[1]}"
        lat, lng, _ = geocode_intersection(api_key, modified, city)
        if lat is not None:
            return lat, lng, modified
    # 尝试只给第二条街道加方位
    for dir2 in directions:
        modified = f"{streets[0]} & {dir2} {streets[1]}"
        lat, lng, _ = geocode_intersection(api_key, modified, city)
        if lat is not None:
            return lat, lng, modified
    return None, None, None

# 处理所有NaN的行
nan_mask = pd.isna(busiest_stations_subs["latitude"])
for index, row in busiest_stations_subs[nan_mask].iterrows():
    lat, lng, modified_addr = retry_with_directions(api_key, row["from_station_name"])
    if lat is not None:
        busiest_stations_subs.at[index, "latitude"] = lat
        busiest_stations_subs.at[index, "longitude"] = lng
        busiest_stations_subs.at[index, "from_station_name"] = modified_addr

3. 针对数十万条数据的性能优化 #

• 批量请求：API支持最多50个地址的批量请求，减少请求次数：

def batch_geocode(api_key, addresses, city="Chicago, IL"):
    base_url = "https://maps.googleapis.com/maps/api/geocode/json?"
    batch_addrs = "|".join([f"{addr}, {city}" for addr in addresses])
    params = {"address": batch_addrs, "key": api_key}
    response = requests.get(base_url, params=params)
    
    results = []
    if response.status_code == 200:
        data = json.loads(response.text)
        for result in data["results"]:
            if result["status"] == "OK":
                loc = result["geometry"]["location"]
                results.append((loc["lat"], loc["lng"], result["formatted_address"]))
            else:
                results.append((None, None, None))
    return results

# 按每50条分组处理
batch_size = 50
for i in range(0, len(busiest_stations_subs), batch_size):
    batch = busiest_stations_subs.iloc[i:i+batch_size]
    addrs = batch["from_station_name"].tolist()
    batch_results = batch_geocode(api_key, addrs)
    for idx, (lat, lng, formatted_addr) in enumerate(batch_results):
        row_idx = batch.index[idx]
        if lat is not None:
            busiest_stations_subs.at[row_idx, "latitude"] = lat
            busiest_stations_subs.at[row_idx, "longitude"] = lng
            busiest_stations_subs.at[row_idx, "formatted_address"] = formatted_addr

• 缓存结果：用字典缓存已成功编码的路口，避免重复请求：

geocode_cache = {}

def geocode_with_cache(api_key, intersection, city="Chicago, IL"):
    if intersection in geocode_cache:
        return geocode_cache[intersection]
    result = geocode_intersection(api_key, intersection, city)
    geocode_cache[intersection] = result
    return result

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