pipeline for Safegraph patterns

[jingjtang]

Addresses #191

• Raw data are patterns.csv from Weekly Patterns which is updated weekly.

  • Each raw file contains weekly data, each row of which is a point of interest
  • Raw files are in .csv.gz format
  • Until 2020-06-15, each raw files is about 1GB
  • Starting from 2020-06-16, the raw files for a single week are splited into four patterns-partX.csv.gz, each part is about 300MB.

• This pipeline can be easily fit to other datasets in Safegraph with columns like "visits_by_day")

• Exceptions in geo have not been finished. Temporarily, found more than 1.5k ZipCodes that are not included in our mapping file.

• Mapping information from naics code to safegraph_brand_id is from brand_info.csv in Places Schema datasets

• Won't download the whole Places Schema datasets but only necessary brand_info.csvs which need to be stored in ./static/brand_info

Questions to be answered:

• Why only 42~43 States are available?
• Why previously assert statement failed for some patterns.csv?
  Sum of visits_by_day is not always equal to raw_visit_counts in those raw files
• Do we want to stick to multi-processing?
• How to ingest new file only?

[huisaddison]

Thanks @jingjtang ! I will try to complete the review of the Safegraph pipeline by early next week.

I do have some early comments / concerns, based on an initial skim of the codebase.

• First, it appears that the upstream changes were "merged" in via commits 2cd3152, 517437c, and dbfb844. I am not absolutely certain, but based on git blame it looks like the files in those commits were pasted into this branch, rather than introduced via a git rebase. If that is the case, I would appreciate @krivard 's opinion regarding whether those commits should be undone and a proper rebase on main performed; or if the blame history is not crucial and we can defer this to merge conflict resolution. A proper rebase would also make the code review process cleaner; because most of the "Files Changed" in this PR are actually from the geomapping utilities.
• Second, although the geomapper utilities are now available in the branch, is the pipeline using those utilities, or still using a pipeline-internal implementation? It appears, based on https://github.com/cmu-delphi/covidcast-indicators/blob/safegraph_patterns/safegraph_patterns/delphi_safegraph_patterns/run.py#L54-L59, that an internal implementation is still being used. If so, I recommend refactoring the code to first use the geomapper utilities before merging into main.

[huisaddison]

@jingjtang in order to run the pipeline for data prior to June 2020, I uncommented line 26. The pipeline runs properly for a couple dates:

[addison@bigchunk-dev-01 safegraph_patterns]$ env/bin/python -m delphi_safegraph_patterns
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2020-04-13-weekly-patterns.csv.gz
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2020-04-06-weekly-patterns.csv.gz
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2019-11-18-weekly-patterns.csv.gz
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2020-05-04-weekly-patterns.csv.gz
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2019-07-22-weekly-patterns.csv.gz
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2019-07-29-weekly-patterns.csv.gz

But then it hangs, without having processed all the data. (top shows no Python processes consuming memory or processor time). Do you know why this may be?

(Without line 26 uncommented, i.e., only running on data following June 2020, the pipeline completed and terminated correctly.)

[huisaddison]

If it helps debug, @jingjtang , I added a print(metric) inside the exporting loop:

[previous output elided]
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2019-07-22-weekly-patterns.csv.gz
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Finished pulling data from /mnt/data/safegraph//weekly-patterns/v2/main-file/2019-07-29-weekly-patterns.csv.gz
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit
Wrote wip_bars_visit
Wrote wip_restaurants_visit

It seems that there is something about the 2019-07-29 data in particular that is causing the pipeline to hang.

[huisaddison]

Really confused as to why the pipeline hangs. pool.map() starts many (>20) jobs concurrently, but activity dies down after a handful complete.

I do notice that memory consumption for individual subprocesses is much higher for these pipelines compared to the original Safegraph pipeline. The July 22nd and 29th jobs consume up to 8GB, but this memory is freed after the completion of their tasks...

[huisaddison]

Upon further investigation, there are four Python processes that appear to be sleeping (?) -- they each consume 1.5GB, but are using 0% CPU and are not bolded in top. There are another 8 Python processes that consume essentially no memory, and have 0% CPU consumption (which makes sense, because pool was given 12 cores.)

I'm guessing these Python processes are blocked, and the fact that no progress is being made on these processes in turn blocks pool.map(). But I honestly have no idea what is blocking these processes :/ If I had to bet money, my guess is that it's something inside construct_signals().

@krivard do you have any suggestions for quick ways to debug these kinds of issues?

top - 14:40:50 up 83 days, 19:17,  2 users,  load average: 0.02, 1.59, 3.87                                                                                                            [0/112]
Tasks: 227 total,   1 running, 226 sleeping,   0 stopped,   0 zombie
%Cpu(s):  0.0 us,  0.0 sy,  0.0 ni, 99.9 id,  0.0 wa,  0.0 hi,  0.0 si,  0.0 st
KiB Mem : 66318348 total, 54923956 free,  7355480 used,  4038912 buff/cache
KiB Swap:  4194300 total,  2792728 free,  1401572 used. 58467448 avail Mem 

   PID USER      PR  NI    VIRT    RES    SHR S  %CPU %MEM     TIME+ COMMAND                                                                                                                  
 85142 addison   20   0 2578056   1.5g   6028 S   0.0  2.4   6:42.88 python                                                                                                                   
 85134 addison   20   0 2582068   1.5g   6028 S   0.0  2.4   6:45.12 python                                                                                                                   
 85185 addison   20   0 2450876   1.4g   5992 S   0.0  2.3   6:37.60 python                                                                                                                   
 85176 addison   20   0 2479640   1.4g   2960 S   0.0  2.1   3:50.95 python                                                                                                                   
 85347 addison   20   0 1042700   6508   1044 S   0.0  0.0   0:00.00 python                                                                                                                   
   553 root      20   0   47668   6404   6204 S   0.0  0.0   6:40.00 systemd-journal                                                                                                          
     1 root      20   0  196044   6292   2328 S   0.0  0.0   8:02.13 systemd                                                                                                                  
 84922 addison   20   0 1042700   5676    880 S   0.3  0.0   0:17.41 python                                                                                                                   
 85267 addison   20   0 1042700   5396      0 S   0.0  0.0   0:00.00 python                                                                                                                   
 85249 addison   20   0 1042700   5240      0 S   0.0  0.0   0:00.00 python                                                                                                                   
 85232 addison   20   0 1042700   5236      0 S   0.0  0.0   0:00.00 python                                                                                                                   
 85221 addison   20   0 1042700   5220      0 S   0.0  0.0   0:00.00 python                                                                                                                   
 85211 addison   20   0 1042700   4824      0 S   0.0  0.0   0:00.00 python                                                                                                                   
 85202 addison   20   0 1042700   4588      0 S   0.0  0.0   0:00.00 python                                                                                                                   
 34728 addison   20   0   28592   3864    768 S   0.0  0.0  25:43.92 tmux                                                                                                                     
 85193 addison   20   0 1042700   3416      0 S   0.0  0.0   0:00.00 python  

[capnrefsmmat]

Can you switch from pool.map to ordinary map, so you can attach pdb to inspect activity when needed?

You could then use ipython -i run.py (or whatever script you want to run), and if it hangs, Ctrl-C will drop you into the IPython console. Then %debug brings up the postmortem debugger so you can see where you are.

[huisaddison]

@capnrefsmmat Thanks for those suggestions! I will use those techniques in the future - before I saw your post, I did a more rudimentary switch to direct iteration and waited for the traceback.

@jingjtang I have found the source of the error. It lies in the assert statement here: https://github.com/cmu-delphi/covidcast-indicators/pull/225/files#diff-fbb4d07961dfbc089de3ac81616f7041R70-R71 This condition is not met, at least for the file /weekly-patterns/v2/main-file/2019-08-19-weekly-patterns.csv.gz. In my for loop modification, the program quits when the assert statement fails. I suppose pool.map() is not sure what to do / the subprocess doesn't return properly, so it freezes it up in the parallel version.

@jingjtang Can you please investigate whether this assert statement should still hold for the final pipeline, and if so why it is not being met by some of the input files?

[jingjtang]

@capnrefsmmat Thanks for those suggestions! I will use those techniques in the future - before I saw your post, I did a more rudimentary switch to direct iteration and waited for the traceback.

@jingjtang I have found the source of the error. It lies in the assert statement here: https://github.com/cmu-delphi/covidcast-indicators/pull/225/files#diff-fbb4d07961dfbc089de3ac81616f7041R70-R71 This condition is not met, at least for the file /weekly-patterns/v2/main-file/2019-08-19-weekly-patterns.csv.gz. In my for loop modification, the program quits when the assert statement fails. I suppose pool.map() is not sure what to do / the subprocess doesn't return properly, so it freezes it up in the parallel version.

@jingjtang Can you please investigate whether this assert statement should still hold for the final pipeline, and if so why it is not being met by some of the input files?

The assert statement should hold. I will look into those files. Thanks @huisaddison

[jingjtang]

@huisaddison I checked 2019-07-22-weekly-patterns.csv.gz, 2019-07-29-weekly-patterns.csv.gz and 2019-08-19-weekly-patterns.csv.gz. It seems there is no problem with either 2019-07-22-weekly-patterns.csv.gz or 2019-07-29-weekly-patterns.csv.gz but does exist problem in the assert statement with 2019-08-19-weekly-patterns.csv.gz.

By definition, the sum of visits_by_day should be equal to raw_visit_counts. But there does exist a row (a specific place) that does not follow this definition. (A problem in the raw data).

So, now I think we can delete this assert statement.

[huisaddison]

OK, great, thanks @jingjtang ! And your investigation checks out -- 7/22 and 7/29 worked properly, they always happened to be the last two that succeeded before the rest failed, but 8/19 indeed failed.

I will remove the assert statement from my local copy and re-run.

[krivard]

We're going to want to rebase this branch and force-push it to resolve the conflicts before merging to main.

The procedure for this is:

$ git checkout safegraph_patterns
$ git rebase -i main
[clean up commit history using the editor; save and quit]
[git will then attempt to play back the revised commit history on top of the current state of main. it will encounter conflicts; you can then choose to:
* resolve them, or 
* abort, and try again with a different way of cleaning up the commit history (possibly by excluding 2cd3152, 517437c, and dbfb844)]
[once playback succeeds, you're done!]

At this point it's then important to check with everyone who plausibly checked out a copy of the safegraph_patterns branch and have them delete their local copy using git branch -D safegraph_patterns. Once you have the only checked-out copy, you can then

$ git push --force origin safegraph_patterns

[huisaddison]

The pipeline ran successfully in "serial mode", so the freezing in parallel mode is a more subtle bug.

I am going to deliver an initial version of this data to the causal modeling team, to allow them to assess its value, and in the meanwhile I will switch over to assessing the new GHT data.

The eventual next steps for this PR are:

• (1) Rebase against main properly
• (2) Debug the parallel version deadlock
• (3) Implement a system for only processing "new" files.

Both (2) and (3) are good ideas, but implementing either one of them would result in a pipeline that runs in a reasonable amount of time in the steady state (i.e., if we are resource-constrained, only one of them is necessary).

[jingjtang]

Added code (commented) for using geomap utils.

• the total population in zip_to_pop.csv seems not up-to-date. Much less than 328 million.
• zip_to_state_id returns upper case instead of lower case

[huisaddison]

Hi @jingjtang - a quick question before you take a vacation! I noticed that the state-level bars data has at most 42 states - a quick line count (wc -l *state*bars*) shows that most files have between 40 and 42 states. Is this a known issue? And is this due to the input data, or is it a bug in the pipeline?

[jingjtang]

@huisaddison. Yes, I also noticed that. It should be a problem due to the input data. (No matter I use the mapping file in ./static or GeoMapper utility)

[huisaddison]

Thanks @jingjtang . My hypotheses are:

• those states have policies against reporting
• in those states, bar / alcohol transactions fall under a different NAISC (?) code

I'll encourage the modeling team to start with the restaurant data and we can look into this further after your time off.

Regarding the bugs in the geomapper utils (missing population, uppercase states) - can you comment in the open PR (if it's under review), or open an issue / PR for these problems? (Unless @krivard suggests an alternative course of action).

[huisaddison]

I did some more sleuthing regarding the missing states. On 2020-08-16, the missing states were:

{'de', 'ak', 'nh', 'me', 'nd', 'sd', 'wv', 'vt', 'wa', 'wy', 'dc'}

The commonality between these states are that they all have small populations (perhaps with the exception of Washington State).

Then, I took a look at the raw visit numbers for bars versus for restaurants. The number of visits to restaurants are already two orders of magnitude larger than the number of visits to bars, so it seems like for those 8-9 states that are missing, Safegraph just doesn't have enough penetration to "see" the few bar visits that may be occurring there.

If we want to improve the spatial coverage of the bars indicator, we may want to take the union over multiple NAISC codes, but that may reduce the precision of the indicator.

[addison@bigchunk-dev-01 receiving]$ cat 20200816_state_wip_bars_visit_num.csv 
geo_id,val,se,sample_size
al,117,NA,NA
ar,49,NA,NA
az,100,NA,NA
ca,56,NA,NA
co,37,NA,NA
ct,44,NA,NA
fl,1172,NA,NA
ga,573,NA,NA
hi,1,NA,NA
ia,27,NA,NA
id,20,NA,NA
il,262,NA,NA
in,457,NA,NA
ks,50,NA,NA
ky,107,NA,NA
la,3,NA,NA
ma,40,NA,NA
md,100,NA,NA
mi,223,NA,NA
mn,130,NA,NA
mo,77,NA,NA
ms,14,NA,NA
mt,16,NA,NA
nc,105,NA,NA
ne,32,NA,NA
nj,62,NA,NA
nm,4,NA,NA
nv,55,NA,NA
ny,209,NA,NA
oh,343,NA,NA
ok,46,NA,NA
or,11,NA,NA
pa,277,NA,NA
ri,9,NA,NA
sc,84,NA,NA
tn,155,NA,NA
tx,718,NA,NA
ut,3,NA,NA
va,175,NA,NA
wi,61,NA,NA

[addison@bigchunk-dev-01 receiving]$ cat 20200816_state_wip_restaurants_visit_num.csv                                                                                                         
geo_id,val,se,sample_size
ak,173,NA,NA
al,32137,NA,NA
ar,12455,NA,NA
az,16428,NA,NA
ca,35986,NA,NA
co,11810,NA,NA
ct,2282,NA,NA
dc,202,NA,NA
de,1883,NA,NA
fl,59712,NA,NA
ga,46685,NA,NA
hi,818,NA,NA
ia,10564,NA,NA
id,2881,NA,NA
il,25639,NA,NA
in,20676,NA,NA
ks,8500,NA,NA
ky,16834,NA,NA
la,14502,NA,NA
ma,4308,NA,NA
md,7899,NA,NA
me,1167,NA,NA
mi,17739,NA,NA
mn,9499,NA,NA
mo,21841,NA,NA
ms,11159,NA,NA
mt,1218,NA,NA
nc,39722,NA,NA
nd,1850,NA,NA
ne,5902,NA,NA
nh,1364,NA,NA
nj,6260,NA,NA
nm,1946,NA,NA
nv,6188,NA,NA
ny,13033,NA,NA
oh,36406,NA,NA
ok,18622,NA,NA
or,4295,NA,NA
pa,15654,NA,NA
ri,826,NA,NA
sc,22652,NA,NA
sd,2984,NA,NA
tn,30333,NA,NA
tx,110730,NA,NA
ut,4037,NA,NA
va,17853,NA,NA
vt,240,NA,NA
wa,5746,NA,NA
wi,12148,NA,NA
wv,4630,NA,NA
wy,1054,NA,NA

[huisaddison]

Will this file basically be exported into the public documentation? If so, we should move this into an internal-only part of the codebase.

[krivard]

This file is not directly included in the public documentation, but

1. it is used to help the person who ultimately drafts the public documentation, and
2. we do intend on open-sourcing this codebase some time this fall

Personal tokens required for API access are encrypted before being committed to the repository, and the keys live on the delphi server, not in git.

[huisaddison]

@jingjtang The issues with the geomap utils that you mentioned, do those only affect Quidel or do they also affect this Safegraph Patterns indicator? (Or worse, do they also affect indicators that are in production?)

[jingjtang]

@huisaddison I think it would affect the safegraph patterns since the data is also provided at zipcode level. The unit test failed for a simple data frame which have integer values at zipcode level but float numbers after using the functions in geo utils to aggregate it from zipcode level to msa. As for other signals, if the aggregation starts from the zipcode level, so yes. If it starts from county level, not sure yet.

[jingjtang]

@huisaddison Talked to Dmitry already. It seems this is not a problem in the geo utils. But just a difference existing between the old mapping and the new one. So only Quidel will be affected. Safegraph patterns will not.

[huisaddison]

Hi @jingjtang - I just took a look. Can you please complete the switch to the GeoMapper utility, test the package, make necessary changes, and re-push?

I do not believe that two of methods being used in the (currently commented out) code are available in the gmpr = GeoMapper(). Specifically, gmpr.add_population_column() and gmpr.replace_geocode():

https://github.com/cmu-delphi/covidcast-indicators/pull/225/files#diff-7d2c72619d85a263632e11587b1b1401488e143b629928cb7ff56f9dd1eecb58R118-R121

They are not showing up as attributes when I play with gmpr in my console, nor do they show up in the geomap.py source in either the main or safegraph_patterns branches.

Once you amend and test the code to work with the GeoMapper utility, ping me on Slack and I will take a second look ASAP. Thank you!

[huisaddison]

Please uncomment this block, test (I think the methods in gmpr used here are no longer available), and repush. Ping me on Slack once this is ready and I will take another look ASAP :)

[jingjtang]

I merged the new geo utils from main to this branch and it should work now. The unit tests have been updated

[huisaddison]

Thanks for updating the code @jingjtang ! It looks like there are a few more kinks to iron out. It appears that the code is trying to set an obsolete keyword argument:

(env) [addison@bigchunk-dev-01 safegraph_patterns]$ env/bin/python -m delphi_safegraph_patterns
usage: aws [options] <command> <subcommand> [<subcommand> ...] [parameters]
To see help text, you can run:

  aws help
  aws <command> help
  aws <command> <subcommand> help
aws: error: argument --endpoint-url: expected one argument
multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/usr/lib64/python3.6/multiprocessing/pool.py", line 119, in worker
    result = (True, func(*args, **kwds))
  File "/usr/lib64/python3.6/multiprocessing/pool.py", line 44, in mapstar
    return list(map(*args))
TypeError: process() got an unexpected keyword argument 'map_df'
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/lib64/python3.6/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/usr/lib64/python3.6/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/addison/safegraph_patterns/covidcast-indicators/safegraph_patterns/delphi_safegraph_patterns/__main__.py", line 11, in <module>
    run_module()  # pragma: no cover
  File "/home/addison/safegraph_patterns/covidcast-indicators/safegraph_patterns/delphi_safegraph_patterns/run.py", line 88, in run_module
    pool.map(process_file, files)
  File "/usr/lib64/python3.6/multiprocessing/pool.py", line 266, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/usr/lib64/python3.6/multiprocessing/pool.py", line 644, in get
    raise self._value
TypeError: process() got an unexpected keyword argument 'map_df'
(env) [addison@bigchunk-dev-01 safegraph_patterns]$ 

Can you fix this and I will re-review? If possible, please include the output of the included unit tests in this PR (hopefully the unit tests do not require a bigchunk instance to run, but if they do, I can look into creating a login into my server for you, or we can talk to Brian about getting you a bigchunk.)

Thanks!

[huisaddison]

Obsolete keyword argument being set here.

[huisaddison]

Also - @jingjtang if you have many other Delphi responsibilities to tend to, let me know and I'll work on this PR tonight (to take this off your plate).

[jingjtang]

@huisaddison It was fixed. Can you try it again? The unit tests are updated

[huisaddison]

Tests pass, linter is mostly happy, pipeline runs - LGTM :)

[huisaddison]

Found typo in documentation.

[huisaddison]

Suggested change

	* `num`: number of new deaths on a given week
	* `num`: number of visits on a given week

[jingjtang]

Thanks.

[Akvannortwick]

Should it be "in a given week"?

The way I read it is:
If "on" it would refer to a given week entirely.
If "in" it would refer to days within a given week.

[jingjtang]

It should be "in". Thanks.