地理编码 如何在10分钟内将25万个地址绑定到坐标?



哈Ha!

在本文中,我将分享我在解决大量地址小的问题方面的经验。如果您曾经使用过地理编码API或使用过在线工具,那么我认为您要等几个小时甚至更长的时间来分享我的痛苦。

这与复杂的优化算法无关,而与使用数据包地理编码服务有关,该服务将地址列表作为输入并返回包含结果的文件。这样可以将处理时间从几小时减少到几分钟。

首先要注意的是:


背景


任务到达了-“绑定到24000个地址的坐标。” 解决问题的方法只有两种:

  1. 用于大学的地理编码Web应用程序;
  2. 根据地理编码器的REST API编写脚本。

在第一种情况下,事实证明Web应用程序在处理了数千个地址后崩溃。在同事之间分配数据集是一个被立即放弃的想法。

因此,您需要使用地址解析器的REST API来编写自己的脚本,并保存结果(这不是完全合法的方式,您需要阅读该服务的使用条款)。出现了一个新问题-当我们在应用程序中使用地址搜索并立即获得结果时,这是一回事,但是当任务是要处理并保存超过一万个地址时,脚本的工作就会大大延迟。您可以等待一两个小时,但是一百万个地址将必须对“大量时间”进行地址解析,因此您需要寻找另一种解决方案!

除了通常的地理编码服务外,大型地理定位服务提供商还提供了一个分组地理编码器(Batch Geocoder),目的是为了在一个请求中处理大量地址。

批量地理编码


服务的名称说明一切-我们有一个程序包(例如,一个csv文件,其中包含以表格形式的地址列表),然后将其上传到服务器,它会为我们完成所有工作。

该过程如下所示:

  1. 准备数据集,以便服务无误地接受它;
  2. 设置工作结果的参数(选择列,分隔符...);
  3. 将文件上传到云;
  4. 等待处理完成;
  5. 下载完成的文件。

多亏了云计算能力,在1小时内用自写脚本完成的工作在1分钟内完成。

下一步是选择使用分组地理编码器的最忠实使用条款的公司。首先,并非每个人都提供这样的服务,其他人则允许您对服务进行严格的限制。另外,如果您的交易量很大,则需要注意额外交易的费用,以防超出免费套餐的限制。

选择批处理地理编码服务提供商


在全球地理定位服务市场中,领先地位被下列公司占据:

  • 谷歌地图
  • HERE Technologies;
  • 地图框
  • 汤姆汤姆
  • ESRI。

当然,您不应忘记Yandex Technologies,该公司在俄罗斯具有相当强的地位。

我将以下参数作为选择提供程序的基础:

  • 每月对地理编码服务的请求数量是免费的;
  • 每天交易数量的限制;
  • 批量地理编码服务的可用性;
  • 可以在免费计划中使用数据包地址解析器。

每个公司都有自己的获利模型。根据项目的不同,一个或另一个模型可以发挥作用,反之亦然。

谷歌地图


要开始使用Google地理服务,您需要做的第一件事就是将信用卡信息添加到您的帐户中。每月限额为200虚拟美元,然后从链接的卡中支付其他交易。在此限制内,您可以使用各种服务,但每个事务的处理方式都不相同。例如,一千个地理编码请求的价格为5美元,但路由构建服务的价格却是其两倍。可以在该站点上找到更多详细信息,我们仅对地理编码服务感兴趣。

如果每月200美元,那么很容易计算出免费交易数量-40,000(地理编码服务)。服务之间没有数据包地址解析器。这意味着您必须编写自己的脚本,结果将是每秒大约1个地址,对于24,000个地址来说,这是6个小时。为了加快该过程,您可以尝试在Google Cloud API平台上运行该脚本,但是我决定寻找替代解决方案。每天的交易数量没有限制,因此一次可以花费4万笔。

这里技术


过去,诺基亚地图以及更深层次的Navteq每月免费提供25万笔交易。与Google Maps类似,此数字适用于所有服务,并且每个服务的使用方式都不同。使用免费软件包时,不需要附加银行卡。如果超出限额,则每增加一千次交易,您需要支付1美元。

将数据包地址解析器作为一项单独的服务(包含在免费计划中)非常重要。根据与通常相同的模型来考虑其中的事务,也就是说,数据包地址解析器将在一次事务中感知文件中的每个地址。
根据文章的标题,很明显,我使用了HERE批处理地址解析器,因为您可以将所有交易都花在地址解析器上,每月执行25万次地址解析操作。但这不是唯一的选择,因此我们看一下其他公司拥有的东西。

地图框


使用MapBox地理编码器时,每月可处理10万笔交易。该公司遵循相同的货币化模型,但要为其他交易付费。只有“批发商”有一个有趣的选择-您进行的交易越多,花费越少(当然,存在降价限制)。例如,从10万到50万,额外的一千个请求将花费0.75美元,从50万到100万-0.60美元,等等,有关更多信息,请参见网站。不幸的是,批处理地理编码器仅在付费帐户中可用。

汤姆


该平台可以每天进行2500次交易,每月大约75,000次。在测试和开发过程中,与竞争对手相比,每日限额看起来并不吸引人,但是额外交易的付款最为灵活。对于额外的一千个请求,有8种付款方式,价格从$ 0.5降低到$ 0.42。

在这些服务中,有一个批处理地址解析器,每个请求最多可以处理1万个地址(但是,必须考虑每日限制)。

Yandex技术


Yandex每日交易限额的模型,但忠诚度达25,000个请求。如果将此数字乘以一个月中的天数,您将得到令人印象深刻的75万。该网站以卢布的形式提供了另外一千笔交易的价格,范围从120卢布不等。高达11卢布

数据包地址解析器不作为服务提供,因此实现某种优化将失败。

ESRI


一个非常诱人的免费计划,每月有100万笔交易。该公司还向每个帐户收取50个信用(约等于5美元)。值得注意的是,这是使用地理定位服务的最忠诚计划。还有一个批处理地理编码服务,但是只有在ArcGIS Online平台上拥有公司帐户的情况下,您才可以使用它。

到底要选择什么?


最简单的方法是通过编译一张小表进行选择:



结果,我的选择落在了这里,因为这是解决我的问题的最佳选择。当然,我还没有完成完整的分析,理想情况下,您需要通过所有地理编码器运行数据集以评估质量。另外,如果您有数百万个地址,则应考虑一个有偿包裹,然后需要考虑添加费用。交易。

本文的目的不是比较公司,而是解决优化大量地址的地理编码的问题。我只是在选择服务提供商时表达了自己的想法。

Python服务指南


首先,您需要在门户网站上为开发人员创建一个帐户,并在项目部分中生成REST API KEY。

现在您可以使用该平台了。我将仅描述HERE数据包地理编码器具有的部分功能:数据加载,状态检查,保存结果。

因此,让我们从导入必要的库开始:

import requests
import json
import time
import zipfile
import io
from bs4 import BeautifulSoup

此外,如果没有发生错误,请创建一个类:

class Batch:

    SERVICE_URL = "https://batch.geocoder.ls.hereapi.com/6.2/jobs"
    jobId = None

    def __init__(self, apikey="your_api_key"):
        self.apikey = apikey

也就是说,在初始化期间,该类必须为REST API传递其自己的密钥。
SERVICE_URL变量是用于批处理地理编码服务的基本URL。
并且在jobId中,将存储地址解析器当前工作的标识符。

重要的一点是应要求提供正确的数据结构。该文件必须包含两列必需的列:recId和searchText。否则,服务将返回包含有关下载错误信息的响应。

这是一个示例数据集:

   recId; searchText
   1; -, . , 6
   2; ,  1,  -., 72
   3; 425 W Randolph St Chicago IL 60606
   4; , DJ106 20-30, Sibiu 557260
   5; 200 S Mathilda Ave Sunnyvale CA 94086
  

用于将文件上传到云的功能:

def start(self, filename, indelim=";", outdelim=";"):
        
        file = open(filename, 'rb')

        params = {
            "action": "run",
            "apiKey": self.apikey,
            "politicalview":"RUS",
            "gen": 9,
            "maxresults": "1",
            "header": "true",
            "indelim": indelim,
            "outdelim": outdelim,
            "outcols": "displayLatitude,displayLongitude,locationLabel,houseNumber,street,district,city,postalCode,county,state,country",
            "outputcombined": "true",
        }

        response = requests.post(self.SERVICE_URL, params=params, data=file)
        self.__stats (response)
        file.close()


一切都非常简单,打开一个包含要读取的地址列表的文件,形成GET请求参数的字典。一些参数值得解释:

  • “操作”:“运行”-开始地址处理;
  • “politicalView”: “RUS” – . ( );
  • “gen”: 9 – ( );
  • “maxresults”: 1 – ;
  • “header”: true – ;
  • “indelim”: “;” – , ;
  • “outdelim”: “;” – ;
  • “outcols”: “” – , ;
  • “outcombined”: true – .

接下来,只需使用请求库发送请求并显示统计信息即可。当然,您需要在函数末尾关闭文件。__stats函数解析服务器的响应,该响应包含正在运行的工作的ID,并且还显示有关该操作的常规信息。

下一步是检查工作状态。该请求以类似的方式形成,只需要传输操作ID。动作参数必须包含值“状态”。__stats函数向控制台显示完整的统计信息,以估计地址解析器的关闭时间。

    def status (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        statusUrl = self.SERVICE_URL + "/" + self.jobId
        
        params = {
            "action": "status",
            "apiKey": self.apikey,
        }
        
        response = requests.get(statusUrl, params=params)
        self.__stats (response)

最重要的功能之一就是保存结果。为了方便起见,最好立即解压缩来自服务器的文件。保存文件的请求与检查状态相同,只是在最后添加/结果。

    def result (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        print("Requesting result data ...")
        
        resultUrl = self.SERVICE_URL + "/" + self.jobId + "/result"
        
        params = {
            "apiKey": self.apikey
        }
        
        response = requests.get(resultUrl, params=params, stream=True)
        
        if (response.ok):    
            zipResult = zipfile.ZipFile(io.BytesIO(response.content))
            zipResult.extractall()
            print("File saved successfully")
        
        else:
            print("Error")
            print(response.text)

解析服务响应的最终功能。她的任务也是保存当前地理编码任务的标识符:

    def __stats (self, response):
        if (response.ok):
            parsedXMLResponse = BeautifulSoup(response.text, "lxml")

            self.jobId = parsedXMLResponse.find('requestid').get_text()
            
            for stat in parsedXMLResponse.find('response').findChildren():
                if(len(stat.findChildren()) == 0):
                    print("{name}: {data}".format(name=stat.name, data=stat.get_text()))

        else:
            print(response.text)

要对其进行测试,只需在脚本文件夹中运行Python解释器。Batch类位于geocoder.py文件中

>>> from geocoder import Batch
>>> service = Batch(apikey="   REST API")
>>> service.start("big_data_addresses.csv", indelim=";", outdelim=";")

requestid: "  Id "
status: accepted
totalcount: 0
validcount: 0
invalidcount: 0
processedcount: 0
pendingcount: 0
successcount: 0
errorcount: 0


伟大的工作开始了。检查状态:

>>> service.status()

requestid: "  Id "
status: completed
jobstarted: 2020-04-27T10:09:58.000Z
jobfinished: 2020-04-27T10:17:18.000Z
totalcount: 249999
validcount: 249999
invalidcount: 0
processedcount: 249999
pendingcount: 0
successcount: 249978
errorcount: 21

我们看到数据集的处理已经完成。在短短7分钟内,就可以对25万个地址进行编码(不包括错误-errorcount)。保留结果即可:

>>> service.result()
Requesting result data ...
File saved successfully

完整批次类别说明


我认为完全添加脚本没有什么坏处:

import requests
import json
import time
import zipfile
import io
from bs4 import BeautifulSoup

class Batch:

    SERVICE_URL = "https://batch.geocoder.ls.hereapi.com/6.2/jobs"
    jobId = None

    def __init__(self, apikey="   REST API "):
        self.apikey = apikey
        
            
    def start(self, filename, indelim=";", outdelim=";"):
        
        file = open(filename, 'rb')

        params = {
            "action": "run",
            "apiKey": self.apikey,
            "politicalview":"RUS",
            "gen": 9,
            "maxresults": "1",
            "header": "true",
            "indelim": indelim,
            "outdelim": outdelim,
            "outcols": "displayLatitude,displayLongitude,locationLabel,houseNumber,street,district,city,postalCode,county,state,country",
            "outputcombined": "true",
        }

        response = requests.post(self.SERVICE_URL, params=params, data=file)
        self.__stats (response)
        file.close()
    

    def status (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        statusUrl = self.SERVICE_URL + "/" + self.jobId
        
        params = {
            "action": "status",
            "apiKey": self.apikey,
        }
        
        response = requests.get(statusUrl, params=params)
        self.__stats (response)
        

    def result (self, jobId = None):

        if jobId is not None:
            self.jobId = jobId
        
        print("Requesting result data ...")
        
        resultUrl = self.SERVICE_URL + "/" + self.jobId + "/result"
        
        params = {
            "apiKey": self.apikey
        }
        
        response = requests.get(resultUrl, params=params, stream=True)
        
        if (response.ok):    
            zipResult = zipfile.ZipFile(io.BytesIO(response.content))
            zipResult.extractall()
            print("File saved successfully")
        
        else:
            print("Error")
            print(response.text)
    

    
    def __stats (self, response):
        if (response.ok):
            parsedXMLResponse = BeautifulSoup(response.text, "lxml")

            self.jobId = parsedXMLResponse.find('requestid').get_text()
            
            for stat in parsedXMLResponse.find('response').findChildren():
                if(len(stat.findChildren()) == 0):
                    print("{name}: {data}".format(name=stat.name, data=stat.get_text()))

        else:
            print(response.text)

结果分析


结果,我从缓慢运行的在线应用程序转到批处理地理编码服务。地理服务提供商的选择完全取决于您面临的任务。我经常收到处理大量地址的请求,而本文中描述的方法有助于大大减少时间。

我希望这篇文章对您有所帮助,当然我也欢迎您提出意见和补充!

All Articles