从 REST API 检索到 Databrick 数据存储的数据

Question

一个新手火花问题。我正在尝试通过分页从 REST API 读取数据，其中 returns 数据。要检索数据，我将调用相同的 API 5 次。我想将该数据保存在数据块 table 中。 API 的结果在 json 中。想法是每天进行 API 调用并保存按日期分区的数据。所有示例都会导致单个 API 调用。

有什么指点吗？谢谢

Answer 1

我正在做与 Google Api 类似的事情。大多数好的 Apis 将通过 returning 分页标记来处理分页 - 例如你要求前 X 条记录，它给出一个标记来表明它还有更多记录，所以你一直循环直到没有更多记录。

在 Spark 方面，您可以使用 Row 将 return 数据不断添加到列表中，然后最终将其转换为 Dataframe（之后您可以在数据块中执行所有常见的操作：（Python 代码使用 Google Api Python Sdk，大小缩写）但你应该明白了:)

from pyspark.sql import *

..
credentials = service_account.Credentials.from_service_account_info(SERVICE_ACCOUNT_FILE, scopes=SCOPES, subject=IMPERSONATED_USER)
service = build('admin', 'reports_v1', credentials=credentials)   #https://developers.google.com/api-client-library/python/start/get_started#build-the-service-object

#  
# https://developers.google.com/resources/api-libraries/documentation/admin/reports_v1/python/latest/admin_reports_v1.userUsageReport.html
# https://developers.google.com/api-client-library/python/start/get_started#build-the-service-object
print('Hitting Google Admin Reports - User Usage Api')
request = service.userUsageReport().get(userKey=keyUserFilter, date=keyDateFilter, filters=fieldFilter, maxResults=pageSize, parameters=fieldSelect)

rows = []
# get all pages until there are none left
while request != None:
  response = request.execute()
  rows.append(Row(id=ur.get("id"), item=response.get("item")))  
  request = service.userUsageReport().get_next(request, response)

print("end of loop")
df1 = spark.createDataFrame(rows)

display(df1)

从 REST API 检索到 Databrick 数据存储的数据

Data retrieved from REST API to Databrick data store

pyspark

databricks