如何将一个 DASK 包附加到另一个 DASK 包？

Question

我正在使用 Pandas 从返回 JSON 对象的 API 中获取大约 200 万条记录。 API 有一次只能返回 5000 个 JSON 对象的限制，因此我遍历 API 调用以获取 JSON。这些是我遵循的步骤： 1. 获取列表中的所有record_ids。 2. 通过将 record_ids 分成 5000 个块来创建 API 调用 (URL)。 3. 遍历创建的 URL 以获取 JSON。 4. 创建上面提取的 JSON 列表。 5. 使用 pd.io.json.json_normalize 创建数据框。

问题是如果我超过了要获取的记录的特定限制，我的内存就会用完。我正在尝试使用 DASK 来帮助解决内存问题。但是，我无法弄清楚如何使用 DASK 包来执行与列表类似的功能（例如附加）。或者，如何将迭代 API 调用返回的更多 JSON 添加到同一个 DASK 包中？

这是我正在使用的代码，它适用于较小的数据集：

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)

简而言之，我希望使用 DASK 包和 DASK 数据框来代替上述代码中的 python 列表和熊猫数据框。

Answer 1

您可以使用 dask.bag.concat 函数连接多个 Dask Bags