Send optimal batch size

2018-07-10 19:06:49 +02:00
parent 286a75fa8e
commit 531ad28158
10 changed files with 153 additions and 90 deletions
--- a/server/extension/cdb_dataservices_server--0.32.0.sql
+++ b/server/extension/cdb_dataservices_server--0.32.0.sql
@@ -1861,7 +1861,8 @@ BEGIN
      monthly_quota NUMERIC,
      used_quota NUMERIC,
      soft_limit BOOLEAN,
-      provider TEXT
+      provider TEXT,
+      max_batch_size NUMERIC
    );
  END IF;
 END $$;
@@ -1872,6 +1873,7 @@ CREATE OR REPLACE FUNCTION cdb_dataservices_server.cdb_service_quota_info(
 RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  from cartodb_services.metrics.user import UserMetricsService
  from datetime import date
+  from cartodb_services.bulk_geocoders import BATCH_GEOCODER_CLASS_BY_PROVIDER

  plpy.execute("SELECT cdb_dataservices_server._connect_to_redis('{0}')".format(username))
  redis_conn = GD["redis_connection_{0}".format(username)]['redis_metrics_connection']
@@ -1889,7 +1891,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_isolines_config.service_type, today)
  soft_limit = user_isolines_config.soft_isolines_limit
  provider = user_isolines_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]

  #-- Hires Geocoder
  service = 'hires_geocoder'
@@ -1901,7 +1903,12 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_geocoder_config.service_type, today)
  soft_limit = user_geocoder_config.soft_geocoding_limit
  provider = user_geocoder_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  batch_geocoder_class = BATCH_GEOCODER_CLASS_BY_PROVIDER.get(provider, None)
+  if batch_geocoder_class and hasattr(batch_geocoder_class, 'MAX_BATCH_SIZE'):
+      max_batch_size = batch_geocoder_class.MAX_BATCH_SIZE
+  else:
+      max_batch_size = 1
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, max_batch_size]]

  #-- Routing
  service = 'routing'
@@ -1913,7 +1920,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_routing_config.service_type, today)
  soft_limit = user_routing_config.soft_limit
  provider = user_routing_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]

  #-- Observatory
  service = 'observatory'
@@ -1925,7 +1932,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_obs_config.service_type, today)
  soft_limit = user_obs_config.soft_limit
  provider = user_obs_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]

  return ret
 $$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED;
--- a/server/extension/sql/200_quotas.sql
+++ b/server/extension/sql/200_quotas.sql
@@ -22,7 +22,8 @@ BEGIN
      monthly_quota NUMERIC,
      used_quota NUMERIC,
      soft_limit BOOLEAN,
-      provider TEXT
+      provider TEXT,
+      max_batch_size NUMERIC
    );
  END IF;
 END $$;
@@ -33,6 +34,7 @@ CREATE OR REPLACE FUNCTION cdb_dataservices_server.cdb_service_quota_info(
 RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  from cartodb_services.metrics.user import UserMetricsService
  from datetime import date
+  from cartodb_services.bulk_geocoders import BATCH_GEOCODER_CLASS_BY_PROVIDER

  plpy.execute("SELECT cdb_dataservices_server._connect_to_redis('{0}')".format(username))
  redis_conn = GD["redis_connection_{0}".format(username)]['redis_metrics_connection']
@@ -50,7 +52,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_isolines_config.service_type, today)
  soft_limit = user_isolines_config.soft_isolines_limit
  provider = user_isolines_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]

  #-- Hires Geocoder
  service = 'hires_geocoder'
@@ -62,7 +64,12 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_geocoder_config.service_type, today)
  soft_limit = user_geocoder_config.soft_geocoding_limit
  provider = user_geocoder_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  batch_geocoder_class = BATCH_GEOCODER_CLASS_BY_PROVIDER.get(provider, None)
+  if batch_geocoder_class and hasattr(batch_geocoder_class, 'MAX_BATCH_SIZE'):
+      max_batch_size = batch_geocoder_class.MAX_BATCH_SIZE
+  else:
+      max_batch_size = 1
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, max_batch_size]]

  #-- Routing
  service = 'routing'
@@ -74,7 +81,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_routing_config.service_type, today)
  soft_limit = user_routing_config.soft_limit
  provider = user_routing_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]

  #-- Observatory
  service = 'observatory'
@@ -86,7 +93,7 @@ RETURNS SETOF cdb_dataservices_server.service_quota_info AS $$
  used_quota = user_service.used_quota(user_obs_config.service_type, today)
  soft_limit = user_obs_config.soft_limit
  provider = user_obs_config.provider
-  ret += [[service, monthly_quota, used_quota, soft_limit, provider]]
+  ret += [[service, monthly_quota, used_quota, soft_limit, provider, 1]]

  return ret
 $$ LANGUAGE plpythonu STABLE PARALLEL RESTRICTED;
--- a/server/lib/python/cartodb_services/cartodb_services/bulk_geocoders.py
+++ b/server/lib/python/cartodb_services/cartodb_services/bulk_geocoders.py
@@ -0,0 +1,11 @@
+from google import GoogleMapsBulkGeocoder
+from here import HereMapsBulkGeocoder
+from tomtom import TomTomBulkGeocoder
+from mapbox import MapboxBulkGeocoder
+
+BATCH_GEOCODER_CLASS_BY_PROVIDER = {
+    'google': GoogleMapsBulkGeocoder,
+    'heremaps': HereMapsBulkGeocoder,
+    'tomtom': TomTomBulkGeocoder,
+    'mapbox': MapboxBulkGeocoder
+}
--- a/server/lib/python/cartodb_services/cartodb_services/here/bulk_geocoder.py
+++ b/server/lib/python/cartodb_services/cartodb_services/here/bulk_geocoder.py
@@ -16,7 +16,7 @@ HereJobStatus = namedtuple('HereJobStatus', 'total_count processed_count status'

 class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
    MAX_BATCH_SIZE = 1000000  # From the docs
-    MIN_BATCHED_SEARCH = 100  # Under this, serial will be used
+    MIN_BATCHED_SEARCH = 1000  # Under this, serial will be used
    BATCH_URL = 'https://batch.geocoder.cit.api.here.com/6.2/jobs'
    # https://developer.here.com/documentation/batch-geocoder/topics/read-batch-request-output.html
    META_COLS = ['relevance', 'matchType', 'matchCode', 'matchLevel', 'matchQualityStreet']
@@ -55,14 +55,17 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
        while True:
            job_info = self._job_status(request_id)
            if job_info.processed_count == last_processed:
+                self._logger.debug('--> no progress ({})'.format(last_processed))
                stalled_retries += 1
                if stalled_retries > self.MAX_STALLED_RETRIES:
                    raise Exception('Too many retries for job {}'.format(request_id))
            else:
+                self._logger.debug('--> progress ({} != {})'.format(job_info.processed_count, last_processed))
                stalled_retries = 0
                last_processed = job_info.processed_count

-            self._logger.debug('--> Job poll check: {}'.format(job_info))
+            self._logger.debug('--> Job poll check ({}): {}'.format(
+                stalled_retries, job_info))
            if job_info.status in self.JOB_FINAL_STATES:
                break
            else:
@@ -95,7 +98,7 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
        request_params.update({
            'gen': 8,
            'action': 'run',
-            #'mailto': 'juanignaciosl@carto.com',
+            # 'mailto': 'juanignaciosl@carto.com',
            'header': 'true',
            'inDelim': '|',
            'outDelim': '|',
@@ -121,8 +124,8 @@ class HereMapsBulkGeocoder(HereMapsGeocoder, StreetPointBulkGeocoder):
                                timeout=(self.connect_timeout, self.read_timeout))
        polling_root = ET.fromstring(polling_r.text)
        return HereJobStatus(
-            total_count=polling_root.find('./Response/TotalCount').text,
-            processed_count=polling_root.find('./Response/ProcessedCount').text,
+            total_count=int(polling_root.find('./Response/TotalCount').text),
+            processed_count=int(polling_root.find('./Response/ProcessedCount').text),
            status=polling_root.find('./Response/Status').text)

    def _download_results(self, job_id):