JSON в науке о данных: Анализ и обработка

# JSON в науке о данных: Анализ и обработка

JSON стал одним из ключевых форматов для хранения и обмена данными в науке о данных. В этом руководстве мы рассмотрим, как эффективно работать с JSON в задачах анализа данных, машинного обучения и визуализации.

Почему JSON популярен в Data Science?

Преимущества JSON:

• Гибкая структура - поддержка вложенных данных
• Читаемость - легко понять человеку
• Универсальность - поддержка всеми языками
• API интеграция - стандарт для веб-API
• NoSQL базы - нативная поддержка в MongoDB, CouchDB

Недостатки:

• Больший размер по сравнению с бинарными форматами
• Медленнее парсинг для очень больших файлов
• Отсутствие стандартных типов для дат и времени

Работа с JSON в Pandas

Загрузка JSON в DataFrame

import pandas as pd
import json

# Простой JSON
json_data = '''
[
  {"имя": "Иван", "возраст": 30, "город": "Москва", "зарплата": 150000},
  {"имя": "Мария", "возраст": 25, "город": "СПб", "зарплата": 120000},
  {"имя": "Петр", "возраст": 35, "город": "Казань", "зарплата": 100000}
]
'''

# Метод 1: read_json
df = pd.read_json(json_data)
print(df)
#      имя  возраст    город  зарплата
# 0   Иван       30   Москва    150000
# 1  Мария       25      СПб    120000
# 2   Петр       35   Казань    100000

# Метод 2: из файла
df = pd.read_json('data.json')

# Метод 3: из URL
df = pd.read_json('https://api.example.com/data.json')

Вложенный JSON

# Вложенная структура
nested_json = '''
{
  "сотрудники": [
    {
      "имя": "Иван",
      "возраст": 30,
      "адрес": {
        "город": "Москва",
        "улица": "Тверская"
      },
      "навыки": ["Python", "SQL", "Machine Learning"]
    },
    {
      "имя": "Мария",
      "возраст": 25,
      "адрес": {
        "город": "СПб",
        "улица": "Невский"
      },
      "навыки": ["R", "Statistics", "Visualization"]
    }
  ]
}
'''

# Загрузка
data = json.loads(nested_json)

# Нормализация вложенной структуры
df = pd.json_normalize(data['сотрудники'])
print(df)
#      имя  возраст адрес.город адрес.улица  \
# 0   Иван       30      Москва    Тверская   
# 1  Мария       25         СПб     Невский   
# 
#                              навыки  
# 0    [Python, SQL, Machine Learning]  
# 1  [R, Statistics, Visualization]  

# Извлечение вложенных списков
df_expanded = df.explode('навыки')
print(df_expanded)

Экспорт в JSON

# Создание DataFrame
df = pd.DataFrame({
    'имя': ['Иван', 'Мария', 'Петр'],
    'возраст': [30, 25, 35],
    'город': ['Москва', 'СПб', 'Казань']
})

# Экспорт в JSON
# Ориентация: records (список объектов)
json_str = df.to_json(orient='records', force_ascii=False, indent=2)
print(json_str)
# [
#   {
#     "имя": "Иван",
#     "возраст": 30,
#     "город": "Москва"
#   },
#   ...
# ]

# Ориентация: columns (словарь колонок)
json_str = df.to_json(orient='columns', force_ascii=False)

# Ориентация: index (словарь с индексами)
json_str = df.to_json(orient='index', force_ascii=False)

# Сохранение в файл
df.to_json('output.json', orient='records', force_ascii=False, indent=2)

Обработка больших JSON файлов

Chunked чтение

import json

# Чтение больших файлов по частям
def процесс_большого_json(filename, chunk_size=1000):
    """
    Обработка большого JSON файла построчно
    """
    результаты = []

    with open(filename, 'r', encoding='utf-8') as f:
        # Предполагаем JSON Lines формат
        chunk = []
        for i, line in enumerate(f):
            try:
                data = json.loads(line)
                chunk.append(data)

                if len(chunk) >= chunk_size:
                    # Обработать chunk
                    df = pd.DataFrame(chunk)
                    # Ваша логика обработки
                    результаты.append(df.describe())
                    chunk = []
            except json.JSONDecodeError as e:
                print(f"Ошибка в строке {i}: {e}")

        # Обработать оставшиеся данные
        if chunk:
            df = pd.DataFrame(chunk)
            результаты.append(df.describe())

    return результаты

# Использование
stats = процесс_большого_json('large_data.jsonl')

Streaming парсинг с ijson

import ijson

def стриминг_обработка(filename):
    """
    Streaming парсинг для очень больших файлов
    """
    записи = []

    with open(filename, 'rb') as f:
        # Парсинг массива объектов
        parser = ijson.items(f, 'item')

        for i, obj in enumerate(parser):
            записи.append(obj)

            # Обработка батчами
            if i % 10000 == 0 and i > 0:
                df = pd.DataFrame(записи)
                # Обработка батча
                print(f"Обработано {i} записей")
                записи = []

    # Последний батч
    if записи:
        df = pd.DataFrame(записи)
        # Финальная обработка

# Использование
стриминг_обработка('очень_большой_файл.json')

Работа с JSON API

Получение данных из REST API

import requests
import pandas as pd

def получить_данные_api(url, параметры=None):
    """
    Получение и обработка JSON из API
    """
    try:
        ответ = requests.get(url, params=параметры)
        ответ.raise_for_status()

        данные = ответ.json()

        # Преобразование в DataFrame
        if isinstance(данные, list):
            df = pd.DataFrame(данные)
        elif isinstance(данные, dict):
            # Если данные вложены
            if 'results' in данные:
                df = pd.DataFrame(данные['results'])
            elif 'data' in данные:
                df = pd.DataFrame(данные['data'])
            else:
                df = pd.json_normalize(данные)

        return df

    except requests.RequestException as e:
        print(f"Ошибка запроса: {e}")
        return None

# Пример использования
df = получить_данные_api('https://api.example.com/users')

# С пагинацией
def получить_все_страницы(base_url, лимит_на_странице=100):
    """
    Получение всех страниц из API
    """
    все_данные = []
    страница = 1

    while True:
        print(f"Загрузка страницы {страница}...")

        params = {
            'page': страница,
            'limit': лимит_на_странице
        }

        ответ = requests.get(base_url, params=params)
        данные = ответ.json()

        if not данные or 'results' not in данные:
            break

        все_данные.extend(данные['results'])

        if len(данные['results']) < лимит_на_странице:
            break

        страница += 1

    return pd.DataFrame(все_данные)

# Использование
df = получить_все_страницы('https://api.example.com/data')

JSON в Machine Learning

Подготовка данных для ML

import pandas as pd
from sklearn.preprocessing import LabelEncoder, StandardScaler
from sklearn.model_selection import train_test_split

# Загрузка JSON данных
json_data = '''
[
  {"возраст": 25, "зарплата": 50000, "город": "Москва", "купил": true},
  {"возраст": 35, "зарплата": 75000, "город": "СПб", "купил": false},
  {"возраст": 45, "зарплата": 100000, "город": "Москва", "купил": true},
  {"возраст": 30, "зарплата": 60000, "город": "Казань", "купил": false}
]
'''

df = pd.read_json(json_data)

# Кодирование категориальных признаков
le = LabelEncoder()
df['город_код'] = le.fit_transform(df['город'])

# Подготовка признаков и целевой переменной
X = df[['возраст', 'зарплата', 'город_код']]
y = df['купил'].astype(int)

# Нормализация
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# Разделение на train/test
X_train, X_test, y_train, y_test = train_test_split(
    X_scaled, y, test_size=0.2, random_state=42
)

print(f"Train размер: {X_train.shape}")
print(f"Test размер: {X_test.shape}")

Сохранение модели с метаданными в JSON

import json
import pickle
from datetime import datetime

def сохранить_модель_с_метаданными(модель, метаданные, путь_модели, путь_мета):
    """
    Сохранение модели и её метаданных
    """
    # Сохранить модель
    with open(путь_модели, 'wb') as f:
        pickle.dump(модель, f)

    # Расширить метаданные
    полные_метаданные = {
        метаданные,

        'дата_создания': datetime.now().isoformat(),
        'версия_модели': '1.0.0',
        'путь_модели': путь_модели
    }

    # Сохранить метаданные в JSON
    with open(путь_мета, 'w', encoding='utf-8') as f:
        json.dump(полные_метаданные, f, ensure_ascii=False, indent=2)

# Пример использования
метаданные = {
    'тип_модели': 'RandomForest',
    'признаки': ['возраст', 'зарплата', 'город'],
    'точность': 0.85,
    'параметры': {
        'n_estimators': 100,
        'max_depth': 10
    },
    'классы': ['не купил', 'купил']
}

сохранить_модель_с_метаданными(
    модель=trained_model,
    метаданные=метаданные,
    путь_модели='model.pkl',
    путь_мета='model_meta.json'
)

import json

# Файл конфигурации эксперимента
config = {
    "эксперимент": {
        "название": "Классификация клиентов",
        "версия": "1.0",
        "автор": "Data Science Team"
    },
    "данные": {
        "источник": "data/customers.json",
        "признаки": ["возраст", "доход", "образование"],
        "целевая": "купил_продукт"
    },
    "предобработка": {
        "масштабирование": "StandardScaler",
        "кодирование": "LabelEncoder",
        "обработка_пропусков": "mean"
    },
    "модель": {
        "тип": "RandomForestClassifier",
        "параметры": {
            "n_estimators": 100,
            "max_depth": 10,
            "random_state": 42
        }
    },
    "валидация": {
        "метод": "cross_validation",
        "фолды": 5,
        "метрики": ["accuracy", "precision", "recall", "f1"]
    }
}

# Сохранение конфигурации
with open('experiment_config.json', 'w', encoding='utf-8') as f:
    json.dump(config, f, ensure_ascii=False, indent=2)

# Использование конфигурации
def запустить_эксперимент(config_path):
    with open(config_path, 'r', encoding='utf-8') as f:
        cfg = json.load(f)

    # Загрузка данных
    df = pd.read_json(cfg['данные']['источник'])

    # Применить конфигурацию
    # ... ваш ML pipeline ...

    return результаты

результаты = запустить_эксперимент('experiment_config.json')

import pandas as pd
import json
from datetime import datetime

# JSON с временными данными
timeseries_json = '''
[
  {"timestamp": "2026-01-01T00:00:00Z", "значение": 100, "сенсор": "temp_1"},
  {"timestamp": "2026-01-01T01:00:00Z", "значение": 102, "сенсор": "temp_1"},
  {"timestamp": "2026-01-01T02:00:00Z", "значение": 98, "сенсор": "temp_1"}
]
'''

# Загрузка и парсинг дат
df = pd.read_json(timeseries_json)
df['timestamp'] = pd.to_datetime(df['timestamp'])
df.set_index('timestamp', inplace=True)

print(df)
#                       значение сенсор
# timestamp                          
# 2026-01-01 00:00:00      100  temp_1
# 2026-01-01 01:00:00      102  temp_1
# 2026-01-01 02:00:00       98  temp_1

# Resample временного ряда
df_hourly = df.resample('H').mean()

# Экспорт обратно в JSON
result_json = df.reset_index().to_json(
    orient='records',
    date_format='iso',
    force_ascii=False
)

import json
import pandas as pd
import geopandas as gpd
from shapely.geometry import Point

# GeoJSON данные
geojson_data = '''
{
  "type": "FeatureCollection",
  "features": [
    {
      "type": "Feature",
      "properties": {
        "название": "Москва",
        "население": 12500000
      },
      "geometry": {
        "type": "Point",
        "coordinates": [37.6173, 55.7558]
      }
    },
    {
      "type": "Feature",
      "properties": {
        "название": "Санкт-Петербург",
        "население": 5400000
      },
      "geometry": {
        "type": "Point",
        "coordinates": [30.3609, 59.9311]
      }
    }
  ]
}
'''

# Загрузка в GeoDataFrame
gdf = gpd.read_file(json.loads(geojson_data))

print(gdf)
#           название  население                     geometry
# 0           Москва   12500000  POINT (37.61730 55.75580)
# 1  Санкт-Петербург    5400000  POINT (30.36090 59.93110)

# Геопространственные операции
# Буфер вокруг точки (в градусах)
gdf['буфер'] = gdf.geometry.buffer(1.0)

# Экспорт обратно в GeoJSON
result_geojson = gdf.to_json()

import pandas as pd

# Сложная вложенная структура
complex_json = '''
{
  "компания": "ТехКорп",
  "отделы": [
    {
      "название": "Разработка",
      "сотрудники": [
        {"имя": "Иван", "зарплата": 150000, "проекты": ["А", "Б"]},
        {"имя": "Мария", "зарплата": 120000, "проекты": ["В"]}
      ]
    },
    {
      "название": "Маркетинг",
      "сотрудники": [
        {"имя": "Петр", "зарплата": 100000, "проекты": ["Г", "Д"]},
        {"имя": "Анна", "зарплата": 90000, "проекты": ["Е"]}
      ]
    }
  ]
}
'''

data = json.loads(complex_json)

# Нормализация с указанием путей
df = pd.json_normalize(
    data,
    record_path=['отделы', 'сотрудники'],
    meta=[
        'компания',
        ['отделы', 'название']
    ],
    meta_prefix='meta_',
    record_prefix='сотр_'
)

print(df)

import plotly.express as px
import pandas as pd
import json

# JSON для визуализации
viz_json = '''
[
  {"город": "Москва", "продажи": 5000, "регион": "Центр"},
  {"город": "СПб", "продажи": 3000, "регион": "Северо-Запад"},
  {"город": "Казань", "продажи": 2000, "регион": "Поволжье"}
]
'''

df = pd.read_json(viz_json)

# Создание графика
fig = px.bar(
    df,
    x='город',
    y='продажи',
    color='регион',
    title='Продажи по городам'
)

# Экспорт графика в JSON
fig_json = fig.to_json()

# Сохранение
with open('график.json', 'w', encoding='utf-8') as f:
    f.write(fig_json)

import jsonschema

# Определение схемы
схема = {
    "type": "object",
    "properties": {
        "имя": {"type": "string"},
        "возраст": {"type": "number", "minimum": 0},
        "email": {"type": "string", "format": "email"}
    },
    "required": ["имя", "email"]
}

# Валидация
def валидировать_запись(запись, схема):
    try:
        jsonschema.validate(instance=запись, schema=схема)
        return True
    except jsonschema.ValidationError as e:
        print(f"Ошибка валидации: {e.message}")
        return False

# Использование
запись = {"имя": "Иван", "возраст": 30, "email": "ivan@example.com"}
if валидировать_запись(запись, схема):
    # Обработка валидной записи
    pass

def безопасная_загрузка_json(filepath):
    """
    Безопасная загрузка JSON с обработкой ошибок
    """
    try:
        with open(filepath, 'r', encoding='utf-8') as f:
            data = json.load(f)
        return pd.DataFrame(data)
    except FileNotFoundError:
        print(f"Файл не найден: {filepath}")
        return pd.DataFrame()
    except json.JSONDecodeError as e:
        print(f"Ошибка парсинга JSON: {e}")
        return pd.DataFrame()
    except Exception as e:
        print(f"Неожиданная ошибка: {e}")
        return pd.DataFrame()

df = безопасная_загрузка_json('data.json')

# Чтение JSON с оптимизацией типов
df = pd.read_json('data.json')

# Оптимизация типов данных
for col in df.select_dtypes(include=['int64']).columns:
    df[col] = pd.to_numeric(df[col], downcast='integer')

for col in df.select_dtypes(include=['float64']).columns:
    df[col] = pd.to_numeric(df[col], downcast='float')

# Категориальные данные
for col in ['город', 'категория']:
    if col in df.columns:
        df[col] = df[col].astype('category')

print(f"Использование памяти: {df.memory_usage(deep=True).sum() / 10242:.2f} MB")

Заключение

JSON - мощный формат для хранения и обмена данными в Data Science. С правильными инструментами и подходами, вы можете эффективно работать с JSON данными любого размера и сложности.

Ключевые выводы:

• Pandas отлично справляется с JSON
• Используйте streaming для больших файлов
• JSON удобен для конфигураций ML
• Нормализация упрощает работу с вложенными данными
• Валидация критически важна для качества данных
• Оптимизируйте типы данных для экономии памяти

С этими знаниями вы готовы использовать JSON в ваших Data Science проектах!