시계열 및 벡터 데이터베이스로 분석 개선하기
시계열 분석은 많은 분야, 특히 사물 인터넷(IoT) 장치에서 중요한 역할을 합니다. 시계열 데이터를 사용하면 특정 기간 동안의 패턴과 추세를 감지할 수 있어, 미래의 시간 의존적 이벤트를 예측하고 분석할 수 있습니다. 시계열 사용 사례의 일반적인 예로는 날씨 기온과 주가 예측, 센서 데이터 모니터링이 있습니다.
시계열 데이터베이스인 InfluxDB는 방대한 양의 시계열 데이터를 저장하기 위한 솔루션을 제공합니다. InfluxDB는 집계 및 다운샘플링과 같은 기법을 사용하여 시간 의존적 데이터를 저장하고 쿼리하는 데 고도로 최적화되어 있습니다. 그러나 시계열 데이터베이스에만 의존하는 것은 어려울 수 있으며, 특히 사용 사례에서 유사도 검색을 수행해야 하는 경우 더욱 그렇습니다.
최근 Zilliz Unstructured Data Meetup에서 InfluxDB의 Developer Advocate인 Zoe Steinkamp는 시간 의존적 사용 사례에서 데이터를 저장하고, 쿼리하고, 유사도 검색을 수행하기 위해 InfluxDB와 Milvus를 결합하는 접근 방식에 대해 논의했습니다.
이 글에서는 이 주제를 더 자세히 살펴보고, 시계열 데이터를 InfluxDB에 저장하고, 데이터를 쿼리하며, 이를 벡터 임베딩으로 변환하고, 임베딩을 Milvus에 저장한 뒤, 마지막으로 Milvus로 유사도 검색을 수행하는 사용 사례를 단계별로 안내하겠습니다. 그럼 더 이상 지체하지 말고 시작해 보겠습니다.
시계열 데이터 이해하기
시계열 데이터는 시간별, 일별, 주별 또는 월별과 같이 특정 간격으로 기록된 시간순 관측값을 나타냅니다.
우리는 일상생활에서 시계열 사용 사례를 찾을 수 있습니다. 시간별 기온 측정값, 일별 도로 교통 통계, 월별 소매 판매 수치에 이르기까지 다양합니다. 시계열 데이터를 분석함으로써 과거의 패턴을 발견하고 이러한 인사이트를 활용해 미래의 의사결정에 반영할 수 있습니다.
시계열 데이터를 분석할 때, 우리는 종종 다음 패턴 중 하나 이상을 보게 됩니다:
계절성: 요일이나 월과 같은 요인의 영향을 받아 고정된 시간 간격으로 반복되는 변동.
추세: 장기간에 걸쳐 데이터가 꾸준히 증가하거나 감소하는 현상.
순환성: 계절성 패턴과 유사한 변동이지만, 고정된 빈도는 없음.
왼쪽 위 - 계절성, 오른쪽 위 - 감소 추세, 왼쪽 아래 - 증가 추세, 오른쪽 아래 - 순환성. 출처.
이러한 패턴을 이해하는 것은 효과적인 시계열 예측에 매우 중요합니다. 시계열 예측은 과거 데이터를 기반으로 시간 의존적 변수의 미래 값을 예측하는 강력한 분석 기법입니다. 예를 들어, 지난 2년간의 주가 기록이 있고 내일의 주가를 예측하고자 한다면, 우리는 본질적으로 시계열 예측을 수행하는 것입니다.
다양한 부문의 기업들은 여러 목적을 위해 일상 운영에 시계열 예측 개념을 적용해 왔습니다:
금융 기관: 주가 예측, 환율 변동 예측, 고객 지출의 비정상적인 패턴 식별, 또는 견고한 리스크 관리 전략 개발.
의료 부문: 질병 확산 모니터링, 환자의 활력 징후를 실시간으로 추적, 또는 전반적인 환자 치료와 결과 개선.
소매 산업: 판매량 예측, 고객 구매 행동 이해, 재고 관리 최적화, 가격 전략 미세 조정, 전반적인 수익성 향상.
사물 인터넷(IoT): 스마트 홈 시스템은 센서 데이터를 사용하여 작업을 자동화하고, 온도 조절기와 홈 허브는 에너지 사용을 최적화하기 위해 정보를 교환합니다.
제조: 기계 가동 중단 시간 감소, 예측 유지보수 전략 구현, 전반적인 운영 효율성 개선.
시간 의존적인 특성으로 인해 시계열 데이터는 일반적으로 실시간으로 지속적으로 수집됩니다. 그 결과, 시계열 데이터의 양은 빠르게 증가할 수 있으며, 저장을 위한 확장 가능하고 효율적인 솔루션이 필요합니다. 바로 여기서 시계열 데이터베이스가 등장합니다.
시계열 데이터베이스로서의 InfluxDB
시계열 데이터베이스는 대량의 시계열 데이터 저장 및 검색 과정을 용이하게 하기 위해 기존 관계형 데이터베이스와 비교해 몇 가지 핵심 특성을 갖추어야 합니다. 앞서 언급했듯이 시계열 데이터는 종종 실시간으로 수집되므로, 데이터베이스에 도달하는 즉시 모든 시스템에서 처리되고 동기화되어야 합니다.
시계열 데이터베이스는 시간 기반 워크로드에서 관계형 데이터베이스보다 더 나은 성능과 효율성을 제공하며, 특히 대량의 데이터와 고빈도 쓰기를 처리할 때 그렇습니다. 또한 시계열 데이터베이스는 데이터 수집, 쿼리, 시간 범위에 따른 검색뿐만 아니라 시간 기반 분석 및 집계와 같은 작업에 대해 고도로 최적화된 솔루션을 제공합니다.
시계열 데이터베이스의 중요한 기능
InfluxDB는 시계열 데이터를 저장하는 데 사용할 수 있는 오픈 소스 시계열 데이터베이스 중 하나입니다. Rust로 작성되었으며, 데이터베이스에서 추출, 변환, 로드(ETL) 작업에 고도로 최적화되어 있습니다. 빠른 데이터 수집과 타임스탬프 동기화를 지원하기 위해 InfluxDB에는 Network Time Protocol (NTP)을 사용하는 내장 시간 서비스가 포함되어 있습니다.
시계열 데이터베이스 내부에 저장하는 시계열 데이터는 일반적으로 태그, 필드, 타임스탬프라는 세 개의 열로만 구성됩니다. 태그에는 측정하는 데이터의 메타데이터가 포함되고, 필드에는 추가로 분석할 값이 포함되며, 타임스탬프는 데이터가 수집된 시점을 나타냅니다.
센서 측정 데이터를 5분마다 저장한다고 가정해 보겠습니다. 시계열 내부의 데이터는 아래 시각화와 비슷하게 보일 것입니다:
시계열 데이터베이스 내부에 저장된 데이터 예시
이제 시계열 데이터베이스의 기본 사항을 알았으니, 벡터 데이터베이스에 대해 이야기해 보겠습니다!
벡터 데이터베이스로서의 Milvus
이름에서 알 수 있듯이, 벡터 데이터베이스는 데이터(이미지, 텍스트, 문서 등 포함)를 벡터로 저장합니다. 우리는 다양한 딥러닝 모델 중 하나를 사용하여 데이터를 벡터로 변환합니다.
벡터의 차원은 이를 생성하는 데 사용하는 기법이나 딥러닝 모델에 따라 달라집니다. 예를 들어, all-MiniLM-L6-v2라는 모델을 사용하면 384차원의 벡터를 얻습니다. 한편, all-mpnet-base-v2라는 모델을 사용하면 768차원의 벡터를 얻습니다.
아시다시피, 벡터는 단순한 수학적 객체가 아닙니다. 벡터는 크기와 방향을 모두 캡슐화하며, 그것이 나타내는 텍스트, 이미지 또는 문서의 의미를 담고 있습니다. 벡터의 이러한 심오한 측면이 벡터 공간에서 서로 가까이 배치되도록 하여 유사한 텍스트나 이미지를 나타낼 수 있게 합니다.
벡터 공간에서 벡터 간의 의미적 유사성
벡터가 데이터의 의미를 담고 있다는 아이디어는 코사인 유사도나 유클리드 거리와 같은 지표를 통해 임의의 벡터 쌍의 유사성을 비교할 수 있게 해줍니다. 두 벡터 간의 코사인 유사도가 1에 가까우면 두 벡터는 매우 유사하며, 그 반대도 마찬가지입니다.
몇 개의 벡터만 있다면 저장과 유사도 검색 수행은 구현하기 간단할 수 있습니다. 하지만 실제 애플리케이션에서는 일반적으로 수십만 또는 수백만 개의 벡터를 다루게 되며, 이로 인해 전체 작업을 시간과 컴퓨팅 리소스 측면에서 유지 관리하는 비용이 더 커집니다. 바로 이 지점에서 벡터 데이터베이스가 필요해집니다.
Milvus와 Zilliz Cloud 같은 벡터 데이터베이스는 수백만 개의 벡터를 저장하기 위한 확장성과 효율성이 매우 높은 프로세스를 제공합니다. 또한 IVF-Flat, HNSW 등 다양한 고급 인덱싱 방법 덕분에 더 빠른 벡터 유사도 검색과 데이터 검색에 고도로 최적화되어 있습니다.
벡터 검색 작업의 전체 워크플로
Milvus 또는 다른 벡터 데이터베이스에서 데이터를 저장하고 검색하는 워크플로는 다음과 같습니다. 먼저 텍스트나 이미지일 수 있는 입력 데이터를 우리가 선택한 기법이나 모델을 사용해 벡터로 변환합니다. 다음으로 이러한 벡터를 해당 메타데이터와 함께 벡터 데이터베이스에 수집하고, 우리가 선택한 인덱싱 방법을 사용해 인덱스를 구축합니다.
검색 과정에서는 먼저 수집 과정에서 사용한 것과 동일한 기법이나 모델을 사용해 쿼리를 벡터로 변환해야 합니다. 다음으로 이 쿼리 벡터를 벡터 데이터베이스에 저장된 벡터들과 비교하기 위해 벡터 검색 작업을 수행합니다. 마지막으로 데이터베이스에서 가장 유사한 벡터들이 결과로 반환됩니다.
시계열 데이터베이스와 벡터 데이터베이스를 결합하는 사용 사례
시계열 데이터베이스와 벡터 데이터베이스는 서로 다른 사용 사례에 고도로 최적화되어 있으므로, 실제 프로젝트에서 두 데이터베이스의 강점을 모두 활용할 수 있습니다.
예를 들어, 스마트 시티에서 실시간 교통 상황을 분석하는 시스템을 개발하는 시나리오를 상상해 봅시다. 이를 달성하기 위해 차량 속도, 차량 수 및 기타 관련 지표를 추적할 수 있는 센서를 여러 위치에 설치할 수 있습니다. 특정 시간 간격 동안의 평균 차량 속도와 차량 수는 시계열 데이터베이스 안에 지속적으로 저장될 수 있습니다.
센서 외에도 실제 교통 상황의 사진이나 동영상을 캡처하기 위해 카메라를 설치할 수도 있습니다. 이 경우 우리가 선택한 딥러닝 모델을 사용해 사진과 동영상을 벡터로 변환하고, 이러한 벡터를 벡터 데이터베이스에 저장할 수 있습니다. 시계열 데이터베이스의 시계열 데이터와 벡터 데이터베이스의 벡터를 결합함으로써 교통 상황에 대한 이상 탐지를 수행할 수 있습니다.
InfluxDB와 Milvus를 결합해 교통 상황을 분석하는 사용 사례 예시
이 섹션에서는 InfluxDB와 Milvus의 도움을 받아 교통 상황과 관련된 간단한 사용 사례를 구현해 보겠습니다. 구체적으로, 시계열 데이터는 InfluxDB 안에 저장하고 해당 벡터는 Milvus 안에 저장할 것입니다.
먼저 타임스탬프, 평균 차량 속도, 차량 수, 이상 유형으로 구성된 더미 시계열 데이터를 생성해 보겠습니다. 이상 유형은 "normal"과 "accident"라는 두 가지 고유한 값으로 구성됩니다.
10분마다 수집된 500개의 "normal" 데이터 포인트를 생성하는 것부터 시작하겠습니다.
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
다음으로, 10분마다 수집된 500개의 "accident" 데이터 포인트를 생성해 보겠습니다. 짐작할 수 있듯이, 도로에서 사고가 발생했음을 나타낼 수 있는 두 가지 징후는 교통 혼잡으로 인해 차량 수가 증가하고 평균 속도가 느려지는 것입니다.
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로 설정하지만, 원하는 크기로 조정할 수 있습니다. 이는 각 관측치에 대해 각 데이터 포인트가 10분마다 수집되므로 다음 240분 동안의 평균 차량 속도 값을 수집한다는 의미입니다. 결과적으로 각 관측치에 대해 24차원 벡터를 얻게 됩니다.
각 관측치에 대해 step size는 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)
# 람다 함수를 적용하여 타임스탬프를 Unix 타임스탬프 형식으로 변환합니다.
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)
데이터프레임 벡터
위에서 볼 수 있듯이, 현재 لدينا는 시작 시간, 종료 시간, 벡터, 이상 유형이 포함된 데이터프레임입니다. 다음으로, 이 데이터를 Milvus 데이터베이스에 직접 수집할 수 있습니다. Milvus를 시작하는 가장 쉬운 방법은 Milvus Lite를 사용하는 것이므로, 먼저 Milvus Lite를 설치한 다음 데이터프레임 내부의 열에 따라 스키마를 생성해 보겠습니다.
! 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")
이제 위에 정의된 대로 스키마를 생성했으므로, 다음 명령으로 데이터프레임을 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 노트북을 통해 확인할 수 있습니다.
예측 작업에 적합한 임베딩으로 시계열 데이터를 변환하기 위한 전처리 방법을 살펴보는 이 글도 확인해 볼 수 있습니다.
계속 읽기
Notion's Vector Search Is Excellent. Their Next Problem Is Harder.
Notion solved vector search scaling in two years. The next bottleneck — offline context engineering, unified data, and the real-time/offline gap — is harder.
Introducing Business Critical Plan: Enterprise-Grade Security and Compliance for Mission-Critical AI Applications
Discover Zilliz Cloud’s Business Critical Plan—offering advanced security, compliance, and uptime for mission-critical AI and vector database workloads.
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.