Данные, полученные из REST API в хранилище данных Databrick - PullRequest
Новая
       14

Данные, полученные из REST API в хранилище данных Databrick

0 голосов
/ 09 марта 2019

Вопрос искры новичка.Я пытаюсь прочитать данные из REST API, который возвращает данные по нумерации страниц.Для извлечения данных мне придется вызывать один и тот же API 5 раз.Я хотел бы сохранить эти данные в таблице данных.Результат API находится в JSON.Идея состоит в том, чтобы каждый день совершать вызов API и сохранять данные, разбитые по дате.Все примеры приводят к одному вызову API.

Есть указатели?Спасибо

1 Ответ

0 голосов
/ 11 апреля 2019

Я делаю что-то похожее с Google Api. Наиболее хороший Apis будет обрабатывать пейджинг, возвращая токен пагинации, например вы запрашиваете первые X записей, и это дает токен, показывающий, что у него больше записей, поэтому вы продолжаете цикл, пока больше нет записей.

Что касается Spark, вы можете использовать Row, чтобы продолжать добавлять возвращаемые данные в список, а затем, наконец, преобразовывать их в 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://stackoverflow.com/questions/29903125/google-reporting-api-customer-usage-report 
# 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)
...