Делюсь опытом и готовыми решениями по сбору и структурированию сырых данных, превращая их в полезный инструмент для аналитиков.

Введение

Сырые данные — это основа для принятия обоснованных решений в современном бизнесе: от прогнозирования поведения потребителей и рыночных трендов до управления товарным ассортиментом и ценообразованием.

Однако сами по себе сырые данные редко несут практическую ценность и доступны только узкому кругу технических специалистов. Чтобы превратить массивы несортированной информации в полезные инсайты, требуется продуманный подход к их сбору, обработке и анализу.

Меня зовут Вячеслав, я руковожу командой парсинга и обработки данных. В этой статье разберу, какие виды данных можно собирать, как их обрабатывать и визуализировать, чтобы повысить их доступность для аналитиков и других сотрудников.

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

Также в статье затронем факторы, влияющие на выбор типов хранилища данных (SQL и NoSQL), подходы к анализу данных с применением Apache Superset, а также возможности автоматизации обработки информации с использованием OCR и GPT инструментов.

Приведенные в статье примеры взяты из разработанных нами решений в сфере аналитики продаж, товарной номенклатуры и мониторинга рынка конкурентов. Однако эти подходы будут применимы и в других областях.

Какие сырые данные могут пригодиться бизнесу

Сырые данные — это исходная, необработанная информация, собранная из различных источников. Это могут быть числовые значения, текст, фотографии, звуковые записи или видео.

Примеры сырых данных:

Логи web-сервера, которые содержат детали запросов, время обращения и статусы ответов.
Результаты анкетирования в сыром виде (без анализа и обобщения).
Данные из сенсоров устройств.
Цены на каждый товар в ассортименте продавца в разные моменты времени.
Цены с сайтов конкурентов.

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

Сырые данные могут быть сложны для понимания из-за их неупорядоченности — их нужно подготовить для анализа и использования.
Они часто содержат ошибки и дубли, которые необходимо исправить. Очистка сырых данных повышает их надежность.
Представлены в исходном формате в виде текстовых файлов или неструктурированных баз данных. Их нужно привести к упорядоченному виду — таблицам, графикам или сводным отчетам.
У сырых данных обычно слишком большой объем, так как они включают всю собранную информацию. В процессе обработки данные фильтруются, и их объем может значительно уменьшаться.

Критерий	Сырые данные	Обработанные данные
Структура	Неструктурированные	Структурированные
Формат	Уникальный для каждого источника	Удобный для анализа, чаще унифицированный
Объём	Больше	Меньше
Скорость взаимодействия	Ниже	Выше

После обработки сырых данных на выходе можно получить, например:

Очищенные и обобщенные отчеты по продажам со средними значениями и трендами.
Исторические данные, подготовленные для моделирования или анализа.
Дашборды, построенные на основе очищенных данных.

Сбор и хранение сырых данных с помощью парсеров

Большинство запросов компаний на данные можно поделить на два вида: статические (которые не меняются со временем, например, характеристики товарной позиции) и динамические (чаще всего это цены, которые в свою очередь могут быть текущими, либо с историей изменений).

Для наполнения хранилища статических данных мы пишем парсеры сайтов с похожей номенклатурой компаний, а также более универсальные агрегаторы, такие как icecat или маркетплейсы с общераспространенным ассортиментом товаров.

Собирать сырые данные с сайтов с SSR рендерингом сложнее, модели и структуру данных приходится придумывать самим. Но зато можно сразу сложить в реляционную SQL БД. Например, привести в нормальную форму наименования характеристик. В дальнейшем это упрощает/ускоряет выборки данных.

С сайтов, выдающих данные через API, мы обычно забираем и кладём json модели ”как есть” в MongoDB. Ибо никогда заранее не знаешь, какая же часть окажется полезной, поэтому чем полнее, тем лучше.

А вот для ответов на вопрос об актуальных ценах такая история подходит плохо, так как время обхода большинства сайтов заведомо больше периода актуальности цены. Поэтому при обогащении информации о цене конкурентов для нужного товара мы идём искать в онлайн-поисковик. Собираем ссылки на аналогичные товары и парсим цены и доступность на текущий момент.

Тут решаются сразу три проблемы:

Так как надо собрать как можно больше цен, то нужно было бы парсить много источников сырых данных в отличие от характеристик, где достаточно только одного достаточно качественного.
Характеристики часто лежат на сайте как попало, их тяжело вытаскивать, поэтому надо написать кастомный парсер под конкретный сайт. Цены же часто размечены микроразметкой и их можно собрать с рандомных сайтов, найденных поисковиком, универсальным парсером.
Использование поискового механизма поисковой системы для подбора релевантных товаров позволяет обойтись без собственной технологии нечеткого поиска (а формат написания названия товара на разных площадках порой сильно отличается), плюс при необходимости можно учесть наличие и региональность цен.

По найденным в результатах поиска ссылкам на релевантные искомому запросу товары можно запустить периодический сбор исторической информации об изменениях цен. В отличие от сбора цен в течение одного прохода, при превентивном парсинге всей номенклатуры крупного источника, который может занимать недели и месяцы, тут мы можем получать данные ежесуточно, так как обычно пул товаров для мониторинга у заказчика на порядок меньше общего массива товаров в наших холодных хранилищах.

В общем виде схема сбора сырых данных для товара выглядит следующим образом:

Конвейер обработки сырых данных в структурированную форму

Превентивный парсинг нескольких источников с сырыми данными о товарах в отдельные базы с разной структурой очевидно делает невозможным одновременный поиск по ним. Поэтому следующий этап — это унификация модели данных в едином хранилище и сбор единого поискового индекса.

Мы уже много лет для этих целей используем Elasticsearch. Для его наполнения у нас запущен пул утилит-индексаторов, содержащих провайдеры, собирающие обобщенную модель товара с полями, присутствующими в большинстве распаршенных источников. Схема работы типичная: парсер источника парсит очередной товар, создаёт событие с его id в Message Queue сервисе, откуда индексатор соответствующим провайдером читает БД с сырыми данными и собирает обобщенную модель для эластика.

Мы находим очень удобным гибкость эластика в работе со схемами и простоте пересборки индексов. Периодически мы меняем, расширяем обобщенную модель и пересобираем индекс под текущие потребности.

Одновременно с хранением массива данных о товарах по унифицированной модели, собранного из нескольких источников сырых данных, мы также пользуемся и движком полнотекстового поиска эластика, так как в номенклатуре товаров компании не всегда указан или вынесен в отдельное поле артикул товара.

Пример модели данных товара для Elasticsearch:

{
  "id": "10_12312312",
  "supplierId": "123456",
  "source": 10,
  "truName": "Светильник светодиодный Duwi Nuovo LED",
  "partNumber": "1111-2222",
  "description": "Накладной настенный светильник",
  "parseDate": "2025-01-14T09:13:38.2866667",
  "detailsUrl": "https://site.ru/catalog/12312312/",
  "prices": [
    {
      "id": "6681726",
      "regionId": [
        77
      ],
      "price": 1628,
      "priceDate": "2024-12-31T11:43:11"
    },
    {
      "id": "6782342",
      "regionId": [
        77
      ],
      "price": 1638,
      "priceDate": "2025-01-14T09:13:38"
    }
  ],
  "specifications": [
    {
      "key": "Глубина",
      "value": "254"
    },
    {
      "key": "Цвет товара",
      "value": "Stainless steel"
    },
    {
      "key": "Высота",
      "value": "63.5"
    },
    {
      "key": "Ширина",
      "value": "254"
    }
  ],
  "category": "Освещение",
  "imageUrl": "https://site.ru/image/12312312.jpg",
  "supplierInn": "***********",
  "supplierKpp": "***********",
  "supplierOgrn": "***********",
  "supplierName": "ООО Ромашка"
}

Далее оперативный парсинг. В момент поступления запроса на получение актуальных цен на товар, мы создаем задание в очереди, реализованной на базе коллекции в MongoDB. Не самое типичное решение, но в нашем случае оказалось очень удобным.

У нас имеется шаблонная консольная утилита на базе BackgroundService, которая в бесконечном цикле опрашивает и исполняет необработанные запросы в какой-либо коллекции методами TryGetMessage, ProcessMessage и CommitMessage.

internal async Task<TMessage> TryGetMessage(CancellationToken cancellationToken) where TMessage : IMessage
{
    var now = DateTime.UtcNow;

    var receiveTimeCriteria = _filters.Lt(f => f.RetainDate, now.Subtract(_options.DefaultMessageLease));
    var deliveryCriteria = _filters.Lt(f => f.TryCount, _options.MaxDeliveryAttempts);
    var isSuccessCriteria = _filters.Eq(f => f.IsSuccess, false);
    var filter = deliveryCriteria & receiveTimeCriteria & isSuccessCriteria;

    var updateReceiveTime = _updates.Set(x => x.RetainDate, now);
    var updateStatus = _updates.Set(x => x.Status, (int)QueueMessageStatusEnum.InProgress);
    var incrementAttempts = _updates.Inc(x => x.TryCount, 1);
    var update = _updates.Combine(updateReceiveTime, incrementAttempts, updateStatus);

    try
    {
        var document = await _collection.FindOneAndUpdateAsync(filter, update, _findOneAndUpdateOptions, cancellationToken);

return document;
    }
    catch (Exception exception)
    {
        _logger.LogError(exception, "Could not received document");
    }

    return default;
}

internal async Task ProcessMessage(IMessage message, CancellationToken cancellationToken)
{
    try
    {
        message.Response = await _myCustomMessageProcessor.ProcessMessage(message, cancellationToken);
        message.IsSuccess = true;
    }
    catch (Exception ex)
    {
        message.IsSuccess = false;
        message.LastErrorReason = ex.Message;
    }
    message.CompleteDate = DateTime.UtcNow;
}

internal async Task<bool> CommitMessage(IMessage message, CancellationToken cancellationToken)
{
    if (message.IsSuccess)
    {
        return await _messageResolver.WriteAck(message, cancellationToken);
    }

    return await _messageResolver.Nack(message, cancellationToken);
}

Тело запроса при этом должно реализовывать интерфейс IMessage

public interface IMessage
 {
     string Id { get; }
     int TryCount { get; }
     bool IsSuccess { get; set; }
     DateTime RetainDate { get; }
     string LastErrorReason { get; set; }
     int Status { get; set; }
 }

Используя эту шаблонную утилиту и расширяя базовое тело IMessage нужными в конкретном сценарии полями (например, поле Request с параметрами для какого-либо http запроса и Response с его ответом), мы легко масштабируем различные обработчики очередей, которые одновременно являются и хранилищем результатов обработки.

{
  "_id": "8de2f6f7-6c3d-48c7-8aec-444ae7602fef",
  "TryCount": 0,
  "Status": 2,
  "IsSuccess": true,
  "RetainDate": "2024-08-22T13:38:03.500+0000",
  "LastErrorReason": "",
  "CreateDate": "2024-08-06T19:16:53.649+0000",
  "CompleteDate": "2024-08-06T19:38:43.185+0000",
  "Request": {},
  "Response": {}
}

Из плюсов такого подхода: минус одна зависимость от брокера сообщений в поддержке инфраструктуры, наглядность процесса обработки, удобство ручной правки параметров и перезапуска событий. В ситуациях, когда по факту поток событий относительно мал, каждое событие обрабатывается десятки секунд и часто далеко не с первой попытки, ибо процесс парсинга всё же довольно нестабилен ввиду различных ограничений со стороны источников и их постоянной изменчивости. Поэтому высокая скорость и низкое потребление ресурсов настоящих брокеров сообщений тут не особо полезна, а вот простота разработки, дебага и мониторинга позволяют оперативно разрешать поток нештатных ситуаций.

Отдельным конвейером идет парсинг нетипичных источников цен. Файлы с различными выгрузками — еще один пример сырых данных, слабо пригодных для автоматической обработки. Прайсы часто бывают и в виде таблиц, и просто pdf файлы без текстового слоя. Для первичной конвертации любых форматов файлов в pdf мы используем проект gotenberg в виде docker контейнера. Он весьма всеяден и имеет простое API. Далее стандартизированный pdf отправляем на распознание в Apache Tika с Tesseract-OCR, либо в какой-либо онлайн сервис. Полученный текст отправляем в один из популярных GPT-сервисов. Существующие на данный момент даже базовые модели уже хорошо умеют извлекать нужную информацию о товаре с помощью подобных промптов:

Определи следующие характеристики из текста:
ИНН продавца,
КПП продавца,
Телефон продавца,
Наличие, если в наличии, то укажи в поле слово 'В наличии'
Размер предоплаты,
Сумма с НДС 
Количество наименований (количество товаров),
Для каждого определи следующий список характеристик:
Артикул,
Наименование товара (работы, услуги),
Наличие, если в наличии, то укажи в поле слово 'В наличии', 
Цена с НДС,
Количество (кол-во)

Анализ и визуализация данных

Итогом всех проведенных манипуляций над сырыми данными становится возможность их дальнейшего анализа уже более широкому кругу специалистов-не разработчиков. После известных событий выбор инструментария бизнес-аналитики несколько сузился, и наши сотрудники переходят на некоммерческие продукты.

На наш взгляд, опенсорсный Apache Superset — достойная альтернатива коммерческим Power BI и Tableau. Это гибкий инструмент для визуализации и быстрого анализа больших объемов информации.

Apache Superset поддерживает разнообразные типы графиков и визуализаций, включая линейные и столбчатые диаграммы, карты, тепловые карты, пай-чарты и многие другие. Гибкость в выборе визуализаций позволяет настроить дашборды для любых бизнес-задач. Отдельно отмечу широкий набор источников данных, такие как PostgreSQL, MySQL, Oracle, Google BigQuery, MSSQL и многие другие благодаря возможности работы через модули SQLAlchemy. А для NoSQL (той же MongoDB) можно воспользоваться лайфхаком: подключить монгу к Apache Drill, а из него уже забирать данные коллекций в Superset.

Базовый пример визуализации изменения цены на товар за последние два года:

Выводы

Подведем итоги:

Чтобы сырые данные превратились в сводные отчеты и дашборды, доступные широкому кругу специалистов, необходимо провести серьезную подготовительную работу от сбора и хранения данных до их визуализации. Это достаточно большие объемы человеко-часов, которые нужно планировать до старта работ.

В задачах, когда данные должны быть собраны из внешних источников, выбор метода парсинга зависит от скорости устаревания данных: редко изменяющиеся характеристики практичнее выкачать заранее с одного крупного источника, динамические — перезакачивать с нужной периодичностью по точечным ссылкам, найденным в поисковых системах или сайтмапах. Подход к хранению зависит от типа данных: SQL используется для структурированных данных, MongoDB — для JSON-моделей, где заранее неизвестно, какие именно параметры будут полезны. Собирать актуальные цены лучше всего через запросы к поисковым системам.

Для унификации неструктурированных данных рекомендую присмотреться к Elasticsearch, который не только обеспечивает удобство поиска, но и позволяет гибко адаптировать модели под меняющиеся бизнес-задачи. Для оперативного парсинга актуальных данных целесообразно выстраивать многостадийный конвейер из очередей задач, которые делают процесс обработки найденных/скачанных результатов более прозрачным, управляемым и отказоустойчивым.

GPT-сервисы хорошо справляются с извлечением нужной информации из объемных файлов и вёрстки. Можно использовать для парсинга сложных, нетипичных источников.

Главное, за что ваш аналитик скажет вам спасибо — это наглядная инфографика. Для замены Power BI и Tableau можно использовать опенсорсные продукты, например Apache Superset.

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

Как превратить сырые данные в аналитический отчет