Skip to content

Ingesting Production Data

Before starting production ingestion

The data structure (also known as the data model or schema) often changes from year to year. In some cases this might require updating the DSL script. Therefore, it is important before launching the ingestion of the full new data (we refer to it as production ingestion) to test that current script can handle it.

Please refer to Test data ingestion section of the Testing Medicare Processing Workflow page on how to test data ingestion.

Ingestion process

Data directories must be organized in a specific way for ingestion by Dorieh pipelines. Consult the Medicare Pipeline Documentation for more detailed guidance on data placement. Afterward, ensure you have successfully tested smaller sample ingestions as described in the Getting Started guide.

See also General description

Organizing data directory

In the ReD environment, the data currently is hosted in:

/data/labs/dominici_nsaph_l3/import/CMS\ Data\ Request\ DUA\ 19945\ -\ RESDAC-9229/13701_copy/13701/
  -- 2015
  -- 2019
  -- 2020
  -- 2021

Selecting Data for Ingestion

  • Single Year Ingestion: To ingest data for a specific year, provide the path with that year's folder, e.g.,

    /data/labs/dominici_nsaph_l3/import/CMS\ Data\ Request\ DUA\ 19945\ -\ RESDAC-9229/13701_copy/13701/2015

    The folder must contain MBSF and MedPAR *.dat and corresponding *.fts files (see FTS files description).

  • Multi-Year Ingestion: To ingest multiple years of data, point to the parent directory:

    /data/labs/dominici_nsaph_l3/import/CMS\ Data\ Request\ DUA\ 19945\ -\ RESDAC-9229/13701_copy/13701

(Consult the section above) for the correct directory structure.

In this case, the ingestion for all 4 years (2015, 2019 - 2021) will be triggered.

Starting Ingestion Workflow

The processing of the Medicare data for 20 years takes several days. Even when adding data for a single year, the entire dataset is reprocessed. This happens because the beneficiaries dataset is aggregated across multiple years of enrollment, and therefore must be recalculated in full. Hence, you need to take several precautions before starting the process:

  1. Ensure the worker node has the capacity to run uninterrupted for at least 7 days. In particular, verify that it will not be terminated earlier because of the end of reservation window.
  2. Ensure that the process is started using nohup command and with background processing (& at the end of line).
  3. Ensure that the progress is logged into a directory available for monitoring.

We recommend running the process in the 

/shared/dorieh-logs/ingestion/

directory. For example, ingestion of 2015 data has been run in

/shared/dorieh-logs/ingestion/2015

The following command has been used to start the ingestion:

cd /shared/dorieh-logs/ingestion/2015
nohup toil-cwl-runner --retryCount 1 --cleanWorkDir never --outdir outputs --workDir work --jobStore j1 /usr/local/lib/python3.9/site-packages/dorieh/cwl/medicare.cwl --database /home/dorieh/database.ini --connection_name dorieh --input /data/labs/dominici_nsaph_l3/import/CMS\ Data\ Request\ DUA\ 19945\ -\ RESDAC-9229/13701_copy/13701/2015 2>&1 > 1_2015prod.log &
tail -f  1_2015prod.log

The same command split for readability over multiple lines (but might be more difficult to copy and paste):

cd /shared/dorieh-logs/ingestion/2015
nohup toil-cwl-runner --retryCount 1 --cleanWorkDir never --outdir outputs \
    --workDir work --jobStore j1 \
    /usr/local/lib/python3.9/site-packages/dorieh/cwl/medicare.cwl \
    --database /home/dorieh/database.ini \
    --connection_name dorieh \
    --input /data/labs/dominici_nsaph_l3/import/CMS\ Data\ Request\ DUA\ 19945\ -\ RESDAC-9229/13701_copy/13701/2015 \
     2>&1 > 1_2015prod.log &
tail -f  1_2015prod.log

Nohup

Do not forget to use nohup!

Monitoring the workflow progress

General monitoring tips

No need for worker node or Singularity shell

Progress can be monitored from any node without needing to ssh into a worker node. There is also no need to start a singularity shell.

The last command from the previous section will continuously update workflow progress:

tail -f  1_2015prod.log

To view the log interactively when logging back on the system use:

less /shared/dorieh-logs/ingestion/2015/1_2015prod.log

Successful completion is indicated by JSON entries detailing workflow step progress and error logs that are saved into outputs directory:

{
  "qc_ev_create_log": {
    "location": "file:///shared/dorieh-logs/ingestion/2015/outputs/data_loader-2025-03-05-09-48-27.log_2",
    "basename": "data_loader-2025-03-05-09-48-27.log",
    "nameroot": "data_loader-2025-03-05-09-48-27",
    "nameext": ".log",
    "class": "File",
    "checksum": "sha1$fa01481303c9030c8095387661a3bdc6851fc1ed",
    "size": 12060,
    "path": "/shared/dorieh-logs/ingestion/2015/outputs/data_loader-2025-03-05-09-48-27.log_2"
  },
  "registry": {
    "location": "file:///shared/dorieh-logs/ingestion/2015/outputs/cms.yaml",
    "basename": "cms.yaml",
    "nameroot": "cms",
    "nameext": ".yaml",
    "class": "File",
    "checksum": "sha1$73700ade239b3a0c5f755ef694f05aebb4442c68",
    "size": 140457,
    "path": "/shared/dorieh-logs/ingestion/2015/outputs/cms.yaml"
  }
}

The workflow has multiple steps and each step produces 2 log entries: a progress log and an error log (the latter should be an empty file), so there many entries like above. Corresponding files are saved into the output directory (defined as a --outdir command line argument).

Above the outputs there will be messages like:

[2025-03-05T11:59:10+0000] [MainThread] [I] [toil.leader] Finished toil run successfully.
[2025-03-05T11:59:10+0000] [MainThread] [I] [toil.cwl.cwltoil] Collecting workflow outputs...
[2025-03-05T11:59:10+0000] [MainThread] [I] [toil.cwl.cwltoil] Saving file:///shared/dorieh-logs/ingestion/2015/outputs/create_admissions.err...

...

[2025-03-05T11:59:13+0000] [MainThread] [I] [toil.cwl.cwltoil] Stored workflow outputs
[2025-03-05T11:59:13+0000] [MainThread] [I] [toil.cwl.cwltoil] Computing output file checksums...
[2025-03-05T11:59:13+0000] [MainThread] [I] [toil.cwl.cwltoil] CWL run complete!
[2025-03-05T11:59:19+0000] [MainThread] [I] [toil.common] Successfully deleted the job store: FileJobStore(/shared/dorieh-logs/ingestion/2015/j1)

You can also grep for Finished toil run successfully or CWL run complete!:

grep 'Finished toil run successfully' 1_2015prod.log
grep 'CWL run complete!' 1_2015prod.log

If the workflow has failed, the last line of the log file will indicate PermanentFail message. However, sometimes, if the failure was caused by OS killing the process or the node termination, the failure message might be absent. If you suspect that something is wrong, check for the timestamp last update message in the log file or for the timestamp of the log file itself. If it is more than several hours old, it is an indication that workflow is no longer running.

Long running steps...

Some steps — particularly those involving in-database transformations — may take considerable time and generate minimal log output.

To obtain more information you can examine log files for individual steps, especially those that are currently running. Toil produces a bit funny directory structure, so we need to use a few tricks to find the logs for currently running workflow steps.

Checking for processing errors

The following command will let you see if you have any errors:

find /shared/dorieh-logs/ingestion/2015/work/ -type f -name "*.err" -size +0c -exec ls -alF {} \;

If the output is empty, it is a good sign - no errors.

Log files on success

In a successful run, most error log files should either be empty files. If you find non-empty .err files, examine them closely.

If you have multiple runs inside your work directory, you can provide a subdirectory for the current run, e.g.:

find /shared/dorieh-logs/ingestion/2015/work//toilwf-c36b795b68935d99be01ed1556c85b1e/ -type f -name "*.err" -size +0c -exec ls -alF {} \;

Finally, to see all error logs just remove -size +0c filter:

find /shared/dorieh-logs/ingestion/2015/work/toilwf-c36b795b68935d99be01ed1556c85b1e/ -type f -name "*.err" -exec ls -alF {} \;

Monitoring individual log steps

First, search for all log files:

find /shared/dorieh-logs/ingestion/2015/work/toilwf-c36b795b68935d99be01ed1556c85b1e/ -type f -name "*.log" -exec ls -alF {} \;

Working directories have random names

Be sure to replace directory name toilwf-c36b795b68935d99be01ed1556c85b1e with the one corresponding to the current run.

Look for the most recent log files. Then examine them.

Verifying data warehouse content

Database access is only available from a singularity shell on a worker node

Checking database content can only be performed from a worker node in dorieh partition and inside a Singularity shell (Apptainer> prompt)

Inside the Singularity shell, run the following queries to ensure that the relevant tables have been updated:

python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini "SELECT obj_description(oid) AS comment FROM pg_class WHERE relname = 'mbsf_abcd_2015'"
python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini "SELECT obj_description(oid) AS comment FROM pg_class WHERE relname = 'beneficiaries'"
python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini "SELECT obj_description(oid) AS comment FROM pg_class WHERE relname = 'enrollments'"
python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini "SELECT obj_description(oid) AS comment FROM pg_class WHERE relname = 'admissions'"

The output should look like:

('CREATED BY Dorieh: {"version": "0.2.5", "url": "https://github.com/NSAPH-Data-Platform/dorieh", "commit": null}. Created at 2025-02-25 23:59:18.116167',)
('CREATED BY Dorieh: {"version": "0.2.5", "url": "https://github.com/NSAPH-Data-Platform/dorieh", "commit": null}. Created at 2025-02-28 10:25:52.848707',)
...

You can compare the timestamp with the start time of your ingestion job to confirm these updates occurred during the current run.

Note

📎 Timestamps shown here are illustrative. You should see values corresponding to your current run.

Then, see how many records we now have in the data warehouse, partitioning by year:

python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini 'SELECT count(*) from medicare.beneficiaries '
python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini 'SELECT year, count(*) from medicare.enrollments GROUP BY year order by year'
python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini 'SELECT year, count(*) from medicare.admissions GROUP BY year order by year'

You can also examine admissions audit table to verify the number of rejected records increased where expected.

python -m dorieh.platform.util.psql  --connection  dorieh --db /home/dorieh/database.ini 'SELECT year, count(*) from medicare_audit.admissions GROUP BY year order by year'