Add JIT meta caching

* separate meta operations from database.py into database_meta.py
* add tests

Author: dshemetov

integrations/acquisition/covidcast/test_covidcast_meta_caching.py

@@ -10,10 +10,10 @@
 
 # first party
 from delphi_utils import Nans
-from delphi.epidata.client.delphi_epidata import Epidata
 import delphi.operations.secrets as secrets
-import delphi.epidata.acquisition.covidcast.database as live
-from delphi.epidata.acquisition.covidcast.covidcast_meta_cache_updater import main
+from ....src.client.delphi_epidata import Epidata
+from ....src.acquisition.covidcast.database_meta import DatabaseMeta
+from ....src.acquisition.covidcast.covidcast_meta_cache_updater import main
 
 # py3tester coverage target (equivalent to `import *`)
 __test_target__ = (
@@ -92,7 +92,7 @@ def test_caching(self):
     self.cnx.commit()
 
     # make sure the live utility is serving something sensible
-    cvc_database = live.Database()
+    cvc_database = DatabaseMeta()
     cvc_database.connect()
     epidata1 = cvc_database.compute_covidcast_meta()
     cvc_database.disconnect(False)

integrations/acquisition/covidcast/test_db.py

@@ -1,9 +1,10 @@
 import unittest
 
 from delphi_utils import Nans
-from delphi.epidata.acquisition.covidcast.database import Database, CovidcastRow
 import delphi.operations.secrets as secrets
 
+from ....src.acquisition.covidcast.database import Database, CovidcastRow
+
 # all the Nans we use here are just one value, so this is a shortcut to it:
 nmv = Nans.NOT_MISSING.value
 

src/acquisition/covidcast/config.py

@@ -0,0 +1,3 @@
+# TODO: Fill these in.
+GEO_TYPES = ["county", "state", "hhs", "msa", "nation", "hrr"]
+ALL_TIME = "19000101-20500101"

src/acquisition/covidcast/covidcast_meta_cache_updater.py

@@ -6,7 +6,7 @@
 import time
 
 # first party
-from delphi.epidata.acquisition.covidcast.database import Database
+from .database_meta import DatabaseMeta
 from delphi.epidata.acquisition.covidcast.logger import get_structured_logger
 from delphi.epidata.client.delphi_epidata import Epidata
 
@@ -18,7 +18,7 @@ def get_argument_parser():
   return parser
 
 
-def main(args, epidata_impl=Epidata, database_impl=Database):
+def main(args, epidata_impl: Epidata = Epidata, database_impl: DatabaseMeta = DatabaseMeta):
   """Update the covidcast metadata cache.
 
   `args`: parsed command-line arguments

src/acquisition/covidcast/database.py

@@ -4,61 +4,14 @@
 """
 
 # third party
-import json
+from typing import Iterable, Sequence
 import mysql.connector
-import numpy as np
 from math import ceil
 
-from queue import Queue, Empty
-import threading
-from multiprocessing import cpu_count
-
-# first party
 import delphi.operations.secrets as secrets
 
-from delphi.epidata.acquisition.covidcast.logger import get_structured_logger
-
-class CovidcastRow():
-  """A container for all the values of a single covidcast row."""
-
-  @staticmethod
-  def fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag):
-    if row_value is None: return None
-    return CovidcastRow(source, signal, time_type, geo_type, time_value,
-                        row_value.geo_value,
-                        row_value.value,
-                        row_value.stderr,
-                        row_value.sample_size,
-                        row_value.missing_value,
-                        row_value.missing_stderr,
-                        row_value.missing_sample_size,
-                        issue, lag)
-
-  @staticmethod
-  def fromCsvRows(row_values, source, signal, time_type, geo_type, time_value, issue, lag):
-    # NOTE: returns a generator, as row_values is expected to be a generator
-    return (CovidcastRow.fromCsvRowValue(row_value, source, signal, time_type, geo_type, time_value, issue, lag)
-            for row_value in row_values)
-
-  def __init__(self, source, signal, time_type, geo_type, time_value, geo_value, value, stderr, 
-               sample_size, missing_value, missing_stderr, missing_sample_size, issue, lag):
-    self.id = None
-    self.source = source
-    self.signal = signal
-    self.time_type = time_type
-    self.geo_type = geo_type
-    self.time_value = time_value
-    self.geo_value = geo_value      # from CSV row
-    self.value = value              # ...
-    self.stderr = stderr            # ...
-    self.sample_size = sample_size  # ...
-    self.missing_value = missing_value # ...
-    self.missing_stderr = missing_stderr # ...
-    self.missing_sample_size = missing_sample_size # from CSV row
-    self.direction_updated_timestamp = 0
-    self.direction = None
-    self.issue = issue
-    self.lag = lag
+from .logger import get_structured_logger
+from .covidcast_row import CovidcastRow
 
 
 # constants for the codes used in the `process_status` column of `signal_load`
@@ -72,25 +25,32 @@ class _PROCESS_STATUS(object):
 class Database:
   """A collection of covidcast database operations."""
 
-  DATABASE_NAME = 'covid'
-
-  load_table = "signal_load"
-  latest_table = "signal_latest" # NOTE: careful!  probably want to use variable `latest_view` instead for semantics purposes
-  latest_view = latest_table + "_v"
-  history_table = "signal_history" # NOTE: careful!  probably want to use variable `history_view` instead for semantics purposes
-  history_view = history_table + "_v"
+  def __init__(self):
+    self.load_table = "signal_load"
+    self.latest_table = "signal_latest" # NOTE: careful!  probably want to use variable `latest_view` instead for semantics purposes
+    self.latest_view = self.latest_table + "_v"
+    self.history_table = "signal_history" # NOTE: careful!  probably want to use variable `history_view` instead for semantics purposes
+    self.history_view = self.history_table + "_v"
 
+    self._connector_impl = mysql.connector
+    self._db_credential_user, self._db_credential_password = secrets.db.epi
+    self._db_host = secrets.db.host
+    self._db_database = 'covid'
 
-  def connect(self, connector_impl=mysql.connector):
+  def connect(self, connector_impl=None, host=None, user=None, password=None, database=None):
     """Establish a connection to the database."""
+    self._connector_impl = connector_impl if connector_impl is not None else self._connector_impl
+    self._db_host = host if host is not None else self._db_host
+    self._db_credential_user = user if user is not None else self._db_credential_user
+    self._db_credential_password = password if password is not None else self._db_credential_password
+    self._db_database = database if database is not None else self._db_database
 
-    u, p = secrets.db.epi
-    self._connector_impl = connector_impl
     self._connection = self._connector_impl.connect(
-        host=secrets.db.host,
-        user=u,
-        password=p,
-        database=Database.DATABASE_NAME)
+        host=self._db_host,
+        user=self._db_credential_user,
+        password=self._db_credential_password,
+        database=self._db_database
+    )
     self._cursor = self._connection.cursor()
 
   def commit(self):
@@ -110,7 +70,6 @@ def disconnect(self, commit):
       self._connection.commit()
     self._connection.close()
 
-
   def count_all_rows(self, tablename=None):
     """Return the total number of rows in table `covidcast`."""
 
@@ -134,11 +93,10 @@ def count_insertstatus_rows(self):
     for (num,) in self._cursor:
       return num
 
-
-  def insert_or_update_bulk(self, cc_rows):
+  def insert_or_update_bulk(self, cc_rows: Iterable[CovidcastRow]):
     return self.insert_or_update_batch(cc_rows)
 
-  def insert_or_update_batch(self, cc_rows, batch_size=2**20, commit_partial=False):
+  def insert_or_update_batch(self, cc_rows: Sequence[CovidcastRow], batch_size: int = 2**20, commit_partial: bool = False):
     """
     Insert new rows (or update existing) into the load table.
     Data inserted this way will not be available to clients until the appropriate steps from src/dbjobs/ have run
@@ -476,131 +434,3 @@ def split_list(lst, n):
     finally:
       self._cursor.execute(drop_tmp_table_sql)
     return total
-
-
-  def compute_covidcast_meta(self, table_name=None):
-    """Compute and return metadata on all COVIDcast signals."""
-    logger = get_structured_logger("compute_covidcast_meta")
-
-    if table_name is None:
-      table_name = self.latest_view
-
-    n_threads = max(1, cpu_count()*9//10) # aka number of concurrent db connections, which [sh|c]ould be ~<= 90% of the #cores available to SQL server
-    # NOTE: this may present a small problem if this job runs on different hardware than the db,
-    #       but we should not run into that issue in prod.
-    logger.info(f"using {n_threads} workers")
-
-    srcsigs = Queue() # multi-consumer threadsafe!
-    sql = f'SELECT `source`, `signal` FROM `{table_name}` GROUP BY `source`, `signal` ORDER BY `source` ASC, `signal` ASC;'
-    self._cursor.execute(sql)
-    for source, signal in self._cursor:
-      srcsigs.put((source, signal))
-
-    inner_sql = f'''
-      SELECT
-        `source` AS `data_source`,
-        `signal`,
-        `time_type`,
-        `geo_type`,
-        MIN(`time_value`) AS `min_time`,
-        MAX(`time_value`) AS `max_time`,
-        COUNT(DISTINCT `geo_value`) AS `num_locations`,
-        MIN(`value`) AS `min_value`,
-        MAX(`value`) AS `max_value`,
-        ROUND(AVG(`value`),7) AS `mean_value`,
-        ROUND(STD(`value`),7) AS `stdev_value`,
-        MAX(`value_updated_timestamp`) AS `last_update`,
-        MAX(`issue`) as `max_issue`,
-        MIN(`lag`) as `min_lag`,
-        MAX(`lag`) as `max_lag`
-      FROM
-        `{table_name}`
-      WHERE
-        `source` = %s AND
-        `signal` = %s
-      GROUP BY
-        `time_type`,
-        `geo_type`
-      ORDER BY
-        `time_type` ASC,
-        `geo_type` ASC
-      '''
-
-    meta = []
-    meta_lock = threading.Lock()
-
-    def worker():
-      name = threading.current_thread().name
-      logger.info("starting thread", thread=name)
-      #  set up new db connection for thread
-      worker_dbc = Database()
-      worker_dbc.connect(connector_impl=self._connector_impl)
-      w_cursor = worker_dbc._cursor
-      try:
-        while True:
-          (source, signal) = srcsigs.get_nowait() # this will throw the Empty caught below
-          logger.info("starting pair", thread=name, pair=f"({source}, {signal})")
-          w_cursor.execute(inner_sql, (source, signal))
-          with meta_lock:
-            meta.extend(list(
-              dict(zip(w_cursor.column_names, x)) for x in w_cursor
-            ))
-          srcsigs.task_done()
-      except Empty:
-        logger.info("no jobs left, thread terminating", thread=name)
-      finally:
-        worker_dbc.disconnect(False) # cleanup
-
-    threads = []
-    for n in range(n_threads):
-      t = threading.Thread(target=worker, name='MetacacheThread-'+str(n))
-      t.start()
-      threads.append(t)
-
-    srcsigs.join()
-    logger.info("jobs complete")
-    for t in threads:
-      t.join()
-    logger.info("all threads terminated")
-
-    # sort the metadata because threaded workers dgaf
-    sorting_fields = "data_source signal time_type geo_type".split()
-    sortable_fields_fn = lambda x: [(field, x[field]) for field in sorting_fields]
-    prepended_sortables_fn = lambda x: sortable_fields_fn(x) + list(x.items())
-    tuple_representation = list(map(prepended_sortables_fn, meta))
-    tuple_representation.sort()
-    meta = list(map(dict, tuple_representation)) # back to dict form
-
-    return meta
-
-
-  def update_covidcast_meta_cache(self, metadata):
-    """Updates the `covidcast_meta_cache` table."""
-
-    sql = '''
-      UPDATE
-        `covidcast_meta_cache`
-      SET
-        `timestamp` = UNIX_TIMESTAMP(NOW()),
-        `epidata` = %s
-    '''
-    epidata_json = json.dumps(metadata)
-
-    self._cursor.execute(sql, (epidata_json,))
-
-  def retrieve_covidcast_meta_cache(self):
-    """Useful for viewing cache entries (was used in debugging)"""
-
-    sql = '''
-      SELECT `epidata`
-      FROM `covidcast_meta_cache`
-      ORDER BY `timestamp` DESC
-      LIMIT 1;
-    '''
-    self._cursor.execute(sql)
-    cache_json = self._cursor.fetchone()[0]
-    cache = json.loads(cache_json)
-    cache_hash = {}
-    for entry in cache:
-      cache_hash[(entry['data_source'], entry['signal'], entry['time_type'], entry['geo_type'])] = entry
-    return cache_hash