Анализ больших объемов данных JSON с несколькими различными типами объектов ⇐ Python

[Anonymous]

Я видел несколько разных вопросов о том, как анализировать большие файлы json, но все из них, с которыми я сталкивался, содержали единообразные данные во всем файле json. Каждый пример с неоднородными объектами json, который я видел, обычно включает в себя цикл for, который в моем случае очень медленный.
Моя проблема в том, что у меня есть около 5-10 ГБ данных, разделенных на набор json-файлов размером около 200–2000 КБ каждый, в которых вкраплено несколько разных json-объектов. Похоже, что большинство решений, которые я видел, предполагают, что все объекты будут одного типа. У меня есть определения для всех типов объектов, но мне нужен быстрый и эффективный способ прочитать их все и привести в удобный формат.
Сейчас у меня есть сценарий для чтения. через файл json пообъектно и проанализировать его таким образом, но в настоящее время обработка всех данных занимает несколько часов, и это то, что мне приходится делать довольно часто, поэтому я надеюсь, что смогу как-то сократить это время . Конечная цель — организовать все эти данные в несколько массивов, которые затем можно будет добавить в существующую базу данных SQL на основе временных меток.
У меня около 15 типов объектов. Каждый тип объекта будет содержать некоторую информацию о том, откуда поступили данные, а затем и сами данные, хотя некоторые объекты включают данные из нескольких источников вложенным образом. Ниже я привел пример самых простых и самых сложных объектов, чтобы дать представление.
Самый сложный: обычно он отправляет информацию примерно о 50 устройствам, но я сократил их до двух. для краткости. Таким образом, для большинства этих сообщений списки устройств и устройств содержат около 50 элементов.

Код: Выделить всё

{
"type": "location",
"source": "gps",
"broadcastStatus": 1,
"timestamp": xxxxxxxxxx,
"deviceStatus": {
"id1": {
"msgCount": 1,
"latestCheckin": "ACK",
"timestamp": xxxxxxxxxx
},
"id2": {
"msgCount": 3,
"latestCheckin": "ACK",
"timestamp": xxxxxxxxxx
}
},
"devices": [
{
"deviceId": "id1",
"position": [
xx.xxxxx,
-xx.xxxxx
],
"attitude": [
x.xxx,
x.xxx,
x.xxx
],
"velocity": [
x.xxx,
x.xxx,
x.xxx
],
"spd": xxx.xxxx,
"percentComplete": x.xxx
},
{
"deviceId": "id2",
"position": [
xx.xxxxx,
-xx.xxxxx
],
"attitude": [
x.xxx,
x.xxx,
x.xxx
],
"velocity": [
x.xxx,
x.xxx,
x.xxx
],
"spd": xxx.xxxx,
"percentComplete": x.xxx
}
]
}

Самое простое: все они одного типа, но существует около 40 каналов, которые передают подобные данные.

Код: Выделить всё

{
"type": "timebased"
"source": "daq",
"deviceId": "id1"
"timestamp": xxxxxxxxxx,
"channelName": "Channel1",
"value": xxx,
"addData": {
"junk1": xxxx,
"junk2": xxxx,
"junk3": xxxx
}
}

Ниже приведен псевдокод, который показывает, как я сейчас все просматриваю и анализирую.

Код: Выделить всё

basePath = r'path\to\folder'
fileList = glob(basePath+'*.json')

timebased = {}  # this is a nested dictionary full of lists for the different devices and data channels
location = {}  # same as above only for the location data

for fileName in fileList:
with open(fileName) as f:
try:
objJson = json.load(f)
except:
objJson = []
print('*******  Failed to load json data from '+fileName+'  **********')
for msg in objJson:
if msg["type"] == "timebased" and msg["source"] == "daq":
## Do some things to process this type of message
timebased[msg["deviceId"]][msg["channelName"]]["Time"].append(msg["timestamp"])
timebased[msg["deviceId"]][msg["channelName"]]["Data"].append(msg["value"])
elif msg["type"] == "location" and msg["source"] == "gps":
## Do some things to process this type of message similar to the above case
..... that keeps going for around 15 different cases
## Then this all gets pushed to a csv for now but ideally will be a DB in the future

Я знаю, что это не все идеально, но мне бы очень хотелось получить помощь в том, как лучше всего обрабатывать эти сообщения и более эффективно организовывать все данные. Я знаю, что добавление в списки - это не идеально, но с этим мне также нужна помощь, поскольку я не знаю, как еще это сделать, поскольку я не буду знать размер заранее. Как я уже сказал, конечной целью было бы иметь кучу списков или массивов со значениями времени и данных, которые я затем мог бы вставить в базу данных типа SQL, предназначенную для хранения подобных данных. Эта база данных типа SQL имеет свой собственный API, который мне придется использовать, который потребует передачи одного списка/массива значений времени и другого равной длины значений данных для каждого канала.
Мой Следующей попыткой должно было стать переключение на asyncio, поскольку я анализирую десятки тысяч файлов и этот ввод-вывод может быть моей узкой точкой, но мне также хотелось получить некоторую информацию о фактическом анализе сообщений.

Подробнее здесь: https://stackoverflow.com/questions/789 ... ject-types