Améliorer l’analyse avec les séries temporelles et les bases de données vectorielles
L’analyse des séries temporelles joue un rôle crucial dans de nombreux domaines, en particulier dans les appareils de l’Internet des objets (IoT). Avec les données de séries temporelles, nous pouvons détecter des modèles et des tendances sur des périodes particulières, ce qui nous permet de prévoir et d’analyser de futurs événements dépendants du temps. Parmi les exemples courants de cas d’utilisation des séries temporelles figurent la prévision des températures météorologiques et des cours boursiers, ainsi que la surveillance des données de capteurs.
En tant que base de données de séries temporelles, InfluxDB fournit une solution pour stocker d’immenses quantités de données de séries temporelles. InfluxDB est hautement optimisé pour le stockage et l’interrogation de données dépendantes du temps à l’aide de techniques telles que les agrégations et le sous-échantillonnage. Cependant, s’appuyer uniquement sur des bases de données de séries temporelles peut s’avérer difficile, surtout si notre cas d’utilisation nous oblige à effectuer une recherche de similarité.
Lors d’une récente intervention au Zilliz Unstructured Data Meetup, Zoe Steinkamp, Developer Advocate chez InfluxDB, a présenté une approche consistant à combiner InfluxDB avec Milvus pour stocker, interroger et effectuer des recherches de similarité sur des cas d’utilisation dépendants du temps.
Dans cet article, nous explorerons ce sujet plus en détail et vous guiderons à travers un cas d’utilisation dans lequel nous stockerons des données de séries temporelles dans InfluxDB, interrogerons les données, les transformerons en embeddings vectoriels, stockerons les embeddings dans Milvus, puis effectuerons enfin une recherche de similarité avec Milvus. Alors, sans plus attendre, commençons.
Comprendre les données de séries temporelles
Les données de séries temporelles représentent des observations ordonnées chronologiquement, enregistrées à des intervalles spécifiques, tels qu’horaires, quotidiens, hebdomadaires ou mensuels.
Nous pouvons trouver des cas d’utilisation des séries temporelles dans notre vie quotidienne : des relevés de température horaires aux statistiques quotidiennes du trafic routier, en passant par les chiffres mensuels des ventes au détail. En analysant les données de séries temporelles, nous pouvons découvrir des modèles historiques et utiliser ces informations pour éclairer les décisions futures.
Lors de l’analyse de données de séries temporelles, nous observons souvent un ou plusieurs des modèles suivants :
Saisonnier : Fluctuations récurrentes à intervalles de temps fixes, influencées par des facteurs tels que les jours de la semaine ou les mois.
Tendance : Une augmentation ou une diminution régulière des données sur une période prolongée.
Cyclique : Fluctuations similaires aux modèles saisonniers, mais sans fréquence fixe.
En haut à gauche - saisonnier, en haut à droite - tendance décroissante, en bas à gauche - tendance croissante, en bas à droite - cyclique. Source.
Comprendre ces modèles est crucial pour une prévision efficace des séries temporelles. La prévision des séries temporelles est une technique analytique puissante qui prédit les valeurs futures de variables dépendantes du temps à partir de données historiques. Par exemple, si nous disposons de l’historique du cours d’une action sur les deux années précédentes et que nous voulons prédire le cours de l’action de demain, nous effectuons essentiellement une prévision de séries temporelles.
Des entreprises de différents secteurs ont appliqué le concept de prévision des séries temporelles dans leurs opérations quotidiennes à diverses fins :
Institutions financières : Prédire les cours boursiers, prévoir les fluctuations des devises, identifier des modèles inhabituels dans les dépenses des clients, ou développer des stratégies robustes de gestion des risques.
Secteur de la santé : Surveiller la propagation des maladies, suivre les signes vitaux des patients en temps réel, ou améliorer la prise en charge globale des patients et les résultats.
Secteur de la vente au détail : Prévoir les volumes de ventes, comprendre les comportements d’achat des clients, optimiser la gestion des stocks, affiner les stratégies de tarification et améliorer la rentabilité globale.
Internet des objets (IoT) : Les systèmes de maison intelligente utilisent les données de capteurs pour automatiser les tâches, tandis que les thermostats et les hubs domestiques échangent des informations afin d’optimiser la consommation d’énergie.
Fabrication : Réduire les temps d’arrêt des machines, mettre en œuvre des stratégies de maintenance prédictive et améliorer l’efficacité opérationnelle globale.
En raison de sa nature dépendante du temps, les données de séries temporelles sont généralement collectées en continu en temps réel. Par conséquent, le volume de données de séries temporelles peut augmenter rapidement, ce qui nécessite une solution évolutive et efficace pour le stockage. C’est là que les bases de données de séries temporelles entrent en jeu.
InfluxDB comme base de données de séries temporelles
Les bases de données de séries temporelles doivent posséder plusieurs caractéristiques clés par rapport aux bases de données relationnelles traditionnelles afin de faciliter le processus de stockage et de récupération de grandes quantités de données de séries temporelles. Comme mentionné précédemment, les données de séries temporelles sont souvent collectées en temps réel ; par conséquent, les données doivent être traitées et synchronisées sur tous les systèmes dès qu’elles atteignent la base de données.
Les bases de données de séries temporelles offrent de meilleures performances et une plus grande efficacité que les bases de données relationnelles pour les charges de travail basées sur le temps, en particulier lorsqu’il s’agit de traiter de grands volumes de données et des écritures à haute fréquence. De plus, une base de données de séries temporelles offre des solutions hautement optimisées pour des opérations telles que l’ingestion de données, les requêtes et la récupération sur des plages temporelles, ainsi que l’analyse et les agrégations basées sur le temps.
Fonctionnalités importantes dans une base de données de séries temporelles
InfluxDB est l’une des bases de données de séries temporelles open source que nous pouvons utiliser pour stocker des données de séries temporelles. Elle est écrite en Rust et est hautement optimisée pour les opérations d’extraction, transformation et chargement (ETL) dans les bases de données. Pour faciliter l’ingestion rapide des données et la synchronisation des horodatages, InfluxDB contient un service de temps intégré qui utilise le Network Time Protocol (NTP).
Les données de séries temporelles que nous stockons dans une base de données de séries temporelles se composent généralement de seulement trois colonnes : une balise, un champ et un horodatage. Une balise contient les métadonnées des données que nous mesurons, un champ contient les valeurs que nous analysons ensuite, et un horodatage indique quand les données ont été collectées.
Disons que nous stockons des données de mesure de capteurs toutes les 5 minutes. Les données dans notre série temporelle ressembleraient à quelque chose comme la visualisation ci-dessous :
Exemple de données stockées dans une base de données de séries temporelles
Maintenant que nous connaissons les fondamentaux des bases de données de séries temporelles, parlons des bases de données vectorielles !
Milvus comme base de données vectorielle
Comme son nom l’indique, une base de données vectorielle stocke des données (y compris des images, des textes, des documents, etc.) sous forme de vecteurs. Nous transformons nos données en vecteurs à l’aide de l’un des nombreux modèles d’apprentissage profond.
La dimension d’un vecteur dépend de la technique ou du modèle d’apprentissage profond que nous utilisons pour le générer. Par exemple, si nous utilisons un modèle appelé all-MiniLM-L6-v2, nous obtiendrons un vecteur de dimension 384. En revanche, si nous utilisons un modèle appelé all-mpnet-base-v2, nous obtiendrons un vecteur de dimension 768.
Comme vous le savez peut-être, un vecteur n’est pas seulement un objet mathématique ; il encapsule à la fois une magnitude et une direction, portant la signification sémantique du texte, de l’image ou du document qu’il représente. Cet aspect profond des vecteurs est ce qui leur permet d’être placés proches les uns des autres dans l’espace vectoriel, représentant des textes ou des images similaires.
Similarité sémantique entre des vecteurs dans un espace vectoriel
L’idée qu’un vecteur porte la signification sémantique des données nous permet de comparer la similarité de n’importe quelle paire de vecteurs au moyen de métriques comme la similarité cosinus ou la distance euclidienne. Si la similarité cosinus entre deux vecteurs est proche de un, alors les deux vecteurs sont très similaires, et inversement.
Le stockage et l’exécution de recherches de similarité peuvent être simples à mettre en œuvre si nous n’avons que quelques vecteurs. Cependant, dans les applications réelles, nous traitons généralement des centaines de milliers ou des millions de vecteurs, ce qui rend l’ensemble de l’opération plus coûteux à maintenir, tant en temps qu’en ressources de calcul. C’est là qu’une base de données vectorielle devient nécessaire.
Les bases de données vectorielles comme Milvus et Zilliz Cloud offrent des processus hautement évolutifs et efficaces pour stocker des millions de vecteurs. Elles sont également fortement optimisées pour des recherches de similarité vectorielle et une récupération des données plus rapides, grâce à leur grande variété de méthodes d’indexation avancées telles que IVF-Flat, HNSW et d’autres.
Flux de travail complet d’une opération de recherche vectorielle
Le flux de travail pour stocker et récupérer des données depuis Milvus ou toute autre base de données vectorielle est le suivant : tout d’abord, nous transformons nos données d’entrée, qui peuvent être des textes ou des images, en vecteurs à l’aide d’une technique ou d’un modèle de notre choix. Ensuite, nous ingérons ces vecteurs avec leurs métadonnées dans la base de données vectorielle et construisons un index à l’aide d’une méthode d’indexation de notre choix.
Pendant le processus de récupération, nous devons d’abord transformer notre requête en vecteur en utilisant la même technique ou le même modèle que celui utilisé pendant le processus d’ingestion. Ensuite, nous effectuons une opération de recherche vectorielle pour comparer ce vecteur de requête avec les vecteurs stockés dans notre base de données vectorielle. Enfin, les vecteurs les plus similaires dans la base de données nous sont renvoyés comme résultats.
Le cas d’utilisation combinant une base de données de séries temporelles et une base de données vectorielle
Étant donné que les bases de données de séries temporelles et les bases de données vectorielles sont fortement optimisées pour des cas d’utilisation différents, nous pouvons tirer parti des points forts des deux bases de données dans nos projets réels.
Par exemple, imaginons un scénario dans lequel nous développons un système pour analyser les conditions de circulation en temps réel dans une ville intelligente. Pour y parvenir, nous installerions des capteurs à de nombreux emplacements capables de suivre la vitesse des véhicules, le nombre de véhicules et d’autres métriques pertinentes. La vitesse moyenne des véhicules et le nombre de véhicules sur des intervalles de temps spécifiques pourraient être stockés en continu dans une base de données de séries temporelles.
En plus des capteurs, nous pourrions également installer des caméras pour capturer des photos ou des vidéos des conditions réelles de circulation. Dans ce cas, nous pouvons transformer les photos et les vidéos en vecteurs à l’aide d’un modèle de deep learning de notre choix et stocker ces vecteurs dans une base de données vectorielle. En combinant les données de séries temporelles issues de la base de données de séries temporelles et les vecteurs issus de la base de données vectorielle, nous sommes capables d’effectuer une détection d’anomalies sur les conditions de circulation.
Exemple de cas d’utilisation combinant InfluxDB et Milvus pour analyser les conditions de circulation
Dans cette section, nous allons implémenter un cas d’utilisation simple lié aux conditions de circulation avec l’aide d’InfluxDB et de Milvus. Plus précisément, nous stockerons les données de séries temporelles dans InfluxDB et les vecteurs correspondants dans Milvus.
Commençons par générer des données de séries temporelles fictives composées d’un horodatage, d’une vitesse moyenne des véhicules, d’un nombre de véhicules et d’un type d’anomalie. Le type d’anomalie comporte deux valeurs distinctes : "normal" et "accident".
Nous commencerons par générer 500 points de données "normaux" collectés toutes les 10 minutes.
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
Ensuite, générons 500 points de données "accident", également collectés toutes les 10 minutes. Comme vous pouvez l’imaginer, deux signes possibles d’un accident dans les rues sont un nombre de véhicules plus élevé et une vitesse moyenne plus lente en raison de la congestion du trafic.
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
Concaténons maintenant les deux dataframe, puis insérons les données concaténées dans InfluxDB avec une commande qui ressemble à ce qui suit :
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')
Et vous pouvez interroger les données dans InfluxDB avec la commande suivante.
query = "SELECT * FROM traffic_data WHERE time >= now() - INTERVAL '90 days'"
pd = client.query(query=query, mode="pandas")
Si vous souhaitez en savoir plus sur les différentes opérations plus en détail, veuillez consulter cette documentation InfluxDB Python Client Library.
Générons maintenant des données vectorielles. Comme illustré ci-dessus, nous pouvons utiliser des photos ou des vidéos des conditions de circulation à des intervalles de temps particuliers comme sources de données et les transformer en vecteurs à l’aide d’un modèle de deep learning. Cependant, comme nous n’avons pas de photos ni de vidéos, nous allons collecter la vitesse moyenne des véhicules sur des intervalles de temps particuliers dans une liste, les transformant essentiellement en vecteurs. Dans l’analyse des séries temporelles, cette méthode est appelée fenêtrage.
Dans l’exemple suivant, nous définirons la taille de fenêtre à 24, mais vous pouvez l’ajuster à la taille de votre choix. Cela signifie que pour chaque observation, nous collecterons les valeurs de vitesse moyenne des véhicules pour les 240 minutes suivantes, puisque chaque point de données est collecté toutes les 10 minutes. Par conséquent, nous obtiendrons un vecteur à 24 dimensions pour chaque observation.
Pour chaque observation, nous utiliserons une taille de pas de 10. Cela signifie que l’heure de début entre deux observations consécutives diffère de 100 minutes. De plus, nous devons effectuer une normalisation sur chaque élément vectoriel pour chaque observation afin de garantir que le vecteur est mis à l’échelle uniformément de 0 à 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)
Vecteurs DataFrame
Comme vous pouvez le voir ci-dessus, nous avons maintenant un dataframe avec l’heure de début, l’heure de fin, le vecteur et le type d’anomalie. Ensuite, nous pouvons ingérer directement ces données dans la base de données Milvus. Le moyen le plus simple de commencer avec Milvus est d’utiliser Milvus Lite, donc installons d’abord Milvus Lite, puis créons un schéma selon les colonnes de notre 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")
Maintenant que nous avons créé un schéma tel que défini ci-dessus, nous pouvons ingérer notre dataframe dans Milvus avec la commande suivante :
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)
Et voilà ! Nous pouvons maintenant effectuer une opération de recherche vectorielle.
Imaginons qu’une semaine plus tard, nous observions une diminution inhabituelle de la vitesse moyenne des véhicules dans une zone spécifique. Dans l’exemple suivant, nous générerons ces données manuellement à l’aide des fonctions que nous avons définies précédemment. Cependant, dans un scénario réel, vous liriez et interrogeriez très probablement ces données depuis InfluxDB en fonction de l’horodatage, puis utiliseriez l’horodatage obtenu pour trouver le vecteur correspondant dans 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)
Ensuite, nous transformons ces données avec les mêmes méthodes de fenêtrage et de normalisation que précédemment.
query_vector = vectorize_data(df_test).vectors.values
Et maintenant, nous pouvons effectuer une opération de recherche vectorielle sur le vecteur de requête avec la commande suivante :
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'}}
"""
Comme vous pouvez le constater, les trois observations les plus similaires à notre requête ont toutes le type d’anomalie « accident », ce qui confirme la faible vitesse moyenne enregistrée dans nos données de requête. Si vous examinez la visualisation vectorielle entre les données de requête et les trois observations les plus similaires, vous remarquerez que les plages de valeurs vectorielles sont similaires entre elles.
Comparaison entre le vecteur de requête et les trois vecteurs les plus similaires
Conclusion
Dans cet article, nous avons exploré la combinaison des bases de données de séries temporelles et des bases de données vectorielles, en nous concentrant sur le cas d’utilisation dans les applications de l’Internet des objets (IoT). Les bases de données de séries temporelles, telles qu’InfluxDB, sont très efficaces pour stocker et interroger des données chronologiques, ce qui est crucial pour des applications comme les prévisions météorologiques, l’analyse des marchés boursiers et la surveillance des capteurs. Cependant, elles présentent des limites lorsqu’il s’agit d’effectuer des recherches de similarité.
Pour surmonter cela, nous avons combiné des bases de données de séries temporelles avec des bases de données vectorielles comme Milvus, qui stockent les données sous forme vectorielle, permettant des recherches de similarité efficaces à l’aide de techniques comme la similarité cosinus ou la distance euclidienne. En combinant les deux bases de données, les forces des deux systèmes peuvent être pleinement exploitées. Comme vous pouvez le voir dans l’exemple ci-dessus, les données de séries temporelles provenant de capteurs peuvent être stockées dans InfluxDB, tandis que les données vectorielles peuvent être stockées dans Milvus. Cette intégration permet des cas d’utilisation avancés comme la détection d’anomalies dans les conditions de circulation en temps réel.
Le code lié à Milvus présenté dans cet article est accessible via ce notebook Colab.
Vous pouvez également consulter cet article qui explore les méthodes de prétraitement pour transformer les données de séries temporelles en un embedding adapté aux tâches de prévision.
Continuer à lire
What Is a Vector Lakebase?
A Vector Lakebase is a unified, lake-native data architecture for AI that combines vector-database-grade serving with open lake storage, reusable lake-level indexes, and a shared semantic layer.
My Wife Wanted Dior. I Spent $600 on Claude Code to Vibe-Code a 2M-Line Database Instead.
Write tests, not code reviews. How a test-first workflow with 6 parallel Claude Code sessions turns a 2M-line C++ codebase into a daily shipping pipeline.
Migrating from S3 Vectors to Zilliz Cloud: Unlocking the Power of Tiered Storage
Learn how Zilliz Cloud bridges cost and performance with tiered storage and enterprise-grade features, and how to migrate data from AWS S3 Vectors to Zilliz Cloud.