Загрузка множества файлов JSON с вложенными структурами данных для формирования графика широты и долготы с использованием Dask
Я наткнулся на видео это, показывающее, как вы можете загружать множество файлов JSON и извлекать данные в фрейм данных Dask. Это выглядело идеально для того, что я давно хотел сделать. У меня много файлов JSON в подкаталогах (по одному на каждый день). Каждый файл JSON содержит данные о поездке для транспортного средства. На самом деле подпоездки, состоящие из нескольких начальных и конечных позиций. Файл может иметь несколько SubTrips в массиве JSON. В StartPosition может быть несколько SubTrips. StartPosition является необязательным в SubTrips. StartPosition содержит Latitude и Longitude. Данные (все строки в файле) выглядят примерно так (см. также пример файла это):
[{"SubTrips":[{"StartPosition":{"IsAvl":false,"Source":"Gps","OdometerKilometres":147918,"DistanceSinceReadingKilometres":0,"AgeOfReadingSeconds":9528,"Pdop":0,"Vdop":0,"Hdop":1,"NumberOfSatellites":10,"Heading":0,"AltitudeMetres":38,"SpeedKilometresPerHour":0,"Longitude":18.4726696014404,"Latitude":-33.8071098327637,"Timestamp":"2019-01-01T06:31:51Z","DriverId":-9119609258885412605,"AssetId":-5651694037363791566,"PositionId":816711615055360000},"EndPosition":{
Мне удалось прочитать файлы JSON, используя следующее:
import dask.bag as db
lines = db.read_text("C:/Trips/2019*/*.json")
records = lines.map(json.loads)
Однако у меня возникли проблемы с извлечением вложенной структуры данных JSON в фрейм данных Dask, где каждая строка является записью StartPosition из файлов JSON. Видео на YouTube показывает только один уровень JSON, тогда как у меня есть несколько уровней и несколько массивов данных. Несколько уровней кажутся выполнимыми, но я терплю неудачу с массивами. Существует также проблема с необязательностью StartPosition, которая была решена ниже с помощью filter. Я могу сделать что-то вроде этого; но для этого требуется только первая поездка файла в массивах, а также первая начальная позиция поездки, состоящей из нескольких начальных позиций:
records.filter(lambda d: "StartPosition" in d[0]["SubTrips"][0]).map(lambda d: d[0]["SubTrips"][0]["StartPosition"]["NumberOfSatellites"]).frequencies(sort=True).compute()
Возможно, с помощью этого можно отсортировать первый массив, но я не уверен:
subtrips = records.flatten()
Как я могу извлечь все данные записи StartPosition в фрейм данных Dask? В основном строка с атрибутами StartPosition, включая в основном широту и долготу.
В конце я хотел бы нанести местоположения Dask с помощью datashader на карту mapbox.
Ответы 1
Похоже, вы очень близки к результату. Помимо видео еще один полезный ресурс — этот учебник.
Чтобы обрабатывать пользовательскую логику (например, необязательность StartPosition), лучше всего определить пользовательские функции:
from json import loads
from dask.bag import read_text
def process(record):
# implement custom logic here
for entry in record.get("SubTrips"):
if data_of_interest := entry.get("StartPosition"):
yield {
k: v
for k, v in data_of_interest.items()
if k in ["DriverId", "Longitude", "Latitude"]
}
def process_json(json_data):
# this will produce a list of records, so make sure to flatten
return [result for record in json_data for result in process(record)]
bag = (
read_text("7044247166111462221*.json")
.map(loads)
.map(process_json)
.flatten()
.to_dataframe()
)
Применение json.loads к текстовым/строковым данным дает словарь, ключевой функцией является process, которая определяет логику преобразования конкретной записи, а process_json является оболочкой для обработки общего случая (предполагая несколько записей в файле, но если есть только одна запись в каждом файле). файл этой обертки не нужен). Прежде чем преобразовать данные в dataframe, нам нужно будет flatten его, чтобы каждый элемент пакета был dict.
Спасибо! Проверил, вроде работает отлично!