02. 피처 엔지니어링 (Feature Engineering)

# 센서값 물리적 범위 검증 (실제 공장에서 매우 중요!)
range_checks = {
    "air_temperature_k": (250, 350),      # 절대온도 범위
    "process_temperature_k": (250, 400),
    "rotational_speed_rpm": (0, 5000),     # RPM 범위
    "torque_nm": (0, 200),                 # 토크 범위
    "tool_wear_min": (0, 500),             # 마모 시간 범위
}
for col, (low, high) in range_checks.items():
    out_of_range = df_bronze.filter((F.col(col) < low) | (F.col(col) > high)).count()
    status = "OK" if out_of_range == 0 else f"{out_of_range}건 이탈"
    print(f"  {col}: [{low}~{high}] {status}")

-- 제품 타입별 고장률 분석
SELECT
  type as product_type,
  COUNT(*) as total,
  SUM(machine_failure) as failures,
  ROUND(SUM(machine_failure) / COUNT(*) * 100, 2) as failure_rate_pct
FROM lgit_pm_bronze
GROUP BY type
ORDER BY failure_rate_pct DESC

psdf = df_bronze.pandas_api()
display(psdf[['air_temperature_k', 'process_temperature_k',
              'rotational_speed_rpm', 'torque_nm', 'tool_wear_min']].describe())

def engineer_pm_features(df: DataFrame) -> DataFrame:
    return (df
        .withColumn("temp_diff", F.col("process_temperature_k") - F.col("air_temperature_k"))
        .withColumn("power", F.col("torque_nm") * F.col("rotational_speed_rpm") * F.lit(2 * math.pi / 60))
        .withColumn("tool_wear_rate",
                    F.when(F.col("rotational_speed_rpm") > 0,
                           F.col("tool_wear_min") / F.col("rotational_speed_rpm")).otherwise(0))
        .withColumn("strain", F.col("torque_nm") * F.col("tool_wear_min"))
        .withColumn("overheat_flag",
                    F.when(F.col("process_temperature_k") - F.col("air_temperature_k") > 8.6, 1).otherwise(0))
        .withColumn("product_quality",
                    F.when(F.col("type") == "L", 0).when(F.col("type") == "M", 1).otherwise(2))
        .withColumn("risk_score",
                    (F.col("tool_wear_min") / F.lit(240.0)) * 0.3 +
                    (F.col("torque_nm") / F.lit(80.0)) * 0.3 +
                    F.when(F.col("process_temperature_k") - F.col("air_temperature_k") > 8.6, 0.4).otherwise(0.0))
    )

# Train/Test 분할
df_training = (
    df_training
    .withColumn("random", F.rand(seed=42))
    .withColumn("split", F.when(F.col("random") < 0.8, "train").otherwise("test"))
    .drop("random")
)

# Delta Lake 테이블로 저장
df_training.write.mode("overwrite").option("overwriteSchema", "true").saveAsTable("lgit_pm_training")

# Unity Catalog 메타데이터 추가
spark.sql(f"""
    COMMENT ON TABLE {catalog}.{db}.lgit_pm_training
    IS '예지보전 학습 데이터: AI4I 2020 데이터셋 기반. 센서 피처 및 파생 피처 포함.'
""")

pdf = df_training.filter("split = 'train'").select(*feature_columns, label_column).toPandas()
target_corr = pdf.corr()[label_column].drop(label_column).sort_values(ascending=False)
for feat, corr_val in target_corr.items():
    print(f"  {feat:30s}: {corr_val:+.4f}")

from databricks.feature_engineering import FeatureEngineeringClient

fe = FeatureEngineeringClient()
fe_table = fe.create_table(
    name=f"{catalog}.{db}.lgit_pm_features",
    primary_keys=["udi"],
    df=spark.table("lgit_pm_training"),
    description="AI4I 2020 예지보전 피처 테이블 — 7개 파생 피처 포함"
)