PySpark ML 피처 엔지니어링 — 대규모 파이프라인과 데이터 누수 방지

① Train/Test 누수: 테스트 데이터의 통계가 학습 전처리에 섞임
   예) 전체 데이터로 평균을 구해 정규화 → 테스트 정보가 학습에 누설
 
② 시간 누수(미래 정보): 예측 시점 이후의 데이터가 피처에 포함
   예) "현재" 고객 등급으로 과거 라벨을 예측 → 미래값 사용

from pyspark.ml.feature import StandardScaler
 
# 먼저 train/test 분리 (전처리보다 먼저!)
train, test = df.randomSplit([0.8, 0.2], seed=42)
 
scaler = StandardScaler(inputCol="features", outputCol="scaled")
 
model = scaler.fit(train)        # ✅ 통계(평균·표준편차)를 train 에서만 학습
train_s = model.transform(train) # 적용
test_s  = model.transform(test)  # 같은 모델로 적용 (test 통계 안 씀)

# ❌ 누수: 전체로 fit 한 뒤 split → test 정보가 scaler 에 섞임
model = scaler.fit(df)           # 전체 통계 사용 → 누수
train, test = df.randomSplit(...)

from pyspark.ml import Pipeline
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
 
# 범주형 인코딩 → 벡터 조립 → 스케일링을 한 파이프라인으로
indexer = StringIndexer(inputCols=["city", "device"],
                        outputCols=["city_idx", "device_idx"],
                        handleInvalid="keep")          # 미지 카테고리 처리
encoder = OneHotEncoder(inputCols=["city_idx", "device_idx"],
                        outputCols=["city_oh", "device_oh"])
assembler = VectorAssembler(
    inputCols=["age", "amount", "city_oh", "device_oh"],
    outputCol="features",
    handleInvalid="skip")
scaler = StandardScaler(inputCol="features", outputCol="scaled")
 
pipeline = Pipeline(stages=[indexer, encoder, assembler, scaler])
 
model = pipeline.fit(train)      # 전체 파이프라인을 train 에서 학습
train_f = model.transform(train)
test_f  = model.transform(test)
 
# 파이프라인 모델을 저장 → 서빙에서 동일 변환 보장
model.write().overwrite().save("/models/feature_pipeline")

from pyspark.ml.feature import FeatureHasher
 
hasher = FeatureHasher(inputCols=["product_id", "city"],
                       outputCol="hashed", numFeatures=1 << 18)

# ❌ 누수: 라벨 시점 이후일 수 있는 현재 등급을 조인
features = labels.join(dim_users_current, "user_id")
 
# ✅ as-of: 라벨 시점 이전의 유효값만
features = labels.join(dim_users_history,
    (labels.user_id == dim_users_history.user_id) &
    (labels.label_time >= dim_users_history.valid_from) &
    (labels.label_time <  dim_users_history.valid_to))

from pyspark.sql.window import Window
from pyspark.sql import functions as F
 
# 라벨 시점 "이전" 데이터로만 집계 (rangeBetween 으로 미래 차단)
w = (Window.partitionBy("user_id")
            .orderBy(F.col("event_time").cast("long"))
            .rangeBetween(-30*86400, -1))   # 직전 30일 ~ 현재 직전(미래 배제)
 
features = df.withColumn("avg_30d", F.avg("amount").over(w))