Migliorare l'analisi con database di serie temporali e vettoriali
L'analisi delle serie temporali svolge un ruolo cruciale in molti campi, in particolare nei dispositivi Internet of Things (IoT). Con i dati delle serie temporali, possiamo rilevare pattern e tendenze in periodi specifici, consentendoci di prevedere e analizzare eventi futuri dipendenti dal tempo. Esempi comuni di casi d'uso delle serie temporali includono la previsione delle temperature meteorologiche e dei prezzi delle azioni e il monitoraggio dei dati dei sensori.
Come database per serie temporali, InfluxDB fornisce una soluzione per archiviare enormi quantità di dati di serie temporali. InfluxDB è altamente ottimizzato per l'archiviazione e l'interrogazione di dati dipendenti dal tempo utilizzando tecniche come aggregazioni e downsampling. Tuttavia, affidarsi esclusivamente ai database di serie temporali può essere impegnativo, soprattutto se il nostro caso d'uso ci richiede di eseguire una ricerca di similarità.
In un recente intervento al Zilliz Unstructured Data Meetup, Zoe Steinkamp, Developer Advocate presso InfluxDB, ha discusso un approccio per combinare InfluxDB con Milvus per archiviare, interrogare ed eseguire ricerche di similarità su casi d'uso dipendenti dal tempo.
In questo articolo, esploreremo questo argomento in maggiore dettaglio e ti guideremo attraverso un caso d'uso in cui archivieremo dati di serie temporali in InfluxDB, interrogheremo i dati, li trasformeremo in embedding vettoriali, archivieremo gli embedding in Milvus e infine eseguiremo una ricerca di similarità con Milvus. Quindi, senza ulteriori indugi, iniziamo.
Comprendere i dati delle serie temporali
I dati delle serie temporali rappresentano osservazioni ordinate cronologicamente registrate a intervalli specifici, come orari, giornalieri, settimanali o mensili.
Possiamo trovare casi d'uso delle serie temporali nella nostra vita quotidiana: dalle letture orarie della temperatura, alle statistiche giornaliere del traffico stradale, fino ai dati mensili delle vendite al dettaglio. Analizzando i dati delle serie temporali, possiamo scoprire pattern storici e utilizzare queste informazioni per orientare le decisioni future.
Quando analizziamo i dati delle serie temporali, spesso osserviamo uno o più dei seguenti pattern:
Stagionale: Fluttuazioni ricorrenti a intervalli di tempo fissi, influenzate da fattori come i giorni della settimana o i mesi.
Tendenza: Un aumento o una diminuzione costante dei dati in un periodo prolungato.
Ciclico: Fluttuazioni simili ai pattern stagionali, ma senza una frequenza fissa.
In alto a sinistra - stagionale, in alto a destra - tendenza decrescente, in basso a sinistra - tendenza crescente, in basso a destra - ciclico. Fonte.
Comprendere questi pattern è fondamentale per un'efficace previsione delle serie temporali. La previsione delle serie temporali è una potente tecnica analitica che predice i valori futuri di variabili dipendenti dal tempo sulla base dei dati storici. Ad esempio, se abbiamo la cronologia dei prezzi di un'azione per i due anni precedenti e vogliamo prevedere il prezzo dell'azione di domani, stiamo essenzialmente eseguendo una previsione delle serie temporali.
Aziende di diversi settori hanno applicato il concetto di previsione delle serie temporali nelle loro operazioni quotidiane per vari scopi:
Istituzioni finanziarie: Prevedere i prezzi delle azioni, prevedere le fluttuazioni valutarie, identificare pattern insoliti nella spesa dei clienti o sviluppare solide strategie di gestione del rischio.
Settore sanitario: Monitorare la diffusione delle malattie, tracciare in tempo reale i segni vitali dei pazienti o migliorare l'assistenza complessiva ai pazienti e gli esiti.
Settore retail: Prevedere i volumi di vendita, comprendere i comportamenti d'acquisto dei clienti, ottimizzare la gestione dell'inventario, perfezionare le strategie di prezzo e aumentare la redditività complessiva.
Internet of Things (IoT): I sistemi smart home utilizzano i dati dei sensori per automatizzare le attività, mentre termostati e hub domestici scambiano informazioni per ottimizzare l'uso dell'energia.
Manufacturing: Ridurre i tempi di inattività delle macchine, implementare strategie di manutenzione predittiva e migliorare l'efficienza operativa complessiva.
A causa della sua natura dipendente dal tempo, i dati di serie temporali vengono in genere raccolti continuamente in tempo reale. Di conseguenza, il volume dei dati di serie temporali può crescere rapidamente, rendendo necessaria una soluzione scalabile ed efficiente per l’archiviazione. È qui che entrano in gioco i database di serie temporali.
InfluxDB come database di serie temporali
I database di serie temporali devono possedere diverse caratteristiche chiave rispetto ai database relazionali tradizionali per facilitare il processo di archiviazione e recupero di grandi quantità di dati di serie temporali. Come menzionato in precedenza, i dati di serie temporali vengono spesso raccolti in tempo reale; pertanto, i dati devono essere elaborati e sincronizzati in tutti i sistemi non appena raggiungono il database.
I database di serie temporali offrono prestazioni ed efficienza migliori rispetto ai database relazionali per carichi di lavoro basati sul tempo, soprattutto quando si gestiscono grandi volumi di dati e scritture ad alta frequenza. Inoltre, un database di serie temporali offre soluzioni altamente ottimizzate per operazioni come l’ingestione dei dati, l’interrogazione e il recupero su intervalli temporali, nonché analisi e aggregazioni basate sul tempo.
Funzionalità importanti in un database di serie temporali
InfluxDB è uno dei database di serie temporali open-source che possiamo utilizzare per archiviare dati di serie temporali. È scritto in Rust ed è altamente ottimizzato per operazioni di extract, transform, load (ETL) nei database. Per facilitare l’ingestione rapida dei dati e la sincronizzazione dei timestamp, InfluxDB contiene un servizio temporale integrato che utilizza il Network Time Protocol (NTP).
I dati di serie temporali che archiviamo all’interno di un database di serie temporali consistono tipicamente di sole tre colonne: un tag, un campo e un timestamp. Un tag contiene i metadati dei dati che misuriamo, un campo contiene i valori che analizziamo ulteriormente e un timestamp indica quando i dati sono stati raccolti.
Supponiamo di archiviare dati di misurazione dei sensori ogni 5 minuti. I dati all’interno della nostra serie temporale apparirebbero più o meno come nella visualizzazione seguente:
Esempio di dati archiviati all’interno di un database di serie temporali
Ora che conosciamo i fondamenti dei database di serie temporali, parliamo dei database vettoriali!
Milvus come database vettoriale
Come suggerisce il nome, un database vettoriale archivia i dati (incluse immagini, testi, documenti, ecc.) in vettori. Trasformiamo i nostri dati in vettori utilizzando uno dei molti modelli di deep learning.
La dimensione di un vettore dipende dalla tecnica o dal modello di deep learning che utilizziamo per generarlo. Ad esempio, se utilizziamo un modello chiamato all-MiniLM-L6-v2, otterremo un vettore di dimensione 384. Nel frattempo, se utilizziamo un modello chiamato all-mpnet-base-v2, otterremo un vettore di dimensione 768.
Come forse saprai, un vettore non è solo un oggetto matematico; incapsula sia grandezza sia direzione, portando con sé il significato semantico del testo, dell’immagine o del documento che rappresenta. Questo aspetto profondo dei vettori è ciò che consente loro di essere collocati vicini l’uno all’altro nello spazio vettoriale, rappresentando testi o immagini simili.
Somiglianza semantica tra vettori in uno spazio vettoriale
L’idea che un vettore porti con sé il significato semantico dei dati ci consente di confrontare la somiglianza di qualsiasi coppia di vettori tramite metriche come la similarità del coseno o la distanza euclidea. Se la similarità del coseno tra due vettori è vicina a uno, allora i due vettori sono molto simili, e viceversa.
Archiviare ed eseguire ricerche di similarità potrebbe essere semplice da implementare se avessimo solo pochi vettori. Tuttavia, nelle applicazioni reali, in genere abbiamo a che fare con centinaia di migliaia o milioni di vettori, il che rende l’intera operazione più costosa da mantenere in termini sia di tempo sia di risorse computazionali. È qui che un database vettoriale diventa necessario.
I database vettoriali come Milvus e Zilliz Cloud offrono processi altamente scalabili ed efficienti per archiviare milioni di vettori. Sono inoltre altamente ottimizzati per ricerche di similarità vettoriale e recupero dei dati più rapidi, grazie alla loro ampia varietà di metodi di indicizzazione avanzati come IVF-Flat, HNSW e altri.
Flusso di lavoro completo di un’operazione di ricerca vettoriale
Il flusso di lavoro per archiviare e recuperare dati da Milvus o da qualsiasi altro database vettoriale è il seguente: per prima cosa, trasformiamo i nostri dati di input, che potrebbero essere testi o immagini, in vettori usando una tecnica o un modello a nostra scelta. Successivamente, inseriamo questi vettori insieme ai loro metadati nel database vettoriale e costruiamo un indice usando un metodo di indicizzazione a nostra scelta.
Durante il processo di recupero, dobbiamo prima trasformare la nostra query in un vettore usando la stessa tecnica o lo stesso modello utilizzato durante il processo di inserimento. Successivamente, eseguiamo un’operazione di ricerca vettoriale per confrontare questo vettore di query con i vettori archiviati nel nostro database vettoriale. Infine, i vettori più simili nel database ci vengono restituiti come risultati.
Il caso d’uso della combinazione di database di serie temporali e database vettoriale
Poiché i database di serie temporali e i database vettoriali sono altamente ottimizzati per casi d’uso diversi, possiamo sfruttare i punti di forza di entrambi i database nei nostri progetti reali.
Per esempio, immaginiamo uno scenario in cui stiamo sviluppando un sistema per analizzare le condizioni del traffico in tempo reale in una smart city. Per ottenere questo risultato, installeremmo sensori in molte posizioni in grado di tracciare la velocità dei veicoli, il numero di veicoli e altre metriche rilevanti. La velocità media dei veicoli e il numero di veicoli su intervalli di tempo specifici potrebbero essere archiviati continuamente all’interno di un database di serie temporali.
Oltre ai sensori, potremmo anche installare telecamere per acquisire foto o video delle condizioni effettive del traffico. In questo caso, possiamo trasformare le foto e i video in vettori usando un modello di deep learning a nostra scelta e archiviare questi vettori in un database vettoriale. Combinando i dati di serie temporali provenienti dal database di serie temporali e i vettori provenienti dal database vettoriale, siamo in grado di eseguire il rilevamento di anomalie sulle condizioni del traffico.
Esempio di caso d’uso della combinazione di InfluxDB e Milvus per analizzare le condizioni del traffico
In questa sezione, implementeremo un semplice caso d’uso relativo alle condizioni del traffico con l’aiuto di InfluxDB e Milvus. Nello specifico, archivieremo i dati di serie temporali all’interno di InfluxDB e i vettori corrispondenti all’interno di Milvus.
Per prima cosa generiamo dati fittizi di serie temporali composti da timestamp, velocità media dei veicoli, numero di veicoli e tipo di anomalia. Il tipo di anomalia consiste in due valori distinti: "normal" e "accident".
Inizieremo generando 500 punti dati "normal" raccolti ogni 10 minuti.
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,
'Conteggio veicoli': vehicle_counts,
'Velocità media': avg_speeds,
'Tipo di anomalia': 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 normale
Successivamente, generiamo 500 punti dati "accident", anch’essi raccolti ogni 10 minuti. Come puoi immaginare, due possibili segnali di un incidente sulle strade sono un conteggio dei veicoli più elevato e una velocità media più lenta a causa della congestione del traffico.
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 incidente
Ora concateniamo i due dataframe e poi inseriamo i dati concatenati in InfluxDB con un comando simile al seguente:
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')
E puoi eseguire una query sui dati all’interno di InfluxDB con il seguente comando.
query = "SELECT * FROM traffic_data WHERE time >= now() - INTERVAL '90 days'"
pd = client.query(query=query, mode="pandas")
Se desideri saperne di più sulle diverse operazioni in modo più dettagliato, consulta questa documentazione InfluxDB Python Client Library.
Ora generiamo alcuni dati vettoriali. Come illustrato sopra, possiamo usare foto o video delle condizioni del traffico a intervalli di tempo specifici come fonti di dati e trasformarli in vettori utilizzando un modello di deep learning. Tuttavia, poiché non disponiamo di foto o video, raccoglieremo la velocità media dei veicoli su intervalli di tempo specifici in una lista, trasformandoli essenzialmente in vettori. Nell’analisi delle serie temporali, questo metodo si chiama windowing.
Nel seguente esempio, imposteremo la dimensione della finestra a 24, ma puoi modificarla a qualsiasi dimensione tu preferisca. Ciò significa che per ogni osservazione raccoglieremo i valori della velocità media dei veicoli per i successivi 240 minuti, poiché ogni punto dati viene raccolto ogni 10 minuti. Di conseguenza, otterremo un vettore a 24 dimensioni per ogni osservazione.
Per ogni osservazione, useremo una dimensione del passo pari a 10. Ciò significa che l’ora di inizio tra osservazioni consecutive differisce di 100 minuti. Inoltre, dobbiamo eseguire la normalizzazione su ogni elemento del vettore per ciascuna osservazione, per garantire che il vettore sia scalato uniformemente da 0 a 1.
window_size = 24
step_size = 10
min_val = 10 # Velocità media minima dei veicoli
max_val = 80 # Velocità media massima dei veicoli
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]
# Crea un nuovo DataFrame dai dati raccolti
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)
# Applica una funzione lambda per convertire i timestamp nel formato 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)
Vettori DataFrame
Come puoi vedere sopra, ciò che abbiamo ora è un dataframe con ora di inizio, ora di fine, vettore e tipo di anomalia. Successivamente, possiamo inserire direttamente questi dati nel database Milvus. Il modo più semplice per iniziare con Milvus è tramite Milvus Lite, quindi installiamo prima Milvus Lite e poi creiamo uno schema in base alle colonne all'interno del nostro 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")
Ora che abbiamo creato uno schema come definito sopra, possiamo inserire il nostro dataframe in Milvus con il seguente 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)
Ed è tutto! Ora possiamo eseguire un'operazione di ricerca vettoriale.
Immaginiamo che una settimana dopo osserviamo una diminuzione insolita della velocità media dei veicoli in un'area specifica. Nell'esempio seguente, genereremo questi dati manualmente utilizzando funzioni che abbiamo definito in precedenza. Tuttavia, in uno scenario reale, molto probabilmente leggeresti e interrogheresti questi dati da InfluxDB in base al timestamp, e useresti il timestamp risultante per trovare il vettore corrispondente all'interno di 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)
Successivamente, trasformiamo questi dati con gli stessi metodi di windowing e normalizzazione di prima.
query_vector = vectorize_data(df_test).vectors.values
E ora possiamo eseguire un'operazione di ricerca vettoriale sul vettore di query con il seguente 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'}}
"""
Come puoi vedere, le tre osservazioni più simili alla nostra query hanno tutte il tipo di anomalia "accident", il che conferma la bassa velocità media registrata nei dati della nostra query. Se esamini la visualizzazione vettoriale tra i dati della query e le tre osservazioni più simili, noterai che l'intervallo dei valori vettoriali è simile tra loro.
Confronto tra il vettore della query e i tre vettori più simili
Conclusione
In questo articolo, abbiamo esplorato la combinazione di database di serie temporali e database vettoriali, concentrandoci sul caso d'uso nelle applicazioni Internet of Things (IoT). I database di serie temporali, come InfluxDB, sono altamente efficienti nell'archiviazione e nell'interrogazione di dati cronologici, il che è cruciale per applicazioni come le previsioni meteorologiche, l'analisi del mercato azionario e il monitoraggio dei sensori. Tuttavia, presentano limitazioni quando si tratta di eseguire ricerche di similarità.
Per superare questo limite, abbiamo combinato i database di serie temporali con database vettoriali come Milvus, che archiviano i dati in forma vettoriale, consentendo ricerche di similarità efficienti tramite tecniche come la similarità coseno o la distanza euclidea. Combinando i due database, i punti di forza di entrambi i sistemi possono essere sfruttati appieno. Come puoi vedere nell'esempio sopra, i dati di serie temporali provenienti dai sensori possono essere archiviati in InfluxDB, mentre i dati vettoriali possono essere archiviati in Milvus. Questa integrazione consente casi d'uso avanzati come il rilevamento di anomalie nelle condizioni del traffico in tempo reale.
Il codice relativo a Milvus presentato in questo articolo è accessibile tramite questo notebook Colab.
Puoi anche dare un'occhiata a questo articolo che esplora metodi di pre-elaborazione per trasformare i dati di serie temporali in un embedding adatto alle attività di previsione.
Continua a leggere
Data Deduplication at Trillion Scale: How to Solve the Biggest Bottleneck of LLM Training
Explore how MinHash LSH and Milvus handle data deduplication at the trillion-scale level, solving key bottlenecks in LLM training for improved AI model performance.
DeepRAG: Thinking to Retrieval Step by Step for Large Language Models
Discover DeepRAG, an advanced retrieval-augmented generation (RAG) model that improves LLM accuracy by retrieving only essential data through step-by-step reasoning.
Vector Databases vs. Hierarchical Databases
Use a vector database for AI-powered similarity search; use a hierarchical database for organizing data in parent-child relationships with efficient top-down access patterns.