Mejora de la analítica con series temporales y bases de datos vectoriales
El análisis de series temporales desempeña un papel crucial en muchos campos, particularmente en dispositivos de Internet de las cosas (IoT). Con los datos de series temporales, podemos detectar patrones y tendencias durante períodos particulares, lo que nos permite pronosticar y analizar futuros eventos dependientes del tiempo. Ejemplos comunes de casos de uso de series temporales incluyen el pronóstico de temperaturas meteorológicas y precios de acciones, y el monitoreo de datos de sensores.
Como base de datos de series temporales, InfluxDB proporciona una solución para almacenar grandes cantidades de datos de series temporales. InfluxDB está altamente optimizado para almacenar y consultar datos dependientes del tiempo utilizando técnicas como agregaciones y downsampling. Sin embargo, depender únicamente de bases de datos de series temporales puede ser desafiante, especialmente si nuestro caso de uso nos exige realizar una búsqueda de similitud.
En una charla reciente en el Zilliz Unstructured Data Meetup, Zoe Steinkamp, Developer Advocate en InfluxDB, discutió un enfoque para combinar InfluxDB con Milvus para almacenar, consultar y realizar búsquedas de similitud en casos de uso dependientes del tiempo.
En este artículo, exploraremos este tema con más detalle y te guiaremos a través de un caso de uso en el que almacenaremos datos de series temporales en InfluxDB, consultaremos los datos, los transformaremos en embeddings vectoriales, almacenaremos los embeddings en Milvus y, finalmente, realizaremos una búsqueda de similitud con Milvus. Así que, sin más preámbulos, comencemos.
Comprensión de los datos de series temporales
Los datos de series temporales representan observaciones ordenadas cronológicamente registradas en intervalos específicos, como por hora, diarios, semanales o mensuales.
Podemos encontrar casos de uso de series temporales en nuestra vida cotidiana: desde lecturas de temperatura por hora, estadísticas diarias de tráfico vial, hasta cifras mensuales de ventas minoristas. Al analizar datos de series temporales, podemos descubrir patrones históricos y utilizar estos conocimientos para fundamentar decisiones futuras.
Al analizar datos de series temporales, a menudo vemos uno o más de los siguientes patrones:
Estacional: Fluctuaciones recurrentes en intervalos de tiempo fijos, influenciadas por factores como días de la semana o meses.
Tendencia: Un aumento o disminución constante en los datos durante un período prolongado.
Cíclico: Fluctuaciones similares a patrones estacionales, pero sin una frecuencia fija.
Arriba a la izquierda - estacional, arriba a la derecha -tendencia decreciente, abajo a la izquierda - tendencia creciente, abajo a la derecha - cíclico. Fuente.
Comprender estos patrones es crucial para una previsión eficaz de series temporales. La previsión de series temporales es una potente técnica analítica que predice valores futuros de variables dependientes del tiempo basándose en datos históricos. Por ejemplo, si tenemos el historial de precios de una acción de los dos años anteriores y queremos predecir el precio de la acción de mañana, esencialmente estamos realizando una previsión de series temporales.
Empresas de diferentes sectores han aplicado el concepto de previsión de series temporales en sus operaciones diarias para diversos propósitos:
Instituciones financieras: Predecir precios de acciones, pronosticar fluctuaciones de divisas, identificar patrones inusuales en el gasto de los clientes o desarrollar estrategias sólidas de gestión de riesgos.
Sector sanitario: Monitorear la propagación de enfermedades, hacer seguimiento de los signos vitales de los pacientes en tiempo real o mejorar la atención y los resultados generales de los pacientes.
Industria minorista: Pronosticar volúmenes de ventas, comprender los comportamientos de compra de los clientes, optimizar la gestión de inventario, ajustar las estrategias de precios y aumentar la rentabilidad general.
Internet de las cosas (IoT): Los sistemas de hogar inteligente utilizan datos de sensores para automatizar tareas, mientras que los termostatos y los centros domésticos intercambian información para optimizar el uso de energía.
Fabricación: Reducir el tiempo de inactividad de las máquinas, implementar estrategias de mantenimiento predictivo y mejorar la eficiencia operativa general.
Debido a su naturaleza dependiente del tiempo, los datos de series temporales suelen recopilarse continuamente en tiempo real. Como resultado, el volumen de datos de series temporales puede crecer rápidamente, lo que requiere una solución escalable y eficiente para el almacenamiento. Aquí es donde entran en juego las bases de datos de series temporales.
InfluxDB como base de datos de series temporales
Las bases de datos de series temporales deben poseer varias características clave en comparación con las bases de datos relacionales tradicionales para facilitar el proceso de almacenamiento y recuperación de grandes cantidades de datos de series temporales. Como se mencionó anteriormente, los datos de series temporales a menudo se recopilan en tiempo real; por lo tanto, los datos deben procesarse y sincronizarse en todos los sistemas tan pronto como llegan a la base de datos.
Las bases de datos de series temporales ofrecen mejor rendimiento y eficiencia que las bases de datos relacionales para cargas de trabajo basadas en el tiempo, especialmente cuando se trata de grandes volúmenes de datos y escrituras de alta frecuencia. Además, una base de datos de series temporales ofrece soluciones altamente optimizadas para operaciones como la ingesta de datos, las consultas y la recuperación en rangos de tiempo, así como análisis y agregaciones basados en el tiempo.
Características importantes en una base de datos de series temporales
InfluxDB es una de las bases de datos de series temporales de código abierto que podemos usar para almacenar datos de series temporales. Está escrita en Rust y está altamente optimizada para operaciones de extracción, transformación y carga (ETL) en bases de datos. Para facilitar la ingesta rápida de datos y la sincronización de marcas de tiempo, InfluxDB contiene un servicio de tiempo integrado que utiliza el Protocolo de Tiempo de Red (NTP).
Los datos de series temporales que almacenamos dentro de una base de datos de series temporales suelen constar de solo tres columnas: una etiqueta, un campo y una marca de tiempo. Una etiqueta contiene metadatos de los datos que medimos, un campo contiene valores que analizamos posteriormente, y una marca de tiempo indica cuándo se recopilaron los datos.
Supongamos que almacenamos datos de mediciones de sensores cada 5 minutos. Los datos dentro de nuestra serie temporal se verían como la visualización siguiente:
Ejemplo de datos almacenados dentro de una base de datos de series temporales
Ahora que conocemos los fundamentos de las bases de datos de series temporales, ¡hablemos de las bases de datos vectoriales!
Milvus como base de datos vectorial
Como su nombre indica, una base de datos vectorial almacena datos (incluidas imágenes, textos, documentos, etc.) en vectores. Transformamos nuestros datos en vectores utilizando uno de los muchos modelos de aprendizaje profundo.
La dimensión de un vector depende de la técnica o del modelo de aprendizaje profundo que usemos para generarlo. Por ejemplo, si usamos un modelo llamado all-MiniLM-L6-v2, obtendremos un vector de dimensión 384. Mientras tanto, si usamos un modelo llamado all-mpnet-base-v2, obtendremos un vector de dimensión 768.
Como quizá sepas, un vector no es solo un objeto matemático; encapsula tanto magnitud como dirección, llevando el significado semántico del texto, la imagen o el documento que representa. Este profundo aspecto de los vectores es lo que permite que se coloquen cerca unos de otros en el espacio vectorial, representando textos o imágenes similares.
Similitud semántica entre vectores en un espacio vectorial
La idea de que un vector porta el significado semántico de los datos nos permite comparar la similitud de cualquier par de vectores mediante métricas como la similitud coseno o la distancia euclidiana. Si la similitud coseno entre dos vectores está cerca de uno, entonces los dos vectores son muy similares, y viceversa.
Almacenar y realizar búsquedas de similitud podría ser sencillo de implementar si solo tenemos unos pocos vectores. Sin embargo, en aplicaciones reales, normalmente trabajamos con cientos de miles o millones de vectores, lo que hace que toda la operación sea más costosa de mantener tanto en términos de tiempo como de recursos computacionales. Aquí es donde una base de datos vectorial se vuelve necesaria.
Las bases de datos vectoriales como Milvus y Zilliz Cloud ofrecen procesos altamente escalables y eficientes para almacenar millones de vectores. También están altamente optimizadas para búsquedas de similitud vectorial y recuperación de datos más rápidas, gracias a su amplia variedad de métodos avanzados de indexación como IVF-Flat, HNSW y otros.
Flujo de trabajo completo de una operación de búsqueda vectorial
El flujo de trabajo para almacenar y recuperar datos de Milvus o de cualquier otra base de datos vectorial es el siguiente: Primero, transformamos nuestros datos de entrada, que podrían ser textos o imágenes, en vectores usando una técnica o modelo de nuestra elección. Luego, ingerimos estos vectores junto con sus metadatos en la base de datos vectorial y construimos un índice usando un método de indexación de nuestra elección.
Durante el proceso de recuperación, primero necesitamos transformar nuestra consulta en un vector usando la misma técnica o modelo que usamos durante el proceso de ingesta. Luego, realizamos una operación de búsqueda vectorial para comparar este vector de consulta con los vectores almacenados en nuestra base de datos vectorial. Finalmente, los vectores más similares de la base de datos se nos devuelven como resultados.
El caso de uso de combinar una base de datos de series temporales y una base de datos vectorial
Dado que las bases de datos de series temporales y vectoriales están altamente optimizadas para distintos casos de uso, podemos aprovechar las fortalezas de ambas bases de datos en nuestros proyectos reales.
Por ejemplo, imaginemos un escenario en el que estamos desarrollando un sistema para analizar las condiciones del tráfico en tiempo real en una ciudad inteligente. Para lograrlo, instalaríamos sensores en muchas ubicaciones que puedan rastrear la velocidad de los vehículos, el conteo de vehículos y otras métricas relevantes. La velocidad promedio de los vehículos y el conteo de vehículos durante intervalos de tiempo específicos podrían almacenarse continuamente dentro de una base de datos de series temporales.
Además de los sensores, también podríamos instalar cámaras para capturar fotos o videos de las condiciones reales del tráfico. En este caso, podemos transformar las fotos y los videos en vectores usando un modelo de aprendizaje profundo de nuestra elección y almacenar estos vectores en una base de datos vectorial. Al combinar los datos de series temporales de la base de datos de series temporales y los vectores de la base de datos vectorial, podemos realizar detección de anomalías en las condiciones del tráfico.
Ejemplo de caso de uso de combinar InfluxDB y Milvus para analizar las condiciones del tráfico
En esta sección, vamos a implementar un caso de uso simple relacionado con las condiciones del tráfico con la ayuda de InfluxDB y Milvus. Específicamente, almacenaremos datos de series temporales dentro de InfluxDB y los vectores correspondientes dentro de Milvus.
Primero generemos datos ficticios de series temporales que consisten en marca de tiempo, velocidad promedio de vehículos, conteo de vehículos y tipo de anomalía. El tipo de anomalía consta de dos valores distintos: "normal" y "accident".
Comenzaremos generando 500 puntos de datos "normal" recopilados cada 10 minutos.
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 Count': vehicle_counts,
'Average Speed': avg_speeds,
'Anomaly Type': 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
A continuación, generemos 500 puntos de datos de "accident", también recopilados cada 10 minutos. Como puedes imaginar, dos posibles señales de un accidente en las calles son un mayor conteo de vehículos y una velocidad promedio más lenta debido a la congestión del tráfico.
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 accidente
Ahora concatenemos los dos dataframes y luego insertemos los datos concatenados en InfluxDB con un comando parecido al siguiente:
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')
Y puedes hacer una consulta de los datos dentro de InfluxDB con el siguiente comando.
query = "SELECT * FROM traffic_data WHERE time >= now() - INTERVAL '90 days'"
pd = client.query(query=query, mode="pandas")
Si quieres obtener más información sobre las diferentes operaciones con más detalle, consulta esta documentación de la Biblioteca de cliente Python de InfluxDB.
Ahora generemos algunos datos vectoriales. Como se ilustró anteriormente, podemos usar fotos o videos de las condiciones del tráfico en intervalos de tiempo particulares como fuentes de datos y transformarlos en vectores mediante un modelo de aprendizaje profundo. Sin embargo, como no tenemos fotos ni videos, vamos a recopilar la velocidad promedio de los vehículos en intervalos de tiempo particulares en una lista, convirtiéndolos esencialmente en vectores. En el análisis de series temporales, este método se llama windowing.
En el siguiente ejemplo, estableceremos el tamaño de la ventana en 24, pero puedes ajustarlo al tamaño que prefieras. Esto significa que para cada observación, recopilaremos los valores de velocidad promedio de los vehículos para los siguientes 240 minutos, ya que cada punto de datos se recopila cada 10 minutos. Como resultado, terminaremos con un vector de 24 dimensiones para cada observación.
Para cada observación, usaremos un tamaño de paso de 10. Esto significa que la hora de inicio entre observaciones consecutivas difiere en 100 minutos. Además, necesitamos realizar la normalización en cada elemento vectorial para cada observación a fin de garantizar que el vector se escale uniformemente de 0 a 1.
window_size = 24
step_size = 10
min_val = 10 # Min average vehicle speed
max_val = 80 # Max average vehicle speed
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]
# Create a new DataFrame from the collected data
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)
# Apply a lambda function to convert timestamps to Unix timestamp format.
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)
Vectores de DataFrame
Como puedes ver arriba, lo que tenemos ahora es un dataframe con hora de inicio, hora de finalización, vector y tipo de anomalía. A continuación, podemos ingerir directamente estos datos en la base de datos Milvus. La forma más sencilla de empezar con Milvus es a través de Milvus Lite, así que primero instalemos Milvus Lite y luego creemos un esquema según las columnas dentro de nuestro 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")
Ahora que hemos creado un esquema como se definió anteriormente, podemos ingerir nuestro dataframe en Milvus con el siguiente comando:
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)
¡Y eso es todo! Ahora podemos realizar una operación de búsqueda vectorial.
Imaginemos que una semana después observamos una disminución inusual en la velocidad promedio de los vehículos en un área específica. En el siguiente ejemplo, generaremos estos datos manualmente usando funciones que hemos definido previamente. Sin embargo, en un escenario real, lo más probable es que leyeras y consultaras estos datos desde InfluxDB según la marca de tiempo, y usaras la marca de tiempo resultante para encontrar el vector correspondiente dentro de 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)
A continuación, transformamos estos datos con los mismos métodos de ventanas y normalización que antes.
query_vector = vectorize_data(df_test).vectors.values
Y ahora podemos realizar una operación de búsqueda vectorial en el vector de consulta con el siguiente comando:
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'}}
"""
Como puedes ver, las tres observaciones más similares a nuestra consulta tienen el tipo de anomalía "accident", lo que confirma la baja velocidad media registrada en los datos de nuestra consulta. Si examinas la visualización vectorial entre los datos de la consulta y las tres observaciones más similares, notarás que el rango de valores vectoriales es similar entre sí.
Comparación entre el vector de consulta y los tres vectores más similares
Conclusión
En este artículo, hemos explorado la combinación de bases de datos de series temporales y bases de datos vectoriales, centrándonos en el caso de uso en aplicaciones de Internet de las cosas (IoT). Las bases de datos de series temporales, como InfluxDB, son muy eficientes para almacenar y consultar datos cronológicos, lo cual es crucial para aplicaciones como la previsión meteorológica, el análisis del mercado de valores y la monitorización de sensores. Sin embargo, se enfrentan a limitaciones cuando se trata de realizar búsquedas de similitud.
Para superar esto, combinamos bases de datos de series temporales con bases de datos vectoriales como Milvus, que almacenan datos en forma vectorial, lo que permite realizar búsquedas de similitud eficientes utilizando técnicas como la similitud coseno o la distancia euclidiana. Al combinar las dos bases de datos, se pueden aprovechar plenamente las fortalezas de ambos sistemas. Como puedes ver en el ejemplo anterior, los datos de series temporales de sensores pueden almacenarse en InfluxDB, mientras que los datos vectoriales pueden almacenarse en Milvus. Esta integración permite casos de uso avanzados como la detección de anomalías en condiciones de tráfico en tiempo real.
El código relacionado con Milvus presentado en este artículo puede consultarse a través de este cuaderno de Colab.
También puedes echar un vistazo a este artículo que explora métodos de preprocesamiento para transformar datos de series temporales en una incrustación adecuada para tareas de previsión.
Sigue leyendo
Vector Lakebase: End the AI Data Silo
Learn how Vector Lakebase unifies vector search, data lakes, and AI data operations so teams can serve RAG and agents without copy-and-sync pipelines.
Announcing VDBBench 1.0: Open-Source VectorDB Benchmarking with Your Real-World Production Workloads
Discover VDBBench 1.0, an open-source tool for benchmarking vector databases with real-world production data, streaming ingestion, and concurrent workloads.
Proactive Monitoring for Vector Database: Zilliz Cloud Integrates with Datadog
we're excited to announce Zilliz Cloud's integration with Datadog, enabling comprehensive monitoring and observability for your vectorDB deployments.