Missing Data Restore

Data Insert - Final Design and Improvements

The final design after the missing data rework has greatly improved on the HDF file processing speed.

1. Linux shell script that is run every day at 20:00 machine time (Eastern for Microcenter and UTC for VM) via crontab:

   1. Use the 3rd Party tool sshpass to download FTP files from the remote server.
   2. Start the Django Management Command to process downloaded files.

2. Python code, for each HDF file:

   1. Prepare the dataframe to be inserted (I don't know if this is documented elsewhere, but I didn't touch this logic at all).

   2. Define an iterator/bulk insert size and use df.to_sql() to perform fast inserts.

   3. The callback method for doing inserts, flows_upsert_copy, is the main area of improvements. The query was re-written to allow for bulk upserts of data. The new design creates a temporary table into which all values are inserted, then the values are copied over into the existing table from the temporary table, allowing for upsert functionality. The temporary table is always deleted after the end of the transaction.

      bulk_upsert_sql = f'''
        BEGIN;
        CREATE TEMP TABLE tmp_tbl_upsert
        (LIKE {table_name} INCLUDING ALL)
        ON COMMIT DROP;
      
        COPY tmp_tbl_upsert ({columns}) FROM STDIN WITH CSV;
      
        INSERT INTO {table_name} ({columns}) SELECT {columns} FROM tmp_tbl_upsert
        ON CONFLICT (time, station_id) DO UPDATE
          SET flow = excluded.flow,
            velocity = excluded.velocity,
            stage = excluded.stage;
        COMMIT;
      '''
      cur.copy_expert(sql=bulk_upsert_sql, file=s_buf)

   4. After upserting the flows, the daily values are upserted as well using a similarly reworked method (df.to_sql with specific SQL query allowing bulk upserts).

Data Insert - First Attempt

The first turned out to not be the most optimal path.

On the first attempt of restoring the missing data, we had designed the following process:

1. Create a schema-only copy of the two flows tables (riverflow and dailyriverflow).
2. Process local HDF files in reverse order.
3. While processing each file, keep track of the first available datetime in that flows file and return it to the main process.
4. Pass the first available datetime (min_timestamp) from each processed file into each subsequent (earlier) file.
5. Trim the dataframes to drop everything after the min_timestamp and insert the remaining data.
6. Insert with df.to_sql()

Schema-only Table Copy

-- NOTE: 'INCLUDING ALL' is required to get a full table reconstruction
--       (including constraints and other pieces) instead of a
--       simple structure copy.
CREATE TABLE new_table_name (LIKE old_table_name INCLUDING ALL);

Daily Flows Calculation

During the missing data processing, it was discovered that the existing logic would not properly calculate daily flows if starting from "day 0". This means, if we're calculating daily flows from the start of a given HDF-derived dataset, the values will not be accurate. This is due to the fact that the HDF files have data beginning at Noon UTC. Practically, this means that half a day is missing and the daily flow calculations (min, max, average, standard deviation, ...) will not be accurate. Our solution for this issue follows:

1. Ensure we're processing old-enough files that there will be some overlap with data already in the database and the files to be processed.
2. When calculating daily values for the first file, skip half a day. You can calculate the start of the second day with the following formula: first_timestamp - (first_timestamp % 86400) + 86400