Улучшение аналитики с помощью временных рядов и векторных баз данных
Анализ временных рядов играет ключевую роль во многих областях, особенно в устройствах Интернета вещей (IoT). С помощью данных временных рядов мы можем выявлять закономерности и тренды за определенные периоды, что позволяет нам прогнозировать и анализировать будущие события, зависящие от времени. Распространенные примеры использования временных рядов включают прогнозирование температуры погоды и цен на акции, а также мониторинг данных датчиков.
Как база данных временных рядов, InfluxDB предоставляет решение для хранения огромных объемов данных временных рядов. InfluxDB высоко оптимизирована для хранения и запросов данных, зависящих от времени, с использованием таких техник, как агрегации и даунсэмплинг. Однако полагаться только на базы данных временных рядов может быть сложно, особенно если наш сценарий использования требует выполнения поиска по сходству.
В недавнем выступлении на Zilliz Unstructured Data Meetup Зои Стейнкамп, Developer Advocate в InfluxDB, обсудила подход к объединению InfluxDB с Milvus для хранения, запросов и выполнения поиска по сходству в сценариях использования, зависящих от времени.
В этой статье мы рассмотрим эту тему более подробно и проведем вас через сценарий использования, в котором мы будем хранить данные временных рядов в InfluxDB, запрашивать данные, преобразовывать их в векторные эмбеддинги, сохранять эмбеддинги в Milvus и, наконец, выполнять поиск по сходству с помощью Milvus. Итак, без лишних слов, начнем.
Понимание данных временных рядов
Данные временных рядов представляют собой хронологически упорядоченные наблюдения, записанные через определенные интервалы, например почасово, ежедневно, еженедельно или ежемесячно.
Мы можем найти сценарии использования временных рядов в повседневной жизни: от почасовых измерений температуры, ежедневной статистики дорожного трафика до ежемесячных показателей розничных продаж. Анализируя данные временных рядов, мы можем выявлять исторические закономерности и использовать эти insights для принятия будущих решений.
При анализе данных временных рядов мы часто видим один или несколько из следующих паттернов:
Сезонный: Повторяющиеся колебания через фиксированные временные интервалы, на которые влияют такие факторы, как дни недели или месяцы.
Тренд: Устойчивое увеличение или уменьшение данных в течение длительного периода.
Циклический: Колебания, похожие на сезонные паттерны, но без фиксированной частоты.
Сверху слева — сезонный, сверху справа — убывающий тренд, снизу слева — возрастающий тренд, снизу справа — циклический. Источник.
Понимание этих паттернов имеет решающее значение для эффективного прогнозирования временных рядов. Прогнозирование временных рядов — это мощная аналитическая техника, которая предсказывает будущие значения переменных, зависящих от времени, на основе исторических данных. Например, если у нас есть история цен акции за предыдущие два года и мы хотим предсказать завтрашнюю цену акции, мы, по сути, выполняем прогнозирование временных рядов.
Компании в разных секторах применяют концепцию прогнозирования временных рядов в своей повседневной деятельности для различных целей:
Финансовые учреждения: Прогнозирование цен на акции, прогнозирование колебаний валютных курсов, выявление необычных паттернов в расходах клиентов или разработка надежных стратегий управления рисками.
Сектор здравоохранения: Мониторинг распространения заболеваний, отслеживание жизненно важных показателей пациентов в реальном времени или улучшение общего ухода за пациентами и результатов лечения.
Розничная индустрия: Прогнозирование объемов продаж, понимание покупательского поведения клиентов, оптимизация управления запасами, тонкая настройка ценовых стратегий и повышение общей прибыльности.
Интернет вещей (IoT): Системы умного дома используют данные датчиков для автоматизации задач, а термостаты и домашние хабы обмениваются информацией для оптимизации энергопотребления.
Производство: Сокращение простоев оборудования, внедрение стратегий предиктивного обслуживания и повышение общей операционной эффективности.
Из-за своей зависимости от времени данные временных рядов обычно собираются непрерывно в реальном времени. В результате объем данных временных рядов может быстро расти, что требует масштабируемого и эффективного решения для хранения. Именно здесь в игру вступают базы данных временных рядов.
InfluxDB как база данных временных рядов
Базы данных временных рядов должны обладать несколькими ключевыми характеристиками по сравнению с традиционными реляционными базами данных, чтобы упростить процесс хранения и извлечения больших объемов данных временных рядов. Как упоминалось ранее, данные временных рядов часто собираются в реальном времени; следовательно, данные необходимо обрабатывать и синхронизировать между всеми системами сразу после того, как они поступают в базу данных.
Базы данных временных рядов обеспечивают более высокую производительность и эффективность, чем реляционные базы данных, для нагрузок, основанных на времени, особенно при работе с большими объемами данных и высокочастотными записями. Кроме того, база данных временных рядов предлагает высокооптимизированные решения для таких операций, как прием данных, запросы и извлечение по временным диапазонам, а также аналитика и агрегации на основе времени.
Важные функции в базе данных временных рядов
InfluxDB — одна из баз данных временных рядов с открытым исходным кодом, которую мы можем использовать для хранения данных временных рядов. Она написана на Rust и высоко оптимизирована для операций extract, transform, load (ETL) в базах данных. Чтобы обеспечить быстрый прием данных и синхронизацию временных меток, InfluxDB содержит встроенную службу времени, использующую Network Time Protocol (NTP).
Данные временных рядов, которые мы храним внутри базы данных временных рядов, обычно состоят всего из трех столбцов: тег, поле и временная метка. Тег содержит метаданные измеряемых нами данных, поле содержит значения, которые мы далее анализируем, а временная метка указывает, когда данные были собраны.
Допустим, мы сохраняем данные измерений датчиков каждые 5 минут. Данные внутри нашего временного ряда будут выглядеть примерно как на визуализации ниже:
Пример данных, хранящихся внутри базы данных временных рядов
Теперь, когда мы знаем основы баз данных временных рядов, давайте поговорим о векторных базах данных!
Milvus как векторная база данных
Как следует из названия, векторная база данных хранит данные (включая изображения, тексты, документы и т. д.) в виде векторов. Мы преобразуем наши данные в векторы с помощью одной из многочисленных моделей глубокого обучения.
Размерность вектора зависит от метода или модели глубокого обучения, которую мы используем для его генерации. Например, если мы используем модель под названием all-MiniLM-L6-v2, мы получим вектор размерности 384. Между тем, если мы используем модель под названием all-mpnet-base-v2, мы получим вектор размерности 768.
Как вы, возможно, знаете, вектор — это не просто математический объект; он заключает в себе как величину, так и направление, передавая семантический смысл текста, изображения или документа, который он представляет. Именно эта глубокая особенность векторов позволяет им располагаться близко друг к другу в векторном пространстве, представляя похожие тексты или изображения.
Семантическое сходство между векторами в векторном пространстве
Идея о том, что вектор несет семантический смысл данных, позволяет нам сравнивать сходство любой пары векторов с помощью таких метрик, как косинусное сходство или евклидово расстояние. Если косинусное сходство между двумя векторами близко к единице, то эти два вектора очень похожи, и наоборот.
Хранение и выполнение поиска по сходству может быть простым в реализации, если у нас всего несколько векторов. Однако в реальных приложениях мы обычно имеем дело с сотнями тысяч или миллионами векторов, что делает всю операцию более затратной для поддержания как с точки зрения времени, так и вычислительных ресурсов. Именно здесь становится необходимой векторная база данных.
Векторные базы данных, такие как Milvus и Zilliz Cloud, предлагают высокомасштабируемые и эффективные процессы для хранения миллионов векторов. Они также хорошо оптимизированы для более быстрого поиска векторного сходства и извлечения данных благодаря широкому разнообразию продвинутых методов индексирования, таких как IVF-Flat, HNSW и другие.
Полный рабочий процесс операции векторного поиска
Рабочий процесс хранения и извлечения данных из Milvus или любой другой векторной базы данных выглядит следующим образом: сначала мы преобразуем наши входные данные, которые могут быть текстами или изображениями, в векторы с помощью выбранной нами техники или модели. Затем мы загружаем эти векторы вместе с их метаданными в векторную базу данных и строим индекс с помощью выбранного нами метода индексирования.
В процессе извлечения нам сначала нужно преобразовать наш запрос в вектор, используя ту же технику или модель, которую мы использовали во время процесса загрузки. Затем мы выполняем операцию векторного поиска, чтобы сравнить этот вектор запроса с векторами, хранящимися в нашей векторной базе данных. Наконец, наиболее похожие векторы в базе данных возвращаются нам в качестве результатов.
Сценарий использования сочетания базы данных временных рядов и векторной базы данных
Поскольку базы данных временных рядов и векторные базы данных хорошо оптимизированы для разных сценариев использования, мы можем использовать сильные стороны обеих баз данных в наших реальных проектах.
Например, представим сценарий, в котором мы разрабатываем систему для анализа дорожной обстановки в реальном времени в умном городе. Для этого мы установили бы датчики во многих местах, которые могут отслеживать скорость транспортных средств, количество транспортных средств и другие релевантные метрики. Средняя скорость транспортных средств и их количество за определенные временные интервалы могли бы непрерывно сохраняться внутри базы данных временных рядов.
Помимо датчиков, мы также могли бы установить камеры для захвата фотографий или видео фактической дорожной обстановки. В этом случае мы можем преобразовать фотографии и видео в векторы с помощью выбранной нами модели глубокого обучения и сохранить эти векторы в векторной базе данных. Объединяя данные временных рядов из базы данных временных рядов и векторы из векторной базы данных, мы можем выполнять обнаружение аномалий в дорожной обстановке.
Пример сценария использования сочетания InfluxDB и Milvus для анализа дорожной обстановки
В этом разделе мы реализуем простой сценарий использования, связанный с дорожной обстановкой, с помощью InfluxDB и Milvus. В частности, мы будем хранить данные временных рядов внутри InfluxDB, а соответствующие векторы — внутри Milvus.
Давайте сначала сгенерируем фиктивные данные временных рядов, состоящие из временной метки, средней скорости транспортных средств, количества транспортных средств и типа аномалии. Тип аномалии состоит из двух различных значений: "normal" и "accident".
Мы начнем с генерации 500 "нормальных" точек данных, собираемых каждые 10 минут.
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import random
def generate_sensor_data(anomaly_type, start_time, vehicle_count_range, avg_speed_range, rows=500, seed = 42):
np.random.seed(seed)
vehicle_counts = np.random.randint(vehicle_count_range[0], vehicle_count_range[1], size=rows)
avg_speeds = np.random.uniform(avg_speed_range[0], avg_speed_range[1], size=rows)
start_time = datetime.strptime(start_time, "%Y-%m-%d %H:%M:%S")
timestamps = [start_time + timedelta(minutes=10*i) for i in range(rows)]
df = pd.DataFrame({
'Timestamp': timestamps,
'Количество транспортных средств': vehicle_counts,
'Средняя скорость': avg_speeds,
'Тип аномалии': anomaly_type
})
return df
vehicle_count_range = (5, 10)
avg_speed_range = (60.0, 80.0)
df_normal = generate_sensor_data("normal", "2024-09-15 18:00:00", vehicle_count_range, avg_speed_range)
DataFrame normal
Далее сгенерируем 500 точек данных "accident", также собираемых каждые 10 минут. Как вы можете представить, двумя возможными признаками аварии на улицах являются более большое количество транспортных средств и более низкая средняя скорость из-за заторов.
vehicle_count_range = (20, 40)
avg_speed_range = (10.0, 20.0)
df_accident = generate_sensor_data("accident", "2024-09-19 05:20:00", vehicle_count_range, avg_speed_range)
DataFrame accident
Теперь объединим два dataframe, а затем вставим объединенные данные в InfluxDB с помощью команды, которая выглядит следующим образом:
from influxdb_client_3 import InfluxDBClient3
df = pd.concat([df_accident, df_normal], axis=0)
client = InfluxDBClient3(token="DATABASE_TOKEN",
host="HOST",
database="DATABASE_NAME")
client.write(bucket="DATABASE_NAME", record=df, data_frame_measurement_name='traffic_data', data_frame_tag_columns=['Anomaly Type'], data_frame_timestamp_column='Timestamp')
И вы можете выполнить запрос к данным внутри InfluxDB с помощью следующей команды.
query = "SELECT * FROM traffic_data WHERE time >= now() - INTERVAL '90 days'"
pd = client.query(query=query, mode="pandas")
Если вы хотите узнать больше о различных операциях подробнее, пожалуйста, ознакомьтесь с документацией InfluxDB Python Client Library.
Теперь давайте сгенерируем некоторые векторные данные. Как показано выше, мы можем использовать фотографии или видео дорожной обстановки в определенные временные интервалы в качестве источников данных и преобразовывать их в векторы с помощью модели глубокого обучения. Однако, поскольку у нас нет фотографий или видео, мы собираемся собрать среднюю скорость транспортных средств за определенные временные интервалы в список, по сути превращая их в векторы. В анализе временных рядов этот метод называется оконным преобразованием.
В следующем примере мы установим размер окна равным 24, но вы можете изменить его на любой размер, который предпочитаете. Это означает, что для каждого наблюдения мы будем собирать значения средней скорости транспортных средств за следующие 240 минут, поскольку каждая точка данных собирается каждые 10 минут. В результате мы получим 24-мерный вектор для каждого наблюдения.
Для каждого наблюдения мы будем использовать размер шага 10. Это означает, что время начала между последовательными наблюдениями отличается на 100 минут. Кроме того, нам нужно выполнить нормализацию каждого элемента вектора для каждого наблюдения, чтобы убедиться, что вектор равномерно масштабирован от 0 до 1.
window_size = 24
step_size = 10
min_val = 10 # Мин. средняя скорость транспортных средств
max_val = 80 # Макс. средняя скорость транспортных средств
def vectorize_data(df):
windows = [
df.iloc[i : i + window_size]
for i in range(0, len(df) - window_size + 1, step_size)
]
start_times = [w["Timestamp"].iloc[0] for w in windows]
end_times = [w["Timestamp"].iloc[-1] for w in windows]
avg_speed_values = [w["Average Speed"].tolist() for w in windows]
anomaly_types = [w["Anomaly Type"].tolist()[0] for w in windows]
# Создаем новый DataFrame из собранных данных
embedding_df = pd.DataFrame(
{"start_time": start_times, "end_time": end_times, "vectors": avg_speed_values, "anomaly_types": anomaly_types}
)
embedding_df["vectors"] = embedding_df["vectors"].apply(normalize_vector)
# Применяем lambda-функцию для преобразования временных меток в формат Unix timestamp.
embedding_df['start_time'] = embedding_df['start_time'].apply(lambda x: pd.Timestamp(x).timestamp()).astype(int)
embedding_df['end_time'] = embedding_df['end_time'].apply(lambda x: pd.Timestamp(x).timestamp()).astype(int)
return embedding_df
# Function to normalize the sensor column
def normalize_vector(vectors: list) -> list:
return (
[0.0] * len(vectors)
if max_val == min_val
else [(v - min_val) / (max_val - min_val) for v in vectors]
)
embedding_df = vectorize_data(df)
векторы DataFrame
Как вы можете видеть выше, теперь у нас есть dataframe со временем начала, временем окончания, вектором и типом аномалии. Далее мы можем напрямую загрузить эти данные в базу данных Milvus. Самый простой способ начать работу с Milvus — использовать Milvus Lite, поэтому давайте сначала установим Milvus Lite, а затем создадим схему в соответствии со столбцами внутри нашего dataframe.
! pip install pymilvus==2.4.6
from pymilvus import MilvusClient, DataType
dim = 24
collection_name = "traffic_data"
milvus_client = MilvusClient("./local_test.db")
has_collection = milvus_client.has_collection(collection_name, timeout=5)
if has_collection:
milvus_client.drop_collection(collection_name)
schema = milvus_client.create_schema(enable_dynamic_field=True)
schema.add_field("id", DataType.INT64, is_primary=True)
schema.add_field("start_time", DataType.INT64)
schema.add_field("end_time", DataType.INT64)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=dim)
schema.add_field("anomaly_type", DataType.VARCHAR, max_length=64)
index_params = milvus_client.prepare_index_params()
index_params.add_index(field_name = "vector", metric_type="L2")
milvus_client.create_collection(collection_name, schema=schema, index_params=index_params, consistency_level="Strong")
Теперь, когда мы создали схему, как определено выше, мы можем загрузить наш dataframe в Milvus с помощью следующей команды:
data = [
{
"id": i,
"vector": embedding_df['vectors'].iloc[i],
"start_time": embedding_df['start_time'].iloc[i],
"end_time": embedding_df['end_time'].iloc[i],
"anomaly_type": embedding_df['anomaly_types'].iloc[i]
}
for i in range(len(embedding_df))
]
insert_result = milvus_client.insert(collection_name, data)
Вот и всё! Теперь мы можем выполнить операцию векторного поиска.
Давайте представим, что неделю спустя мы наблюдаем необычное снижение средней скорости транспортных средств в определенной области. В следующем примере мы сгенерируем эти данные вручную, используя функции, которые мы определили ранее. Однако в реальном сценарии вы, скорее всего, считывали бы и запрашивали эти данные из InfluxDB на основе временной метки и использовали бы полученную временную метку, чтобы найти соответствующий вектор внутри Milvus.
vehicle_count_range = (20, 40)
avg_speed_range = (10.0, 20.0)
df_test = generate_sensor_data("accident", "2024-09-26 12:00:00", vehicle_count_range, avg_speed_range, rows=30, seed = 1)
Далее мы преобразуем эти данные с помощью тех же методов оконного разбиения и нормализации, что и раньше.
query_vector = vectorize_data(df_test).vectors.values
И теперь мы можем выполнить операцию векторного поиска по вектору запроса с помощью следующей команды:
result = milvus_client.search(collection_name, query_vector, limit=3, output_fields=["vector", "anomaly_type"])
for hits in result:
for hit in hits:
print(f"hit: {hit}")
"""
Output:
hit: {'id': 21, 'distance': 0.04718003422021866, 'entity': {'anomaly_type': 'accident'}}
hit: {'id': 41, 'distance': 0.0641128420829773, 'entity': {'anomaly_type': 'accident'}}
hit: {'id': 34, 'distance': 0.06488440930843353, 'entity': {'anomaly_type': 'accident'}}
"""
Как вы можете видеть, три наиболее похожих на наш запрос наблюдения имеют тип аномалии "accident", что подтверждает низкую среднюю скорость, зафиксированную в данных нашего запроса. Если вы изучите визуализацию векторов между данными запроса и тремя наиболее похожими наблюдениями, вы заметите, что диапазон значений векторов похож между собой.
Сравнение между вектором запроса и тремя наиболее похожими векторами
Заключение
В этой статье мы рассмотрели сочетание баз данных временных рядов и векторных баз данных, сосредоточившись на сценарии использования в приложениях Интернета вещей (IoT). Базы данных временных рядов, такие как InfluxDB, очень эффективны для хранения и запроса хронологических данных, что крайне важно для таких приложений, как прогнозирование погоды, анализ фондового рынка и мониторинг датчиков. Однако они сталкиваются с ограничениями, когда речь идет о выполнении поиска по сходству.
Чтобы преодолеть это, мы объединили базы данных временных рядов с векторными базами данных, такими как Milvus, которые хранят данные в векторной форме, обеспечивая эффективный поиск по сходству с использованием таких методов, как косинусное сходство или евклидово расстояние. Благодаря объединению двух баз данных можно полностью использовать сильные стороны обеих систем. Как вы можете видеть в приведенном выше примере, данные временных рядов с датчиков могут храниться в InfluxDB, а векторные данные — в Milvus. Такая интеграция позволяет реализовывать продвинутые сценарии использования, такие как обнаружение аномалий в условиях дорожного движения в реальном времени.
Код, связанный с Milvus, представленный в этой статье, доступен через этот блокнот Colab.
Вы также можете ознакомиться с этой статьей, в которой рассматриваются методы предварительной обработки для преобразования данных временных рядов во вложение, подходящее для задач прогнозирования.
Читать далее
Zilliz Cloud Update: Tiered Storage, Business Critical Plan, Cross-Region Backup, and Pricing Changes
This release offers a rebuilt tiered storage with lower costs, a new Business Critical plan for enhanced security, and pricing updates, among other features.
Our Journey to 35K+ GitHub Stars: The Real Story of Building Milvus from Scratch
Join us in celebrating Milvus, the vector database that hit 35.5K stars on GitHub. Discover our story and how we’re making AI solutions easier for developers.
Why Not All VectorDBs Are Agent-Ready
Explore why choosing the right vector database is critical for scaling AI agents, and why traditional solutions fall short in production.