{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.12.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"nvidiaTeslaT4","dataSources":[{"sourceType":"competition","sourceId":31254,"databundleVersionId":3103714}],"dockerImageVersionId":31329,"isInternetEnabled":true,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"**1. Data Collection**","metadata":{}},{"cell_type":"code","source":"!pip install pyspark synapseml","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-04-16T15:41:06.005638Z","iopub.execute_input":"2026-04-16T15:41:06.005960Z","iopub.status.idle":"2026-04-16T15:41:10.905695Z","shell.execute_reply.started":"2026-04-16T15:41:06.005907Z","shell.execute_reply":"2026-04-16T15:41:10.904683Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"import pyspark\nfrom pyspark.sql import SparkSession\n\nspark = SparkSession.builder.appName(\"Final_Project\").getOrCreate()\nsc = spark.sparkContext","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-04-16T15:41:29.672877Z","iopub.execute_input":"2026-04-16T15:41:29.673974Z","iopub.status.idle":"2026-04-16T15:41:39.351957Z","shell.execute_reply.started":"2026-04-16T15:41:29.673923Z","shell.execute_reply":"2026-04-16T15:41:39.350955Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"BASE = '/kaggle/input/competitions/h-and-m-personalized-fashion-recommendations/'\n\ntxn = spark.read.csv(BASE + 'transactions_train.csv', header=True, inferSchema=True)\nart = spark.read.csv(BASE + 'articles.csv',  header=True, inferSchema=True)\ncus = spark.read.csv(BASE + 'customers.csv', header=True, inferSchema=True)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-04-16T15:44:35.796048Z","iopub.execute_input":"2026-04-16T15:44:35.796794Z","iopub.status.idle":"2026-04-16T15:45:18.404845Z","shell.execute_reply.started":"2026-04-16T15:44:35.796760Z","shell.execute_reply":"2026-04-16T15:45:18.403850Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"**2. Data Preparation**","metadata":{}},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"**3. Modeling**","metadata":{}},{"cell_type":"code","source":"# ── Cell 22: SMOTE → PySpark (using imbalanced-learn bridge) ──────────────────\n# Note: Native PySpark doesn't have SMOTE.\n# Approach: oversample minority class using PySpark's built-in sampling,\n# or collect to driver for SMOTE then redistribute (feasible if data fits in memory).\n\nfrom pyspark.sql import functions as F\nfrom pyspark.sql import SparkSession\n\nspark = SparkSession.builder.getOrCreate()\n\n# Option A: Simple oversampling (pure PySpark, no driver collect needed)\nmajority = train_df.filter(F.col(\"label\") == 0)\nminority = train_df.filter(F.col(\"label\") == 1)\n\nratio = majority.count() / minority.count()          # ~12.8x\noversampled_minority = minority.sample(\n    withReplacement=True, fraction=ratio, seed=SEED\n)\ntrain_balanced_df = majority.union(oversampled_minority)\n\n# Option B: SMOTE on driver (if dataset fits in memory ~same as your original)\n# X_train_res_pd, y_train_res_pd = smote.fit_resample(X_train_pd, y_train_pd)\n# train_balanced_df = spark.createDataFrame(\n#     pd.concat([X_train_res_pd, y_train_res_pd], axis=1)\n# )\n\nprint(\"Before:\", train_df.groupBy(\"label\").count().show())\nprint(\"After :\", train_balanced_df.groupBy(\"label\").count().show())","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"**LGBM**","metadata":{}},{"cell_type":"code","source":"# ── Cell 23a: LightGBM Default Model ─────────────────────────────────────────\nfrom synapse.ml.lightgbm import LightGBMClassifier\nfrom pyspark.ml.feature import VectorAssembler\nfrom pyspark.ml.evaluation import MulticlassClassificationEvaluator\nfrom pyspark.sql import functions as F\n\nSEED = 42\n\n# Assemble features\nfeature_cols = [c for c in train_balanced_df.columns if c != \"label\"]\nassembler    = VectorAssembler(inputCols=feature_cols, outputCol=\"features\")\n\ntrain_vec = assembler.transform(train_balanced_df)\nval_vec   = assembler.transform(val_df)\n\nf1_evaluator = MulticlassClassificationEvaluator(\n    labelCol=\"label\", predictionCol=\"prediction\", metricName=\"f1\"\n)\n\n# Fit default model\nlgb_default   = LightGBMClassifier(\n    labelCol=\"label\",\n    featuresCol=\"features\",\n    predictionCol=\"prediction\",\n    probabilityCol=\"probability\",\n    seed=SEED,\n    verbosity=-1,\n)\ndefault_model = lgb_default.fit(train_vec)\n\n# Evaluate\nf1_default = f1_evaluator.evaluate(default_model.transform(val_vec))\nprint(f\"Default model — Validation F1: {f1_default:.4f}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-04-16T15:54:33.107669Z","iopub.execute_input":"2026-04-16T15:54:33.108338Z","iopub.status.idle":"2026-04-16T15:54:33.113548Z","shell.execute_reply.started":"2026-04-16T15:54:33.108287Z","shell.execute_reply":"2026-04-16T15:54:33.112473Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ── Cell 23b: LightGBM Tuned Model ───────────────────────────────────────────\nfrom synapse.ml.lightgbm import LightGBMClassifier\n\n# Fit tuned model\nlgb_tuned   = LightGBMClassifier(\n    numIterations=500,\n    learningRate=0.05,\n    numLeaves=31,\n    baggingFraction=0.8,\n    featureFraction=0.8,\n    minSumHessianInLeaf=20,\n    isUnbalance=True,\n    labelCol=\"label\",\n    featuresCol=\"features\",\n    predictionCol=\"prediction\",\n    probabilityCol=\"probability\",\n    seed=SEED,\n    verbosity=-1,\n    numThreads=-1,\n)\ntuned_model = lgb_tuned.fit(train_vec)\n\n# Evaluate\nf1_tuned = f1_evaluator.evaluate(tuned_model.transform(val_vec))\nprint(f\"Tuned model   — Validation F1: {f1_tuned:.4f}\")\nprint(f\"F1 improvement: {f1_tuned - f1_default:+.4f}\")\n\n# Best model + test predictions\nlgb_model   = tuned_model if f1_tuned >= f1_default else default_model\npred_lgb_df = lgb_model.transform(assembler.transform(test_df)).select(\n    \"id\", F.col(\"probability\")[1].alias(\"pred_lgb\")\n)","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"**XGB**","metadata":{}},{"cell_type":"code","source":"# ── Cell 24a: XGBoost Default Model ──────────────────────────────────────────\nfrom xgboost.spark import SparkXGBClassifier\nfrom pyspark.ml.feature import VectorAssembler\nfrom pyspark.ml.evaluation import MulticlassClassificationEvaluator\nfrom pyspark.sql import functions as F\n\nSEED = 42\n\n# Assemble features\nfeature_cols = [c for c in train_balanced_df.columns if c != \"label\"]\nassembler    = VectorAssembler(inputCols=feature_cols, outputCol=\"features\")\n\ntrain_vec = assembler.transform(train_balanced_df)\nval_vec   = assembler.transform(val_df)\n\nf1_evaluator = MulticlassClassificationEvaluator(\n    labelCol=\"label\", predictionCol=\"prediction\", metricName=\"f1\"\n)\n\n# scale_pos_weight from class distribution\ncounts    = train_balanced_df.groupBy(\"label\").count().collect()\ncount_map = {row[\"label\"]: row[\"count\"] for row in counts}\nspw       = count_map[0] / count_map[1]\nprint(f\"scale_pos_weight: {spw:.4f}\")\n\n# Fit default model\nxgb_default   = SparkXGBClassifier(\n    scale_pos_weight=spw,\n    label_col=\"label\",\n    features_col=\"features\",\n    prediction_col=\"prediction\",\n    probability_col=\"probability\",\n    num_workers=4,\n    use_gpu=False,\n    seed=SEED,\n)\ndefault_model = xgb_default.fit(train_vec)\n\n# Evaluate\nf1_default = f1_evaluator.evaluate(default_model.transform(val_vec))\nprint(f\"Default model — Validation F1: {f1_default:.4f}\")","metadata":{"trusted":true},"outputs":[],"execution_count":null},{"cell_type":"code","source":"# ── Cell 24b: XGBoost Tuned Model ────────────────────────────────────────────\nfrom xgboost.spark import SparkXGBClassifier\n\n# Fit tuned model\nxgb_tuned   = SparkXGBClassifier(\n    n_estimators=500,\n    learning_rate=0.05,\n    max_depth=6,\n    subsample=0.8,\n    colsample_bytree=0.8,\n    scale_pos_weight=spw,\n    eval_metric=\"logloss\",\n    tree_method=\"hist\",\n    device=\"cpu\",\n    label_col=\"label\",\n    features_col=\"features\",\n    prediction_col=\"prediction\",\n    probability_col=\"probability\",\n    num_workers=4,\n    use_gpu=False,\n    seed=SEED,\n)\ntuned_model = xgb_tuned.fit(train_vec)\n\n# Evaluate\nf1_tuned = f1_evaluator.evaluate(tuned_model.transform(val_vec))\nprint(f\"Tuned model   — Validation F1: {f1_tuned:.4f}\")\nprint(f\"F1 improvement: {f1_tuned - f1_default:+.4f}\")\n\n# Best model + test predictions\nxgb_model   = tuned_model if f1_tuned >= f1_default else default_model\npred_xgb_df = xgb_model.transform(assembler.transform(test_df)).select(\n    \"id\", F.col(\"probability\")[1].alias(\"pred_xgb\")\n)","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}