I am using Pandas to fetch around 2 million records from an API that returns a JSON object. The API has a limit of returning only 5000 JSON object at a time, so I iterate over the API calls to fetch the JSONs. These are the steps that I follow: 1. Get all the record_ids in a list. 2. Create API calls (URLs) by breaking the record_ids into chunks of 5000 each. 3. Iterate over the created URLs to fetch the JSONs. 4. Create a list of JSONs that were fetched above. 5. Use pd.io.json.json_normalize to create the dataframe.
The problem is that I am running out of memory if I exceed a certain limit of records to be fetched. I am trying to use DASK to help with the memory issue. However, I am unable to figure out how to use DASK bags to perform a similar function as lists (e.g. Append). Or, how do I add on more JSONs returned by the iterative API calls onto the same DASK bag?
This is the code that I am using, and it work fine for smaller datasets:
import pandas as pd
import json
import requests
import getpass
# Specify the date range and system for which the recordIDs need to be fetched
recordIDsURL = 'http://example.com:8071/records/getIds?system=ABC&daterange=2019-01-15,2019-10-15'
# Specify the record service API which returns the record info for provided record ids
recordServiceURL = 'http://example:8071/records/'
# Get the recordIds for the provided date range and system
request = requests.get(recordIDsURL, auth = requests.auth.HTTPBasicAuth(username, password))
# Put the recordIds into a list
listid = request.json()
# Divide the recordIDs into smaller lists containing 5000 recordIDs
listChunks = [listid[x:x+5000] for x in range(0, len(listid), 5000)]
# Make a list for disctinct URLs for calling the API
url = [0 for i in range(len(listChunks))]
# Make a list for storing the result of the URL calls
recordRequest = [0 for i in range(len(listChunks))]
# Make a list for converting the result of the URL calls into a list of JSONs
jsonList = [0 for i in range(len(listChunks))]
# Iterate over the URL calls
for i in range(len(listChunks)):
url[i] = recordServiceURL + (','.join(listChunks[i]))
recordRequest[i] = requests.get(url[i], auth = requests.auth.HTTPBasicAuth(username, password))
jsonList[i] = recordRequest[i].json()
# Merge the JSON list into a single JSON to load into DF
mergeJson = []
for i in jsonList:
mergeJson += i
df = pd.io.json.json_normalize(mergeJson)
In a nut sheet, I am hoping to use DASK bags and DASK dataframe in place of the python list and pandas dataframe in the above code.