제로부터 시작하는 MLOps 도구와 활용 - 4. 데이터 관리 (2/2)

‘제로부터 시작하는 MLOps 도구와 활용’ 시리즈 기사를 마지막으로 작성 했던 것이 거의 5개월 전이네요. 요즘은 MLOps 관련 업무를 하지 않아서, 솔직히 흥미가 많이 떨어졌습니다. 그래도 계획했던 내용들은 모두 마무리 짓고 싶습니다. 시간이 흘러 초기화되어 버린 내용들은 다시 예전 경험들을 곰곰이 떠올려보고, 최근 트렌드도 찾아보면서 ChatGPT와 함께 작성해 보겠습니다.

제로부터 시작하는 MLOps 도구와 활용 - 4. 데이터 관리 (1/2)에서 이어지는 내용 입니다.

4.2 데이터 버전 관리

4.2.1 데이터 버전 관리 개요

4.2.2 Git LFS (Git Large File Storage)

• macOS

$ brew install git-lfs

• Linux (Debian/Ubuntu)

$ sudo apt install git-lfs

$ git lfs install

결과

$ git lfs track "*.petastorm" "*.tfrecord" "*.csv" "*.parquet"
$ git add .gitattributes
$ git commit -m "Enable Git LFS for the dataset file"
$ cat .gitattributes

결과

 

$ git add datasets/train.petastorm
$ git add datasets/val.petastorm
$ git commit -m "Add new train/val dataset"

결과

4.2.3 DVC

• macOS

$ brew install dvc

• Linux (Debian/Ubuntu)

$ sudo wget \
       https://dvc.org/deb/dvc.list \
       -O /etc/apt/sources.list.d/dvc.list
$ wget -qO - https://dvc.org/deb/iterative.asc | gpg --dearmor > packages.iterative.gpg
$ sudo install -o root -g root -m 644 packages.iterative.gpg /etc/apt/trusted.gpg.d/
$ rm -f packages.iterative.gpg
$ sudo apt update
$ sudo apt install dvc

• pip

$ pip install dvc

$ dvc init

결과

$ git status

결과

$ git commit -m "Initialize DVC"

$ dvc get https://github.com/iterative/dataset-registry \
          get-started/data.xml -o data/data.xml

결과

$ dvc add data/data.xml

결과

$ git add data/data.xml.dvc data/.gitignore
$ git commit -m "Add raw data"

결과

$ mkdir /tmp/dvcstore
$ dvc remote add -d myremote /tmp/dvcstore

결과

$ dvc remote add -d storage s3://mybucket/dvcstore

$ dvc push

결과

$ dvc pull

결과

$ cp data/data.xml /tmp/data.xml
$ cat /tmp/data.xml >> data/data.xml

$ dvc add data/data.xml

결과

$ dvc push
$ git commit data/data.xml.dvc -m "Dataset updates"

결과

$ git checkout HEAD~1 data/data.xml.dvc
$ dvc checkout

결과

DVC에 대한 좀 더 많은 정보는 DVC 공식 문서를 참고하시기 바랍니다.

4.3 피처 스토어 (Feature Store)

4.3.1 피처 스토어 개요

4.3.2 Feast

• pip

$ pip install feast

$ feast init some_ml_project

결과

• data/ 샘플 데이터 parquet 파일이 포함되어 있습니다.

# 다음은 피처 정의 파일 예시입니다.
from datetime import timedelta

import pandas as pd

from feast import (
    Entity,
    FeatureService,
    FeatureView,
    Field,
    FileSource,
    PushSource,
    RequestSource,
)
from feast.on_demand_feature_view import on_demand_feature_view
from feast.types import Float32, Float64, Int64

# 드라이버의 엔티티를 정의합니다. 엔티티는 피처을 가져오는데 사용되는 기본키라고 생각하면 됩니다.
driver = Entity(name="driver", join_keys=["driver_id"])

# parquet 파일에서 데이터를 읽습니다. parquet 파일은 로컬 개발시 편리합니다.
# 프로덕션의 경우 BigQuery와 같은 데이터 웨어하우스를 사용할 수 있습니다. 자세한 내용은 Feast 문서를 참조하세요.
driver_stats_source = FileSource(
    name="driver_hourly_stats_source",
    path="/Users/taehun/Projects/some_ml_project/feature_repo/data/driver_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

# parquet 파일에는 driver_id 컬럼, timestamp 및 3개의 피처 컬럼이 포함된 샘플 데이터가 포함되어 있습니다.
# 여기에서는 이 데이터를 온라인으로 모델에 제공할 수 있는 FeatureView를 정의합니다.
driver_stats_fv = FeatureView(
    # 이 FeatureView의 고유 이름입니다. 단일 프로젝트에서 중복될 수 없습니다.
    name="driver_hourly_stats",
    entities=[driver],
    ttl=timedelta(days=1),
    # 아래에 정의된 피처 목록은 스토어에 피처를 구체화하는 스키마 역할을 하며,
    # 학습 데이터 세트 구축 또는 피처 서빙을 위한 검색시 참조로 사용됩니다.
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64, description="Average daily trips"),
    ],
    online=True,
    source=driver_stats_source,
    # 태그는 각 FeatureView에 붙는 사용자 정의 키/값 쌍입니다.
    tags={"team": "driver_performance"},
)

# 요청 시점에만 사용할 수 있는 기능/정보를 인코딩하는 RequestSource 정의(예: 사용자가 시작한 HTTP 요청의 일부)
input_request = RequestSource(
    name="vals_to_add",
    schema=[
        Field(name="val_to_add", dtype=Int64),
        Field(name="val_to_add_2", dtype=Int64),
    ],
)

# 기존 FeatureView 및 RequestSource를 기반으로 새 피처을 생성할 수 있는 온디맨드 FeatureView를 정의
@on_demand_feature_view(
    sources=[driver_stats_fv, input_request],
    schema=[
        Field(name="conv_rate_plus_val1", dtype=Float64),
        Field(name="conv_rate_plus_val2", dtype=Float64),
    ],
)
def transformed_conv_rate(inputs: pd.DataFrame) -> pd.DataFrame:
    df = pd.DataFrame()
    df["conv_rate_plus_val1"] = inputs["conv_rate"] + inputs["val_to_add"]
    df["conv_rate_plus_val2"] = inputs["conv_rate"] + inputs["val_to_add_2"]
    return df


# 피처을 모델 버전으로 그룹화합니다.
driver_activity_v1 = FeatureService(
    name="driver_activity_v1",
    features=[
        driver_stats_fv[["conv_rate"]],  # FeatureView에서 서브 피처 선택
        transformed_conv_rate,  # FeatureView에서 모든 피처 선택
    ],
)
driver_activity_v2 = FeatureService(
    name="driver_activity_v2", features=[driver_stats_fv, transformed_conv_rate]
)

# 오프라인, 온라인 또는 둘 다에서 사용할 수 있도록 데이터를 Feast로 푸시하는 방법을 정의합니다.
driver_stats_push_source = PushSource(
    name="driver_stats_push_source",
    batch_source=driver_stats_source,
)

# 소스가 푸시 소스로 변경된 위의 FeatureView의 약간 수정된 버전을 정의합니다. 
# 이렇게 하면 이 FeatureView에 대한 새로운 피처를 온라인 스토어에 직접 푸시할 수 있습니다.
driver_stats_fresh_fv = FeatureView(
    name="driver_hourly_stats_fresh",
    entities=[driver],
    ttl=timedelta(days=1),
    schema=[
        Field(name="conv_rate", dtype=Float32),
        Field(name="acc_rate", dtype=Float32),
        Field(name="avg_daily_trips", dtype=Int64),
    ],
    online=True,
    source=driver_stats_push_source,  # 위에서 변경됨
    tags={"team": "driver_performance"},
)

# 기존 FeatureView 및 RequestSource를 기반으로 새 피처을 생성할 수 있는 온디맨드 FeatureView를 정의
@on_demand_feature_view(
    sources=[driver_stats_fresh_fv, input_request],  # relies on fresh version of FV
    schema=[
        Field(name="conv_rate_plus_val1", dtype=Float64),
        Field(name="conv_rate_plus_val2", dtype=Float64),
    ],
)

def transformed_conv_rate_fresh(inputs: pd.DataFrame) -> pd.DataFrame:
    df = pd.DataFrame()
    df["conv_rate_plus_val1"] = inputs["conv_rate"] + inputs["val_to_add"]
    df["conv_rate_plus_val2"] = inputs["conv_rate"] + inputs["val_to_add_2"]
    return df

driver_activity_v3 = FeatureService(
    name="driver_activity_v3",
    features=[driver_stats_fresh_fv, transformed_conv_rate_fresh],
)

project: some_ml_project
# 기본적으로 registry는 파일이지만 확장성이 뛰어난 SQL-backend registry로 전환할 수 있습니다.
registry: data/registry.db
# provider는 주로 기본 오프라인/온라인 스토어를 지정하고 지정된 클라우드에 레지스트리를 저장합니다.
provider: local
online_store:
    type: sqlite
    path: data/online_store.db
entity_key_serialization_version: 2

import subprocess
from datetime import datetime

import pandas as pd
from feast import FeatureStore
from feast.data_source import PushMode


def run_demo():
    store = FeatureStore(repo_path=".")
    print("\n--- Run feast apply ---")
    subprocess.run(["feast", "apply"])

    print("\n--- Historical features for training ---")
    fetch_historical_features_entity_df(store, for_batch_scoring=False)

    print("\n--- Historical features for batch scoring ---")
    fetch_historical_features_entity_df(store, for_batch_scoring=True)

    print("\n--- Load features into online store ---")
    store.materialize_incremental(end_date=datetime.now())

    print("\n--- Online features ---")
    fetch_online_features(store)

    print("\n--- Online features retrieved (instead) through a feature service---")
    fetch_online_features(store, source="feature_service")

    print(
        "\n--- Online features retrieved (using feature service v3, which uses a feature view with a push source---"
    )
    fetch_online_features(store, source="push")

    print("\n--- Simulate a stream event ingestion of the hourly stats df ---")
    event_df = pd.DataFrame.from_dict(
        {
            "driver_id": [1001],
            "event_timestamp": [
                datetime.now(),
            ],
            "created": [
                datetime.now(),
            ],
            "conv_rate": [1.0],
            "acc_rate": [1.0],
            "avg_daily_trips": [1000],
        }
    )
    print(event_df)
    store.push("driver_stats_push_source", event_df, to=PushMode.ONLINE_AND_OFFLINE)

    print("\n--- Online features again with updated values from a stream push---")
    fetch_online_features(store, source="push")

    print("\n--- Run feast teardown ---")
    subprocess.run(["feast", "teardown"])


def fetch_historical_features_entity_df(store: FeatureStore, for_batch_scoring: bool):
    # 참고: 오프라인 스토어의 모든 엔티티에 대해 검색하는 방법에 대한 자세한 내용은
    # https://docs.feast.dev/getting-started/concepts/feature-retrieval을 참조하세요.
    entity_df = pd.DataFrame.from_dict(
        {
            # entity's join key -> entity values
            "driver_id": [1001, 1002, 1003],
            # "event_timestamp" (reserved key) -> timestamps
            "event_timestamp": [
                datetime(2021, 4, 12, 10, 59, 42),
                datetime(2021, 4, 12, 8, 12, 10),
                datetime(2021, 4, 12, 16, 40, 26),
            ],
            # (optional) label name -> label values. Feast does not process these
            "label_driver_reported_satisfaction": [1, 5, 3],
            # values we're using for an on-demand transformation
            "val_to_add": [1, 2, 3],
            "val_to_add_2": [10, 20, 30],
        }
    )
    # For batch scoring, we want the latest timestamps
    if for_batch_scoring:
        entity_df["event_timestamp"] = pd.to_datetime("now", utc=True)

    training_df = store.get_historical_features(
        entity_df=entity_df,
        features=[
            "driver_hourly_stats:conv_rate",
            "driver_hourly_stats:acc_rate",
            "driver_hourly_stats:avg_daily_trips",
            "transformed_conv_rate:conv_rate_plus_val1",
            "transformed_conv_rate:conv_rate_plus_val2",
        ],
    ).to_df()
    print(training_df.head())


def fetch_online_features(store, source: str = ""):
    entity_rows = [
        # {join_key: entity_value}
        {
            "driver_id": 1001,
            "val_to_add": 1000,
            "val_to_add_2": 2000,
        },
        {
            "driver_id": 1002,
            "val_to_add": 1001,
            "val_to_add_2": 2002,
        },
    ]
    if source == "feature_service":
        features_to_fetch = store.get_feature_service("driver_activity_v1")
    elif source == "push":
        features_to_fetch = store.get_feature_service("driver_activity_v3")
    else:
        features_to_fetch = [
            "driver_hourly_stats:acc_rate",
            "transformed_conv_rate:conv_rate_plus_val1",
            "transformed_conv_rate:conv_rate_plus_val2",
        ]
    returned_features = store.get_online_features(
        features=features_to_fetch,
        entity_rows=entity_rows,
    ).to_dict()
    for key, value in sorted(returned_features.items()):
        print(key, " : ", value)


if __name__ == "__main__":
    run_demo()

import pandas as pd
pd.read_parquet("data/driver_stats.parquet")

1. feast apply 명령어로 피처 정의 등록

2. 학습 데이터 세트 생성(get_historical_features 사용)

3. 배치 예측을 위한 피처 생성(get_historical_features 사용)

4. 온라인 스토어에 배치 피처 수집(materialize_incremental 사용)

5. 실시간 예측을 강화하기 위해 온라인 피처 가져오기(get_online_features 사용)

6. 오프라인/온라인 스토어에 스트리밍 피처 수집(푸시 사용)

7. 온라인 피처가 업데이트 되었는지/새로운 피처인지 확인

$ feast apply

결과

1. 사용자가 타임스탬프가 있는 라벨 테이블을 쿼리하고 이를 학습 데이터 생성을 위한 Entity 데이터 프레임으로 Feast에 전달할 수 있습니다.

2. 사용자가 Entity를 가져오는 SQL 쿼리로 해당 테이블을 쿼리할 수도 있습니다. 자세한 내용은 피처 검색 문서를 참조하세요.

• 학습 데이터 생성 (gen_train_data.py)

from datetime import datetime
import pandas as pd

from feast import FeatureStore

# 참고: 오프라인 스토어의 모든 엔티티에 대해 검색하는 방법에 대한 자세한 내용은
# https://docs.feast.dev/getting-started/concepts/feature-retrieval을 참조하세요.
entity_df = pd.DataFrame.from_dict(
    {
        # entity의 조인 키 -> entity 값
        "driver_id": [1001, 1002, 1003],
        # "event_timestamp" (예약키) -> timestamps
        "event_timestamp": [
            datetime(2021, 4, 12, 10, 59, 42),
            datetime(2021, 4, 12, 8, 12, 10),
            datetime(2021, 4, 12, 16, 40, 26),
        ],
        # (optional) 라벨 이름 -> 라벨 값. Feast는 다음을 처리하지 않습니다.
        "label_driver_reported_satisfaction": [1, 5, 3],
        # 온디맨드 변환에 사용 중인 값
        "val_to_add": [1, 2, 3],
        "val_to_add_2": [10, 20, 30],
    }
)

store = FeatureStore(repo_path=".")

training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "driver_hourly_stats:conv_rate",
        "driver_hourly_stats:acc_rate",
        "driver_hourly_stats:avg_daily_trips",
        "transformed_conv_rate:conv_rate_plus_val1",
        "transformed_conv_rate:conv_rate_plus_val2",
    ],
).to_df()

print("----- Feature schema -----\n")
print(training_df.info())

print()
print("----- Example features -----\n")
print(training_df.head())

결과

• 오프라인 예측 실행 (일괄 예측, batch_inference.py)

# (...... 생략) -> 'gen_train_data.py' 내용과 동일

# 배치 예측은 현재 타임스탬프를 사용
entity_df["event_timestamp"] = pd.to_datetime("now", utc=True)

training_df = store.get_historical_features(
    entity_df=entity_df,
    features=[
        "driver_hourly_stats:conv_rate",
        "driver_hourly_stats:acc_rate",
        "driver_hourly_stats:avg_daily_trips",
        "transformed_conv_rate:conv_rate_plus_val1",
        "transformed_conv_rate:conv_rate_plus_val2",
    ],
).to_df()

print("\n----- Example features -----\n")
print(training_df.head())

결과

 

$ CURRENT_TIME=$(date -u +"%Y-%m-%dT%H:%M:%S")
$ feast materialize-incremental $CURRENT_TIME

결과

 

• fetch_feature_vectors.py

from pprint import pprint
from feast import FeatureStore

store = FeatureStore(repo_path=".")

feature_vector = store.get_online_features(
    features=[
        "driver_hourly_stats:conv_rate",
        "driver_hourly_stats:acc_rate",
        "driver_hourly_stats:avg_daily_trips",
    ],
    entity_rows=[
        # {join_key: entity_value}
        {"driver_id": 1004},
        {"driver_id": 1005},
    ],
).to_dict()

pprint(feature_vector)

결과

• feature_service.py

from pprint import pprint
from feast import FeatureStore

feature_store = FeatureStore('.')  # Initialize the feature store

feature_service = feature_store.get_feature_service("driver_activity_v1")
feature_vector = feature_store.get_online_features(
    features=feature_service,
    entity_rows=[
        # {join_key: entity_value}
        {
            "driver_id": 1001,
            "val_to_add": 1000,
            "val_to_add_2": 2000,
        },
        {
            "driver_id": 1002,
            "val_to_add": 1001,
            "val_to_add_2": 2002,
        },
    ],
).to_dict()

pprint(feature_vector)

결과

 

$ feast ui

Feast에 대한 좀 더 많은 정보는 Feast 공식 문서를 참고하시기 바랍니다.

• 중앙화된 피처 관리: Feast는 데이터 소스에서 추출한 피처를 중앙 집중화하여 관리함으로써 데이터의 재사용성과 일관성을 높입니다.

• 피처 버전 관리: Feast는 피처의 버전을 관리하므로 모델의 성능 변화를 추적하고, 시간에 따른 데이터의 변화를 감지하는 데 유용합니다.

• 실시간 및 배치 데이터 처리 지원: Feast는 실시간 데이터 스트리밍 처리와 배치 데이터 처리를 지원하여, 다양한 유형의 머신러닝 애플리케이션에 적용할 수 있습니다.

• 확장성과 유연성: 클라우드와 온프레미스 환경 모두에서 호환되며, 대용량 데이터와의 호환성을 제공합니다.

• 프로덕션 환경에 적합: Feast는 고성능 프로덕션 환경에 적합하게 설계되어 안정적인 데이터 서비스를 제공합니다.

• 설정과 사용의 복잡성: Feast의 설정과 사용은 초보자에게 다소 복잡하게 느껴질 수 있으며, 이로 인해 초기 학습 곡선이 가파를 수 있습니다.

• 인프라 요구사항: Feast는 특히 클라우드 환경에서 최적화되어 있으며, 이는 특정 인프라에 대한 의존성을 만들 수 있습니다.

• 커뮤니티와 지원: 비교적 새로운 도구인 만큼, Feast의 사용자 커뮤니티와 지원 자료가 다른 잘 알려진 도구들에 비해 제한적일 수 있습니다.

4.3.3 기타 피처 스토어 도구들

4.4 데이터 버전 관리 Vs. 피처 스토어

1. 데이터의 변화 추적: 프로젝트에서 사용되는 데이터 세트가 지속적으로 업데이트되고 변화하는 경우, 이러한 변화를 정확하게 추적하기 위해 필요합니다.

2. 실험 재현성 보장: 모델의 학습 및 평가를 위해 사용된 특정 데이터 세트의 버전을 기록함으로써, 실험의 재현성을 보장하고자 할 때 중요합니다.

3. 팀 협업 효율성: 여러 사람이 동일한 데이터 세트를 사용하여 작업하는 경우, 모든 팀원이 동일한 데이터 세트의 버전을 사용하고 있는지 확인하는 데 도움이 됩니다.

1. 피처 관리의 중앙화: 다양한 머신러닝 모델과 프로젝트에서 재사용 가능한 피처 세트를 중앙화하여 관리하고자 할 때 유용합니다.

2. 모델 성능 향상: 일관된 피처 세트를 사용하여 모델의 성능과 일관성을 향상시키고자 할 때 필요합니다.

3. 실시간 및 배치 데이터 처리: 실시간 데이터 스트리밍 처리와 배치 데이터 처리를 모두 지원하므로, 다양한 유형의 머신러닝 애플리케이션에 적용하고자 할 때 적합합니다.

4.5 요약

• 데이터 라벨링은 원시 데이터에 의미 있는 태그를 부여해 머신러닝 모델 학습에 사용할 수 있도록 하는 과정입니다.
• Label Studio는 다양한 데이터 유형과 프로젝트를 지원하는 유연한 오픈 소스 데이터 라벨링 도구입니다.

• 데이터 버전 관리는 머신러닝 모델의 학습과 평가에 사용되는 데이터 세트의 변경 사항을 추적하고 관리하는 과정으로, 재현성과 팀 협업의 효율성을 높이는 데 중요합니다.
• Git LFS는 대용량 파일을 Git 저장소에 포함 할 수 있도록 합니다. 원격 저장소 용량 제한에 따른 제약사항이 있습니다.
• DVC는 데이터 파일을 별도의 저장소에 보관하며 메타데이터만을 Git 저장소에 저장합니다.

• 피처 스토어는 머신러닝 모델을 위한 피처(특성) 데이터를 중앙화하여 관리하는 시스템입니다.
• Feast는 중앙화된 피처 관리, 피처 버전 관리, 실시간 및 배치 데이터 처리를 지원하는 피처 스토어 도구입니다. 데이터 버전 관리와 피처 스토어의 효과적인 활용은 머신러닝 모델의 성능과 일관성을 향상시키는 데 기여합니다.