Aprimorando a análise com séries temporais e bancos de dados vetoriais
A análise de séries temporais desempenha um papel crucial em muitas áreas, particularmente em dispositivos de Internet das Coisas (IoT). Com dados de séries temporais, podemos detectar padrões e tendências ao longo de períodos específicos, permitindo-nos prever e analisar eventos futuros dependentes do tempo. Exemplos comuns de casos de uso de séries temporais incluem a previsão de temperaturas meteorológicas e preços de ações e o monitoramento de dados de sensores.
Como um banco de dados de séries temporais, InfluxDB fornece uma solução para armazenar grandes quantidades de dados de séries temporais. O InfluxDB é altamente otimizado para armazenar e consultar dados dependentes do tempo usando técnicas como agregações e downsampling. No entanto, depender apenas de bancos de dados de séries temporais pode ser desafiador, especialmente se nosso caso de uso exigir que realizemos uma busca por similaridade.
Em uma palestra recente no Zilliz Unstructured Data Meetup, Zoe Steinkamp, Developer Advocate na InfluxDB, discutiu uma abordagem para combinar o InfluxDB com o Milvus para armazenar, consultar e realizar buscas por similaridade em casos de uso dependentes do tempo.
Neste artigo, exploraremos este tópico com mais detalhes e guiaremos você por um caso de uso em que armazenaremos dados de séries temporais no InfluxDB, consultaremos os dados, os transformaremos em embeddings vetoriais, armazenaremos os embeddings no Milvus e, finalmente, realizaremos uma busca por similaridade com o Milvus. Então, sem mais delongas, vamos começar.
Entendendo Dados de Séries Temporais
Dados de séries temporais representam observações ordenadas cronologicamente registradas em intervalos específicos, como de hora em hora, diariamente, semanalmente ou mensalmente.
Podemos encontrar casos de uso de séries temporais em nossa vida cotidiana: desde leituras de temperatura de hora em hora, estatísticas diárias de tráfego rodoviário, até números mensais de vendas no varejo. Ao analisar dados de séries temporais, podemos descobrir padrões históricos e usar esses insights para orientar decisões futuras.
Ao analisar dados de séries temporais, frequentemente vemos um ou mais dos seguintes padrões:
Sazonal: Flutuações recorrentes em intervalos de tempo fixos, influenciadas por fatores como dias da semana ou meses.
Tendência: Um aumento ou diminuição constante nos dados ao longo de um período prolongado.
Cíclico: Flutuações semelhantes a padrões sazonais, mas sem uma frequência fixa.
Superior esquerdo - sazonal, superior direito - tendência decrescente, inferior esquerdo - tendência crescente, inferior direito - cíclico. Fonte.
Entender esses padrões é crucial para uma previsão eficaz de séries temporais. A previsão de séries temporais é uma poderosa técnica analítica que prevê valores futuros de variáveis dependentes do tempo com base em dados históricos. Por exemplo, se temos o histórico de preços de uma ação dos últimos dois anos e queremos prever o preço da ação de amanhã, estamos essencialmente realizando previsão de séries temporais.
Empresas de diferentes setores aplicaram o conceito de previsão de séries temporais em suas operações diárias para diversos fins:
Instituições Financeiras: Prever preços de ações, prever flutuações cambiais, identificar padrões incomuns nos gastos dos clientes ou desenvolver estratégias robustas de gerenciamento de risco.
Setor de Saúde: Monitorar a disseminação de doenças, acompanhar os sinais vitais dos pacientes em tempo real ou melhorar o atendimento e os resultados gerais dos pacientes.
Setor de Varejo: Prever volumes de vendas, entender comportamentos de compra dos clientes, otimizar o gerenciamento de estoque, ajustar estratégias de preços e aumentar a lucratividade geral.
Internet das Coisas (IoT): Sistemas de casa inteligente usam dados de sensores para automatizar tarefas, enquanto termostatos e hubs domésticos trocam informações para otimizar o uso de energia.
Manufatura: Reduzir o tempo de inatividade das máquinas, implementar estratégias de manutenção preditiva e melhorar a eficiência operacional geral.
Devido à sua natureza dependente do tempo, os dados de séries temporais geralmente são coletados continuamente em tempo real. Como resultado, o volume de dados de séries temporais pode crescer rapidamente, exigindo uma solução escalável e eficiente para armazenamento. É aqui que os bancos de dados de séries temporais entram em cena.
InfluxDB como um Banco de Dados de Séries Temporais
Bancos de dados de séries temporais precisam possuir várias características-chave em comparação com bancos de dados relacionais tradicionais para facilitar o processo de armazenamento e recuperação de grandes quantidades de dados de séries temporais. Como mencionado anteriormente, os dados de séries temporais são frequentemente coletados em tempo real; portanto, os dados precisam ser processados e sincronizados em todos os sistemas assim que chegam ao banco de dados.
Bancos de dados de séries temporais oferecem melhor desempenho e eficiência do que bancos de dados relacionais para cargas de trabalho baseadas em tempo, especialmente ao lidar com grandes volumes de dados e gravações de alta frequência. Além disso, um banco de dados de séries temporais oferece soluções altamente otimizadas para operações como ingestão de dados, consultas e recuperação em intervalos de tempo, bem como análises e agregações baseadas em tempo.
Recursos importantes em um banco de dados de séries temporais
InfluxDB é um dos bancos de dados de séries temporais de código aberto que podemos usar para armazenar dados de séries temporais. Ele é escrito em Rust e é altamente otimizado para operações de extrair, transformar, carregar (ETL) em bancos de dados. Para facilitar a ingestão rápida de dados e a sincronização de carimbos de data/hora, o InfluxDB contém um serviço de tempo integrado que usa o Network Time Protocol (NTP).
Os dados de séries temporais que armazenamos dentro de um banco de dados de séries temporais normalmente consistem em apenas três colunas: uma tag, um campo e um carimbo de data/hora. Uma tag contém metadados dos dados que medimos, um campo contém valores que analisamos posteriormente, e um carimbo de data/hora indica quando os dados foram coletados.
Digamos que estamos armazenando dados de medição de sensores a cada 5 minutos. Os dados dentro da nossa série temporal se pareceriam com a visualização abaixo:
Exemplo de dados armazenados dentro de um banco de dados de séries temporais
Agora que conhecemos os fundamentos dos bancos de dados de séries temporais, vamos falar sobre bancos de dados vetoriais!
Milvus como um Banco de Dados Vetorial
Como o nome sugere, um banco de dados vetorial armazena dados (incluindo imagens, textos, documentos etc.) em vetores. Transformamos nossos dados em vetores usando um dos muitos modelos de deep learning.
A dimensão de um vetor depende da técnica ou do modelo de deep learning que usamos para gerá-lo. Por exemplo, se usarmos um modelo chamado all-MiniLM-L6-v2, obteremos um vetor de dimensão 384. Enquanto isso, se usarmos um modelo chamado all-mpnet-base-v2, obteremos um vetor de dimensão 768.
Como você deve saber, um vetor não é apenas um objeto matemático; ele encapsula tanto magnitude quanto direção, carregando o significado semântico do texto, imagem ou documento que representa. Esse aspecto profundo dos vetores é o que permite que eles sejam colocados próximos uns dos outros no espaço vetorial, representando textos ou imagens semelhantes.
Similaridade semântica entre vetores em um espaço vetorial
A ideia de que um vetor carrega o significado semântico dos dados nos permite comparar a similaridade de qualquer par de vetores por meio de métricas como similaridade de cosseno ou distância euclidiana. Se a similaridade de cosseno entre dois vetores estiver próxima de um, então os dois vetores são altamente semelhantes, e vice-versa.
Armazenar e realizar buscas por similaridade pode ser simples de implementar se tivermos apenas alguns vetores. No entanto, em aplicações reais, normalmente lidamos com centenas de milhares ou milhões de vetores, o que torna toda a operação mais cara de manter em termos de tempo e recursos computacionais. É aqui que um banco de dados vetorial se torna necessário.
Bancos de dados vetoriais como Milvus e Zilliz Cloud oferecem processos altamente escaláveis e eficientes para armazenar milhões de vetores. Eles também são altamente otimizados para buscas de similaridade vetorial e recuperação de dados mais rápidas, graças à sua ampla variedade de métodos avançados de indexação, como IVF-Flat, HNSW e outros.
Fluxo de trabalho completo de uma operação de busca vetorial
O fluxo de trabalho para armazenar e recuperar dados do Milvus ou de qualquer outro banco de dados vetorial é o seguinte: primeiro, transformamos nossos dados de entrada, que podem ser textos ou imagens, em vetores usando uma técnica ou modelo de nossa escolha. Em seguida, ingerimos esses vetores juntamente com seus metadados no banco de dados vetorial e construímos um índice usando um método de indexação de nossa escolha.
Durante o processo de recuperação, primeiro precisamos transformar nossa consulta em um vetor usando a mesma técnica ou modelo que usamos durante o processo de ingestão. Em seguida, realizamos uma operação de busca vetorial para comparar esse vetor de consulta com os vetores armazenados em nosso banco de dados vetorial. Por fim, os vetores mais semelhantes no banco de dados nos são retornados como resultados.
O caso de uso de combinar banco de dados de séries temporais e banco de dados vetorial
Como bancos de dados de séries temporais e vetoriais são altamente otimizados para diferentes casos de uso, podemos aproveitar os pontos fortes de ambos os bancos de dados em nossos projetos reais.
Por exemplo, vamos imaginar um cenário em que estamos desenvolvendo um sistema para analisar condições de tráfego em tempo real em uma cidade inteligente. Para isso, instalaríamos sensores em muitos locais que podem rastrear a velocidade dos veículos, a contagem de veículos e outras métricas relevantes. A velocidade média dos veículos e a contagem de veículos em intervalos de tempo específicos poderiam ser armazenadas continuamente dentro de um banco de dados de séries temporais.
Além dos sensores, também poderíamos instalar câmeras para capturar fotos ou vídeos das condições reais de tráfego. Nesse caso, podemos transformar as fotos e os vídeos em vetores usando um modelo de deep learning de nossa escolha e armazenar esses vetores em um banco de dados vetorial. Ao combinar os dados de séries temporais do banco de dados de séries temporais e os vetores do banco de dados vetorial, conseguimos realizar detecção de anomalias nas condições de tráfego.
Exemplo de caso de uso combinando InfluxDB e Milvus para analisar condições de tráfego
Nesta seção, vamos implementar um caso de uso simples relacionado às condições de tráfego com a ajuda do InfluxDB e do Milvus. Especificamente, armazenaremos dados de séries temporais dentro do InfluxDB e os vetores correspondentes dentro do Milvus.
Vamos primeiro gerar dados fictícios de séries temporais que consistem em timestamp, velocidade média dos veículos, contagem de veículos e tipo de anomalia. O tipo de anomalia consiste em dois valores distintos: "normal" e "accident".
Começaremos gerando 500 pontos de dados "normal" coletados a 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
Em seguida, vamos gerar 500 pontos de dados de "accident", também coletados a cada 10 minutos. Como você pode imaginar, dois possíveis sinais de um acidente nas ruas são uma contagem maior de veículos e uma velocidade média mais lenta devido ao congestionamento do tráfego.
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
Agora vamos concatenar os dois dataframes e então inserir os dados concatenados no InfluxDB com um comando parecido com o seguinte:
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 você pode fazer uma consulta aos dados dentro do InfluxDB com o seguinte comando.
query = "SELECT * FROM traffic_data WHERE time >= now() - INTERVAL '90 days'"
pd = client.query(query=query, mode="pandas")
Se quiser saber mais sobre diferentes operações em mais detalhes, consulte esta documentação da InfluxDB Python Client Library.
Agora vamos gerar alguns dados vetoriais. Como ilustrado acima, podemos usar fotos ou vídeos das condições do tráfego em intervalos de tempo específicos como fontes de dados e transformá-los em vetores usando um modelo de deep learning. No entanto, como não temos fotos ou vídeos, vamos coletar a velocidade média dos veículos em intervalos de tempo específicos em uma lista, essencialmente transformando-os em vetores. Na análise de séries temporais, esse método é chamado de janelamento.
No exemplo a seguir, definiremos o tamanho da janela como 24, mas você pode ajustá-lo para qualquer tamanho que preferir. Isso significa que, para cada observação, coletaremos os valores da velocidade média dos veículos dos próximos 240 minutos, já que cada ponto de dados é coletado a cada 10 minutos. Como resultado, teremos um vetor de 24 dimensões para cada observação.
Para cada observação, usaremos um tamanho de passo de 10. Isso significa que o horário de início entre observações consecutivas difere em 100 minutos. Além disso, precisamos realizar a normalização em cada elemento do vetor para cada observação, a fim de garantir que o vetor seja escalonado 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)
Vetores do DataFrame
Como você pode ver acima, o que temos agora é um dataframe com horário de início, horário de término, vetor e tipo de anomalia. Em seguida, podemos ingerir diretamente esses dados no banco de dados Milvus. A maneira mais fácil de começar com o Milvus é por meio do Milvus Lite, então vamos primeiro instalar o Milvus Lite e depois criar um esquema de acordo com as colunas dentro do nosso 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")
Agora que criamos um esquema conforme definido acima, podemos ingerir nosso dataframe no Milvus com o seguinte 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)
E é isso! Agora podemos realizar uma operação de busca vetorial.
Vamos imaginar que, uma semana depois, observamos uma diminuição incomum na velocidade média dos veículos em uma área específica. No exemplo a seguir, geraremos esses dados manualmente usando funções que definimos anteriormente. No entanto, em um cenário real, você provavelmente leria e consultaria esses dados no InfluxDB com base no timestamp e usaria o timestamp resultante para encontrar o vetor correspondente dentro do 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)
Em seguida, transformamos esses dados com os mesmos métodos de janelamento e normalização de antes.
query_vector = vectorize_data(df_test).vectors.values
E agora podemos realizar uma operação de busca vetorial no vetor de consulta com o seguinte 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 você pode ver, as três observações mais semelhantes à nossa consulta têm todas o tipo de anomalia "accident", o que confirma a baixa velocidade média registrada nos dados da nossa consulta. Se você examinar a visualização vetorial entre os dados da consulta e as três observações mais semelhantes, perceberá que o intervalo dos valores vetoriais é semelhante entre si.
Comparação entre o vetor da consulta e os três vetores mais semelhantes
Conclusão
Neste artigo, exploramos a combinação de bancos de dados de séries temporais e bancos de dados vetoriais, com foco no caso de uso em aplicações de Internet das Coisas (IoT). Bancos de dados de séries temporais, como o InfluxDB, são altamente eficientes para armazenar e consultar dados cronológicos, o que é crucial para aplicações como previsão do tempo, análise do mercado de ações e monitoramento de sensores. No entanto, eles enfrentam limitações quando se trata de realizar buscas por similaridade.
Para superar isso, combinamos bancos de dados de séries temporais com bancos de dados vetoriais como o Milvus, que armazenam dados em forma vetorial, permitindo buscas por similaridade eficientes usando técnicas como similaridade de cosseno ou distância euclidiana. Ao combinar os dois bancos de dados, os pontos fortes de ambos os sistemas podem ser plenamente aproveitados. Como você pode ver no exemplo acima, dados de séries temporais de sensores podem ser armazenados no InfluxDB, enquanto dados vetoriais podem ser armazenados no Milvus. Essa integração permite casos de uso avançados, como detecção de anomalias em condições de tráfego em tempo real.
O código relacionado ao Milvus apresentado neste artigo pode ser acessado por meio de este notebook Colab.
Você também pode dar uma olhada neste artigo que explora métodos de pré-processamento para transformar dados de séries temporais em um embedding adequado para tarefas de previsão.
Continue lendo
Introducing Functions and Model Inference on Zilliz Cloud: Automatic Embedding and Reranking with Hosted Models
Zilliz Cloud Functions auto-generate embeddings via OpenAI, Voyage AI, Cohere, or Zilliz Hosted Models. Built-in reranking — just insert text and search.
Vector Databases vs. Graph Databases
Use a vector database for AI-powered similarity search; use a graph database for complex relationship-based queries and network analysis.
AI Integration in Video Surveillance Tools: Transforming the Industry with Vector Databases
Discover how AI and vector databases are revolutionizing video surveillance with real-time analysis, faster threat detection, and intelligent search capabilities for enhanced security.