피처 테이블 관리 (Feature Table Management)

from databricks.feature_engineering import FeatureEngineeringClient
from pyspark.sql import functions as F

fe = FeatureEngineeringClient()

# 1. 원본 데이터에서 피처 계산
transactions_df = spark.table("catalog.schema.transactions")

customer_features_df = (
    transactions_df
    .groupBy("customer_id")
    .agg(
        F.count("*").alias("total_transactions"),
        F.avg("amount").alias("avg_transaction_amount"),
        F.max("amount").alias("max_transaction_amount"),
        F.stddev("amount").alias("stddev_transaction_amount"),
        F.countDistinct("merchant_category").alias("unique_merchants"),
        F.avg(F.when(F.col("hour").between(0, 6), 1).otherwise(0)).alias("night_transaction_ratio"),
        F.max("transaction_date").alias("last_transaction_date")
    )
)

# 2. 피처 테이블 생성
fe.create_table(
    name="catalog.schema.customer_features",
    primary_keys=["customer_id"],
    df=customer_features_df,
    description="고객별 거래 패턴 피처 (30일 집계)",
    tags={"team": "fraud-detection", "refresh": "daily"}
)

# 시간별 집계 피처 (타임스탬프 포함)
hourly_features_df = (
    transactions_df
    .groupBy("customer_id", F.window("transaction_time", "1 hour").alias("time_window"))
    .agg(
        F.count("*").alias("hourly_transaction_count"),
        F.sum("amount").alias("hourly_transaction_sum")
    )
    .withColumn("timestamp", F.col("time_window.end"))
    .drop("time_window")
)

fe.create_table(
    name="catalog.schema.customer_hourly_features",
    primary_keys=["customer_id"],
    timestamp_keys=["timestamp"],    # Point-in-time lookup용
    df=hourly_features_df,
    description="고객별 시간대별 거래 피처"
)

from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup

fe = FeatureEngineeringClient()

# 라벨 데이터 (기본 키 + 타겟 변수만 포함)
labels_df = spark.table("catalog.schema.fraud_labels")
# 컬럼: customer_id, transaction_id, is_fraud, event_time

# 피처 조회 정의
feature_lookups = [
    # 고객 피처 전체 가져오기
    FeatureLookup(
        table_name="catalog.schema.customer_features",
        lookup_key="customer_id"
        # feature_names를 생략하면 기본 키를 제외한 모든 피처를 가져옵니다
    ),

    # 특정 피처만 선택하여 가져오기
    FeatureLookup(
        table_name="catalog.schema.merchant_features",
        feature_names=["merchant_risk_score", "avg_fraud_rate"],
        lookup_key="merchant_id"
    ),

    # 조인 키 이름이 다른 경우 매핑
    FeatureLookup(
        table_name="catalog.schema.product_features",
        lookup_key=["product_id"],
        rename_outputs={"product_category": "item_category"}  # 피처 이름 변경
    )
]

# 학습 데이터셋 생성
training_set = fe.create_training_set(
    df=labels_df,
    feature_lookups=feature_lookups,
    label="is_fraud",
    exclude_columns=["transaction_id"]  # 학습에 불필요한 컬럼 제외
)

# Pandas DataFrame으로 변환하여 학습
training_df = training_set.load_df()
display(training_df)

# Point-in-time Lookup 예시
training_set = fe.create_training_set(
    df=labels_df,  # event_time 컬럼 포함
    feature_lookups=[
        FeatureLookup(
            table_name="catalog.schema.customer_hourly_features",
            lookup_key="customer_id",
            timestamp_lookup_key="event_time"  # 이 시점 기준으로 조회
        )
    ],
    label="is_fraud"
)

# event_time이 2025-03-15 14:00인 레코드 → 14:00 이전의 최신 피처만 가져옴
# event_time이 2025-03-15 16:00인 레코드 → 16:00 이전의 최신 피처만 가져옴

from databricks.feature_engineering import FeatureEngineeringClient
from databricks.feature_engineering.online_store_spec import AmazonDynamoDBSpec, AzureCosmosDBSpec

fe = FeatureEngineeringClient()

# AWS DynamoDB로 퍼블리시
dynamodb_spec = AmazonDynamoDBSpec(
    region="ap-northeast-2",
    table_name="customer_features_online",
    read_secret_prefix="feature-store/dynamo",
    write_secret_prefix="feature-store/dynamo"
)

fe.publish_table(
    name="catalog.schema.customer_features",
    online_store=dynamodb_spec,
    mode="merge"  # 기존 데이터와 병합
)

# Azure Cosmos DB로 퍼블리시
cosmosdb_spec = AzureCosmosDBSpec(
    account_uri="https://my-cosmos.documents.azure.com:443/",
    read_secret="feature-store/cosmos-key",
    write_secret="feature-store/cosmos-key"
)

fe.publish_table(
    name="catalog.schema.customer_features",
    online_store=cosmosdb_spec,
    mode="merge"
)

from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import f1_score
import mlflow

fe = FeatureEngineeringClient()

# ── Step 1: 피처 계산 및 테이블 생성 ──
customer_features_df = (
    spark.table("catalog.schema.transactions")
    .groupBy("customer_id")
    .agg(
        F.count("*").alias("total_transactions"),
        F.avg("amount").alias("avg_amount"),
        F.stddev("amount").alias("stddev_amount"),
        F.countDistinct("merchant_id").alias("unique_merchants")
    )
)

fe.create_table(
    name="catalog.ml_features.customer_features",
    primary_keys=["customer_id"],
    df=customer_features_df,
    description="고객별 거래 통계 피처"
)

# ── Step 2: 학습 데이터 생성 (FeatureLookup) ──
labels_df = spark.table("catalog.schema.fraud_labels")

training_set = fe.create_training_set(
    df=labels_df,
    feature_lookups=[
        FeatureLookup(
            table_name="catalog.ml_features.customer_features",
            lookup_key="customer_id"
        )
    ],
    label="is_fraud"
)

training_df = training_set.load_df().toPandas()

# ── Step 3: 모델 학습 ──
X = training_df.drop(columns=["is_fraud", "customer_id"])
y = training_df["is_fraud"]

model = GradientBoostingClassifier(n_estimators=200, learning_rate=0.1)
model.fit(X, y)

# ── Step 4: 모델 저장 (피처 메타데이터 포함) ──
with mlflow.start_run(run_name="fraud-with-feature-store"):
    fe.log_model(
        model=model,
        artifact_path="model",
        flavor=mlflow.sklearn,
        training_set=training_set,  # 피처 조회 정보가 모델에 기록됨
        registered_model_name="catalog.schema.fraud_model_with_features"
    )

# ── Step 5: 서빙 시 자동 피처 조회 ──
# 엔드포인트에 배포 후, 기본 키만 전달하면 피처는 자동 조회됩니다
# 요청 예시: {"dataframe_records": [{"customer_id": "C12345"}]}
# → customer_features 테이블에서 피처를 자동으로 가져와 예측 수행

# 방법 1: merge (기존 키와 겹치면 업데이트, 새 키는 추가)
fe.write_table(
    name="catalog.ml_features.customer_features",
    df=new_features_df,
    mode="merge"
)

# 방법 2: overwrite (전체 교체)
fe.write_table(
    name="catalog.ml_features.customer_features",
    df=full_features_df,
    mode="overwrite"
)