{"metadata":{"kernelspec":{"display_name":"Python 3","language":"python","name":"python3"},"language_info":{"name":"python","version":"3.12.13","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"}},"nbformat_minor":4,"nbformat":4,"cells":[{"id":"a1b2c3d4","cell_type":"markdown","source":"# 🔬 Human Protein Atlas — Image Classification (PySpark on Kaggle)\n\n**Task**: Multi-label classification of subcellular protein localization patterns  \n**Labels**: 28 organelle classes (can be multiple per sample)  \n**Input**: 4 filter images per sample — green (protein), blue (nucleus), red (microtubules), yellow (ER)  \n**Prediction target**: Green channel drives the label; others are reference channels.\n\n---\n### Pipeline Overview\n1. Initialize PySpark Session\n2. Load & explore `train.csv` / `sample_submission.csv`\n3. Image feature extraction (pixel statistics per channel)\n4. Multi-label encoding\n5. Train multi-label classifiers (one per label via PySpark MLlib)\n6. Evaluate on validation split\n7. Generate submission file","metadata":{}},{"id":"42f5c281-de25-4c9a-94c0-0e19f25438b0","cell_type":"markdown","source":"","metadata":{}},{"id":"b1b2c3d5","cell_type":"markdown","source":"## 📦 Cell 1 — Install & Initialize PySpark","metadata":{}},{"id":"c1d2e3f4","cell_type":"code","source":"import subprocess\nsubprocess.run([\"pip\", \"install\", \"pyspark\", \"--quiet\"], check=True)\n\nimport os\nimport warnings\nwarnings.filterwarnings('ignore')\n\nfrom pyspark.sql import SparkSession\nfrom pyspark.sql import functions as F\nfrom pyspark.sql.types import (\n    StructType, StructField, StringType, IntegerType,\n    FloatType, ArrayType, DoubleType\n)\n\nspark = (\n    SparkSession.builder\n    .appName(\"HumanProteinAtlas_Classification\")\n    .config(\"spark.driver.memory\", \"8g\")\n    .config(\"spark.executor.memory\", \"8g\")\n    .config(\"spark.sql.shuffle.partitions\", \"50\")\n    .config(\"spark.serializer\", \"org.apache.spark.serializer.KryoSerializer\")\n    .getOrCreate()\n)\n\nspark.sparkContext.setLogLevel(\"WARN\")\nprint(f\"Spark version : {spark.version}\")\nprint(f\"App name      : {spark.sparkContext.appName}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T04:59:23.493904Z","iopub.execute_input":"2026-08-19T04:59:23.494648Z","iopub.status.idle":"2026-08-19T04:59:35.909752Z","shell.execute_reply.started":"2026-08-19T04:59:23.494616Z","shell.execute_reply":"2026-08-19T04:59:35.908936Z"}},"outputs":[],"execution_count":null},{"id":"d1e2f3g4","cell_type":"markdown","source":"## 📂 Cell 2 — Define Paths & Label Mapping","metadata":{}},{"id":"e1f2g3h4","cell_type":"code","source":"BASE_PATH   = \"/kaggle/input/competitions/human-protein-atlas-image-classification\"\nTRAIN_DIR   = os.path.join(BASE_PATH, \"train\")\nTEST_DIR    = os.path.join(BASE_PATH, \"test\")\nTRAIN_CSV   = os.path.join(BASE_PATH, \"train.csv\")\nSAMPLE_SUB  = os.path.join(BASE_PATH, \"sample_submission.csv\")\n\nLABEL_MAP = {\n    0:  \"Nucleoplasm\",\n    1:  \"Nuclear membrane\",\n    2:  \"Nucleoli\",\n    3:  \"Nucleoli fibrillar center\",\n    4:  \"Nuclear speckles\",\n    5:  \"Nuclear bodies\",\n    6:  \"Endoplasmic reticulum\",\n    7:  \"Golgi apparatus\",\n    8:  \"Peroxisomes\",\n    9:  \"Endosomes\",\n    10: \"Lysosomes\",\n    11: \"Intermediate filaments\",\n    12: \"Actin filaments\",\n    13: \"Focal adhesion sites\",\n    14: \"Microtubules\",\n    15: \"Microtubule ends\",\n    16: \"Cytokinetic bridge\",\n    17: \"Mitotic spindle\",\n    18: \"Microtubule organizing center\",\n    19: \"Centrosome\",\n    20: \"Lipid droplets\",\n    21: \"Plasma membrane\",\n    22: \"Cell junctions\",\n    23: \"Mitochondria\",\n    24: \"Aggresome\",\n    25: \"Cytosol\",\n    26: \"Cytoplasmic bodies\",\n    27: \"Rods & rings\",\n}\nNUM_CLASSES = len(LABEL_MAP)\nCHANNELS    = [\"green\", \"blue\", \"red\", \"yellow\"]\n\nprint(f\"Number of classes : {NUM_CLASSES}\")\nprint(f\"Channels          : {CHANNELS}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T04:59:35.911389Z","iopub.execute_input":"2026-08-19T04:59:35.911796Z","iopub.status.idle":"2026-08-19T04:59:35.919569Z","shell.execute_reply.started":"2026-08-19T04:59:35.911771Z","shell.execute_reply":"2026-08-19T04:59:35.918822Z"}},"outputs":[],"execution_count":null},{"id":"f1g2h3i4","cell_type":"markdown","source":"## 📊 Cell 3 — Load train.csv & sample_submission.csv into Spark","metadata":{}},{"id":"g1h2i3j4","cell_type":"code","source":"train_df = spark.read.csv(TRAIN_CSV, header=True, inferSchema=True)\nsub_df   = spark.read.csv(SAMPLE_SUB, header=True, inferSchema=True)\n\nprint(\"TRAIN CSV Schema:\")\ntrain_df.printSchema()\ntrain_df.show(5, truncate=False)\n\nprint(\"\\nSAMPLE SUBMISSION Schema:\")\nsub_df.printSchema()\nsub_df.show(5, truncate=False)\n\nprint(f\"\\nTotal training samples : {train_df.count()}\")\nprint(f\"Total test samples     : {sub_df.count()}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T04:59:35.920435Z","iopub.execute_input":"2026-08-19T04:59:35.920803Z","iopub.status.idle":"2026-08-19T04:59:41.527569Z","shell.execute_reply.started":"2026-08-19T04:59:35.920781Z","shell.execute_reply":"2026-08-19T04:59:41.526999Z"}},"outputs":[],"execution_count":null},{"id":"h1i2j3k4","cell_type":"markdown","source":"## 🔍 Cell 4 — Exploratory Data Analysis","metadata":{}},{"id":"i1j2k3l4","cell_type":"code","source":"import pandas as pd\nimport numpy as np\nimport matplotlib.pyplot as plt\nfrom collections import Counter\n\ntrain_pd = train_df.toPandas()\n\nall_labels = []\nfor target_str in train_pd[\"Target\"]:\n    all_labels.extend([int(x) for x in str(target_str).split()])\n\nlabel_counts = Counter(all_labels)\nsorted_labels = sorted(label_counts.items(), key=lambda x: -x[1])\nlabel_names_sorted = [LABEL_MAP[lbl] for lbl, _ in sorted_labels]\nlabel_freqs        = [cnt            for _,   cnt in sorted_labels]\n\nfig, axes = plt.subplots(2, 1, figsize=(16, 14))\n\ncolors = plt.cm.viridis(np.linspace(0.1, 0.9, len(label_names_sorted)))\naxes[0].barh(label_names_sorted, label_freqs, color=colors)\naxes[0].set_xlabel(\"Frequency\", fontsize=12)\naxes[0].set_title(\"Label Frequency in Training Set\", fontsize=14, fontweight='bold')\naxes[0].invert_yaxis()\nfor i, (name, freq) in enumerate(zip(label_names_sorted, label_freqs)):\n    axes[0].text(freq + 50, i, str(freq), va='center', fontsize=9)\n\ntrain_pd[\"label_count\"] = train_pd[\"Target\"].apply(lambda x: len(str(x).split()))\nlc_counts = train_pd[\"label_count\"].value_counts().sort_index()\naxes[1].bar(lc_counts.index, lc_counts.values, color='steelblue', edgecolor='navy')\naxes[1].set_xlabel(\"Number of Labels per Sample\", fontsize=12)\naxes[1].set_ylabel(\"Number of Samples\", fontsize=12)\naxes[1].set_title(\"Distribution of Label Count per Sample\", fontsize=14, fontweight='bold')\nfor x, y in zip(lc_counts.index, lc_counts.values):\n    axes[1].text(x, y + 20, str(y), ha='center', fontsize=9)\n\nplt.tight_layout()\nplt.savefig(\"/kaggle/working/eda_label_distribution.png\", dpi=120, bbox_inches='tight')\nplt.show()\nprint(f\"\\nAvg labels per sample : {train_pd['label_count'].mean():.2f}\")\nprint(f\"Max labels per sample : {train_pd['label_count'].max()}\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T04:59:41.529350Z","iopub.execute_input":"2026-08-19T04:59:41.529633Z","iopub.status.idle":"2026-08-19T04:59:43.249023Z","shell.execute_reply.started":"2026-08-19T04:59:41.529605Z","shell.execute_reply":"2026-08-19T04:59:43.248070Z"}},"outputs":[],"execution_count":null},{"id":"j1k2l3m4","cell_type":"markdown","source":"## 🖼️ Cell 5 — Image Visualization (Sample Inspection)","metadata":{}},{"id":"k1l2m3n4","cell_type":"code","source":"from PIL import Image\n\ndef show_sample(sample_id, split=\"train\"):\n    base_dir  = TRAIN_DIR if split == \"train\" else TEST_DIR\n    channels  = [\"green\", \"blue\", \"red\", \"yellow\"]\n    ch_colors = [\"Greens\", \"Blues\", \"Reds\", \"YlOrBr\"]\n    ch_titles = [\n        \"Green (Protein - TARGET)\",\n        \"Blue (Nucleus)\",\n        \"Red (Microtubules)\",\n        \"Yellow (ER)\"\n    ]\n\n    fig, axes = plt.subplots(1, 4, figsize=(20, 5))\n    fig.suptitle(f\"Sample ID: {sample_id}\", fontsize=14, fontweight='bold')\n\n    for ax, ch, cmap, title in zip(axes, channels, ch_colors, ch_titles):\n        img_path = os.path.join(base_dir, f\"{sample_id}_{ch}.png\")\n        if os.path.exists(img_path):\n            img = np.array(Image.open(img_path))\n            ax.imshow(img, cmap=cmap)\n            ax.set_title(title, fontsize=10)\n        else:\n            ax.text(0.5, 0.5, f\"Not found:\\n{img_path}\",\n                    ha='center', va='center', transform=ax.transAxes, fontsize=8)\n        ax.axis('off')\n\n    plt.tight_layout()\n    plt.show()\n\nsample_ids = train_pd[\"Id\"].head(3).tolist()\nfor sid in sample_ids:\n    label_ids_sample  = [int(x) for x in str(train_pd.loc[train_pd['Id']==sid, 'Target'].values[0]).split()]\n    label_names_sample = [LABEL_MAP[i] for i in label_ids_sample]\n    print(f\"ID: {sid}  |  Labels: {label_names_sample}\")\n    show_sample(sid, split=\"train\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T04:59:43.250049Z","iopub.execute_input":"2026-08-19T04:59:43.250384Z","iopub.status.idle":"2026-08-19T04:59:45.726101Z","shell.execute_reply.started":"2026-08-19T04:59:43.250360Z","shell.execute_reply":"2026-08-19T04:59:45.725303Z"}},"outputs":[],"execution_count":null},{"id":"l1m2n3o4","cell_type":"markdown","source":"## 🧮 Cell 6 — Feature Extraction via PySpark UDF\n\nExtract per-channel pixel statistics as numerical features for MLlib.  \n**36 features = 4 channels × 9 statistics** (mean, std, min, max, p10, p25, p50, p75, p90)","metadata":{}},{"id":"m1n2o3p4","cell_type":"code","source":"# Cell 6 — Pandas-based feature extraction (avoids UDF deadlock on Kaggle)\nimport pandas as pd\nimport numpy as np\nfrom PIL import Image\nimport os\nfrom tqdm.notebook import tqdm\n\nCHANNELS_ORDERED = [\"green\", \"blue\", \"red\", \"yellow\"]\nSTATS_KEYS = [\"mean\", \"std\", \"min\", \"max\", \"p10\", \"p25\", \"p50\", \"p75\", \"p90\"]\n\ndef compute_stats(arr):\n    arr = arr.astype(np.float32).ravel()\n    return [\n        float(arr.mean()), float(arr.std()),\n        float(arr.min()),  float(arr.max()),\n        float(np.percentile(arr, 10)), float(np.percentile(arr, 25)),\n        float(np.percentile(arr, 50)), float(np.percentile(arr, 75)),\n        float(np.percentile(arr, 90)),\n    ]\n\ndef extract_features_pandas(id_list, base_dir):\n    \"\"\"Extract 36 pixel-stat features per sample on the driver.\"\"\"\n    rows = []\n    for sample_id in tqdm(id_list, desc=\"Extracting features\"):\n        feats = []\n        for ch in CHANNELS_ORDERED:\n            img_path = os.path.join(base_dir, f\"{sample_id}_{ch}.png\")\n            try:\n                arr = np.array(Image.open(img_path).convert(\"L\"))\n                feats.extend(compute_stats(arr))\n            except Exception:\n                feats.extend([0.0] * 9)\n        rows.append(feats)\n    \n    col_names = [f\"{ch}_{s}\" for ch in CHANNELS_ORDERED for s in STATS_KEYS]\n    return pd.DataFrame(rows, columns=col_names)\n\nprint(f\"Feature vector length: {len(CHANNELS_ORDERED) * len(STATS_KEYS)}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T04:59:45.727279Z","iopub.execute_input":"2026-08-19T04:59:45.727554Z","iopub.status.idle":"2026-08-19T04:59:45.798258Z","shell.execute_reply.started":"2026-08-19T04:59:45.727531Z","shell.execute_reply":"2026-08-19T04:59:45.797688Z"}},"outputs":[],"execution_count":null},{"id":"n1o2p3q4","cell_type":"markdown","source":"## ⚙️ Cell 7 — Build Training Feature DataFrame","metadata":{}},{"id":"o1p2q3r4","cell_type":"code","source":"# Cell 7 — Build training Spark DataFrame from pandas features\nfrom pyspark.ml.linalg import Vectors, VectorUDT\nfrom pyspark.sql.functions import udf as spark_udf, col\nfrom pyspark.sql.types import DoubleType\n\ntrain_ids     = train_pd[\"Id\"].tolist()\ntrain_targets = train_pd[\"Target\"].tolist()\n\nprint(\"Extracting training features on driver ...\")\ntrain_feats_pd = extract_features_pandas(train_ids, TRAIN_DIR)\ntrain_feats_pd[\"Id\"]     = train_ids\ntrain_feats_pd[\"Target\"] = train_targets\n\n# Convert to Spark DataFrame\ntrain_spark_raw = spark.createDataFrame(train_feats_pd)\n\n# Assemble feature columns into a DenseVector\nfrom pyspark.ml.feature import VectorAssembler\nfeat_cols = [f\"{ch}_{s}\" for ch in CHANNELS_ORDERED for s in STATS_KEYS]\n\nassembler = VectorAssembler(inputCols=feat_cols, outputCol=\"features\")\ntrain_feat_df = assembler.transform(train_spark_raw)\ntrain_feat_df.cache()\n\nprint(f\"✅ Training features ready: {train_feat_df.count()} rows × {len(feat_cols)} features\")\ntrain_feat_df.select(\"Id\", \"Target\", \"features\").show(3, truncate=70)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T05:35:55.706185Z","iopub.execute_input":"2026-08-19T05:35:55.706690Z","iopub.status.idle":"2026-08-19T06:13:04.271931Z","shell.execute_reply.started":"2026-08-19T05:35:55.706661Z","shell.execute_reply":"2026-08-19T06:13:04.271234Z"}},"outputs":[],"execution_count":null},{"id":"p1q2r3s4","cell_type":"markdown","source":"## Cell 8 — Extract Primary Label (Single Class per Sample)","metadata":{}},{"id":"q1r2s3t4","cell_type":"code","source":"# Cell 8 — Use primary label for multiclass classification\nfrom pyspark.sql.functions import col, split, udf\nfrom pyspark.sql.types import IntegerType\n\n@spark_udf(IntegerType())\ndef get_primary_label(target_str):\n    \"\"\"Take the first label in Target string as the primary class.\"\"\"\n    return int(str(target_str).split()[0])\n\ntrain_feat_df = train_feat_df.withColumn(\"label\", get_primary_label(col(\"Target\")))\n\nprint(\"Primary label distribution (top 10):\")\ntrain_feat_df.groupBy(\"label\").count().orderBy(\"count\", ascending=False).show(10)\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:38:23.825678Z","iopub.execute_input":"2026-08-19T06:38:23.825952Z","iopub.status.idle":"2026-08-19T06:38:24.850375Z","shell.execute_reply.started":"2026-08-19T06:38:23.825930Z","shell.execute_reply":"2026-08-19T06:38:24.849259Z"}},"outputs":[],"execution_count":null},{"id":"r1s2t3u4","cell_type":"markdown","source":"## ✂️ Cell 9 — Train / Validation Split","metadata":{}},{"id":"s1t2u3v4","cell_type":"code","source":"# Cell 9 — 80/20 split\ntrain_split, val_split = train_feat_df.randomSplit([0.8, 0.2], seed=42)\ntrain_split.cache()\nval_split.cache()\nprint(f\"Training   : {train_split.count()} samples\")\nprint(f\"Validation : {val_split.count()} samples\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:38:39.810897Z","iopub.execute_input":"2026-08-19T06:38:39.811297Z","iopub.status.idle":"2026-08-19T06:38:41.979783Z","shell.execute_reply.started":"2026-08-19T06:38:39.811270Z","shell.execute_reply":"2026-08-19T06:38:41.978507Z"}},"outputs":[],"execution_count":null},{"id":"t1u2v3w4","cell_type":"markdown","source":"## Cell 10 — 5 Classifier Pipelines","metadata":{}},{"id":"u1v2w3x4","cell_type":"code","source":"# Cell 10 — Define 5 classifier pipelines\nfrom pyspark.ml.feature import StandardScaler, MinMaxScaler\nfrom pyspark.ml.classification import (\n    RandomForestClassifier,\n    LogisticRegression,\n    DecisionTreeClassifier,\n    NaiveBayes,\n    MultilayerPerceptronClassifier,\n)\nfrom pyspark.ml import Pipeline\n\n# StandardScaler for most classifiers\nstd_scaler = StandardScaler(\n    inputCol=\"features\", outputCol=\"scaled_features\",\n    withMean=True, withStd=True\n)\n\n# MinMaxScaler for NaiveBayes (requires non-negative features)\nmm_scaler = MinMaxScaler(\n    inputCol=\"features\", outputCol=\"scaled_features\"\n)\n\n# ── 1. Random Forest ───────────────────────────────────────────────────────────\nrf = RandomForestClassifier(\n    featuresCol=\"scaled_features\", labelCol=\"label\",\n    numTrees=100, maxDepth=10, seed=42,\n    subsamplingRate=0.8, featureSubsetStrategy=\"sqrt\"\n)\n\n# ── 2. Logistic Regression ─────────────────────────────────────────────────────\nlr = LogisticRegression(\n    featuresCol=\"scaled_features\", labelCol=\"label\",\n    maxIter=100, regParam=0.1, elasticNetParam=0.0,\n    family=\"multinomial\"       # multiclass\n)\n\n# ── 3. Decision Tree ───────────────────────────────────────────────────────────\ndt = DecisionTreeClassifier(\n    featuresCol=\"scaled_features\", labelCol=\"label\",\n    maxDepth=10, seed=42\n)\n\n# ── 4. Naive Bayes — Gaussian (works with any real-valued features) ────────────\nnb = NaiveBayes(\n    featuresCol=\"scaled_features\", labelCol=\"label\",\n    smoothing=1.0,\n    modelType=\"gaussian\"    # ← changed from \"multinomial\"\n)\n\n# Use StandardScaler instead of MinMaxScaler now\nclassifiers = {\n    \"Random Forest\"          : Pipeline(stages=[std_scaler, rf]),\n    \"Logistic Regression\"    : Pipeline(stages=[std_scaler, lr]),\n    \"Decision Tree\"          : Pipeline(stages=[std_scaler, dt]),\n    \"Naive Bayes (Gaussian)\" : Pipeline(stages=[std_scaler, nb]),  # ← std_scaler now\n    \"Neural Network (MLP)\"   : Pipeline(stages=[std_scaler, mlp]),\n}\n\n\n# ── 5. Multilayer Perceptron (Neural Network) ──────────────────────────────────\n# layers: [input_size, hidden1, hidden2, num_classes]\nn_features = len(assembler.getInputCols())\nmlp = MultilayerPerceptronClassifier(\n    featuresCol=\"scaled_features\", labelCol=\"label\",\n    layers=[n_features, 64, 32, 28],\n    maxIter=100, seed=42, blockSize=128\n)\n\n# ── Pipelines ──────────────────────────────────────────────────────────────────\nclassifiers = {\n    \"Random Forest\"          : Pipeline(stages=[std_scaler, rf]),\n    \"Logistic Regression\"    : Pipeline(stages=[std_scaler, lr]),\n    \"Decision Tree\"          : Pipeline(stages=[std_scaler, dt]),\n    \"Naive Bayes\"            : Pipeline(stages=[mm_scaler,  nb]),  # MinMax\n    \"Neural Network (MLP)\"   : Pipeline(stages=[std_scaler, mlp]),\n}\n\nprint(f\"Classifiers defined: {list(classifiers.keys())}\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:58:05.495877Z","iopub.execute_input":"2026-08-19T06:58:05.496652Z","iopub.status.idle":"2026-08-19T06:58:05.526948Z","shell.execute_reply.started":"2026-08-19T06:58:05.496622Z","shell.execute_reply":"2026-08-19T06:58:05.526283Z"}},"outputs":[],"execution_count":null},{"id":"v1w2x3y4","cell_type":"markdown","source":"## Cell 11 — Train All 5","metadata":{}},{"id":"w1x2y3z4","cell_type":"code","source":"# Cell 11 — Train all 5 classifiers\nimport time\n\ntrained_models = {}\n\nfor name, pipeline in classifiers.items():\n    print(f\"\\n[Training] {name} ...\")\n    t0 = time.time()\n    trained_models[name] = pipeline.fit(train_split)\n    elapsed = time.time() - t0\n    print(f\"  ✅ Done in {elapsed:.1f}s\")\n\nprint(\"\\nAll classifiers trained.\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:58:10.488075Z","iopub.execute_input":"2026-08-19T06:58:10.488873Z","iopub.status.idle":"2026-08-19T06:58:49.160259Z","shell.execute_reply.started":"2026-08-19T06:58:10.488847Z","shell.execute_reply":"2026-08-19T06:58:49.159408Z"}},"outputs":[],"execution_count":null},{"id":"x1y2z3a4","cell_type":"markdown","source":"## Cell 12 — Evaluate All 5 (Accuracy, Precision, Recall, F1)","metadata":{}},{"id":"y1z2a3b4","cell_type":"code","source":"# Cell 12 — Evaluate all classifiers and build comparison table\nfrom pyspark.ml.evaluation import MulticlassClassificationEvaluator\n\nmetrics = [\"accuracy\", \"f1\", \"weightedPrecision\", \"weightedRecall\"]\nmetric_labels = [\"Accuracy\", \"F1 Score\", \"Precision\", \"Recall\"]\n\ncomparison = []\n\nfor name, model in trained_models.items():\n    print(f\"Evaluating: {name} ...\")\n    preds = model.transform(val_split)\n\n    row = {\"Classifier\": name}\n    for metric, label in zip(metrics, metric_labels):\n        evaluator = MulticlassClassificationEvaluator(\n            labelCol=\"label\",\n            predictionCol=\"prediction\",\n            metricName=metric\n        )\n        row[label] = round(evaluator.evaluate(preds), 4)\n\n    comparison.append(row)\n    print(f\"  Accuracy={row['Accuracy']:.4f}  F1={row['F1 Score']:.4f}  \"\n          f\"Precision={row['Precision']:.4f}  Recall={row['Recall']:.4f}\")\n\ncomparison_df = pd.DataFrame(comparison).set_index(\"Classifier\")\nprint(\"\\n\" + \"=\"*70)\nprint(\"CLASSIFIER COMPARISON TABLE\")\nprint(\"=\"*70)\nprint(comparison_df.to_string())\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:58:49.161618Z","iopub.execute_input":"2026-08-19T06:58:49.161861Z","iopub.status.idle":"2026-08-19T06:58:56.513363Z","shell.execute_reply.started":"2026-08-19T06:58:49.161840Z","shell.execute_reply":"2026-08-19T06:58:56.512549Z"}},"outputs":[],"execution_count":null},{"id":"z1a2b3c4","cell_type":"markdown","source":"## Cell 13 — Comparison Chart","metadata":{}},{"id":"a2b3c4d5","cell_type":"code","source":"# Cell 13 — Side-by-side bar chart comparison\nfig, axes = plt.subplots(2, 2, figsize=(16, 12))\naxes = axes.flatten()\n\ncolors = [\"#4C72B0\", \"#DD8452\", \"#55A868\", \"#C44E52\", \"#8172B2\"]\n\nfor ax, metric in zip(axes, metric_labels):\n    values = comparison_df[metric].values\n    bars = ax.bar(comparison_df.index, values, color=colors, edgecolor='white', linewidth=1.2)\n    ax.set_title(f\"{metric}\", fontsize=13, fontweight='bold')\n    ax.set_ylabel(metric, fontsize=11)\n    ax.set_ylim(0, min(1.0, max(values) * 1.25))\n    ax.set_xticklabels(comparison_df.index, rotation=20, ha='right', fontsize=9)\n    ax.grid(axis='y', linestyle='--', alpha=0.5)\n\n    # Value labels on bars\n    for bar, val in zip(bars, values):\n        ax.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.005,\n                f\"{val:.3f}\", ha='center', va='bottom', fontsize=9, fontweight='bold')\n\nplt.suptitle(\"Classifier Performance Comparison\\n(28-Class Protein Organelle Classification)\",\n             fontsize=15, fontweight='bold', y=1.01)\nplt.tight_layout()\nplt.savefig(\"/kaggle/working/classifier_comparison.png\", dpi=120, bbox_inches='tight')\nplt.show()\n\n# Best classifier per metric\nprint(\"\\nBest Classifier per Metric:\")\nfor metric in metric_labels:\n    best = comparison_df[metric].idxmax()\n    print(f\"  {metric:<15s}: {best}  ({comparison_df.loc[best, metric]:.4f})\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:59:15.528057Z","iopub.execute_input":"2026-08-19T06:59:15.528314Z","iopub.status.idle":"2026-08-19T06:59:16.605399Z","shell.execute_reply.started":"2026-08-19T06:59:15.528293Z","shell.execute_reply":"2026-08-19T06:59:16.604680Z"}},"outputs":[],"execution_count":null},{"id":"b2c3d4e5","cell_type":"markdown","source":"# Part 2(Timing Analysis)","metadata":{}},{"id":"b7368a1c-ba15-495f-8d0c-e2dcdfa92906","cell_type":"markdown","source":"## Cell 14 — CrossValidator Timing (vary setParallelism)","metadata":{}},{"id":"c2d3e4f5","cell_type":"code","source":"# Cell 14 — Timing: CrossValidator setParallelism (1, 2, 4)\nfrom pyspark.ml.tuning import CrossValidator, ParamGridBuilder\nfrom pyspark.ml.evaluation import MulticlassClassificationEvaluator\nfrom pyspark.ml.classification import LogisticRegression\nfrom pyspark.ml.feature import StandardScaler\nfrom pyspark.ml import Pipeline\nimport time\n\n# Use Logistic Regression (fast) for timing experiment\nscaler_cv = StandardScaler(\n    inputCol=\"features\", outputCol=\"scaled_features\",\n    withMean=True, withStd=True\n)\nlr_cv = LogisticRegression(\n    featuresCol=\"scaled_features\", labelCol=\"label\",\n    family=\"multinomial\", maxIter=20\n)\npipeline_cv = Pipeline(stages=[scaler_cv, lr_cv])\n\n# Small param grid (2 values → 2 × 3 folds = 6 fits)\nparamGrid = (\n    ParamGridBuilder()\n    .addGrid(lr_cv.regParam, [0.01, 0.1])\n    .build()\n)\n\nevaluator_cv = MulticlassClassificationEvaluator(\n    labelCol=\"label\", predictionCol=\"prediction\", metricName=\"accuracy\"\n)\n\nparallelism_values = [1, 2, 4]\ncv_timing = {}\n\nfor p in parallelism_values:\n    print(f\"\\n[CrossValidator] setParallelism={p} ...\")\n    cv = (\n        CrossValidator(\n            estimator=pipeline_cv,\n            estimatorParamMaps=paramGrid,\n            evaluator=evaluator_cv,\n            numFolds=3\n        )\n        .setParallelism(p)\n    )\n    t0 = time.time()\n    cv_model = cv.fit(train_split)\n    elapsed = time.time() - t0\n    best_acc = max(cv_model.avgMetrics)\n    cv_timing[p] = elapsed\n    print(f\"  ✅  Time={elapsed:.2f}s  |  Best Accuracy={best_acc:.4f}\")\n\nprint(\"\\nCrossValidator Timing Summary:\")\nfor p, t in cv_timing.items():\n    print(f\"  setParallelism={p}: {t:.2f}s\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T07:00:41.715552Z","iopub.execute_input":"2026-08-19T07:00:41.716084Z","iopub.status.idle":"2026-08-19T07:01:46.130357Z","shell.execute_reply.started":"2026-08-19T07:00:41.716057Z","shell.execute_reply":"2026-08-19T07:01:46.129428Z"}},"outputs":[],"execution_count":null},{"id":"d2e3f4g5","cell_type":"markdown","source":"## Cell 15 — Shuffle Partitions Timing (vary spark.sql.shuffle.partitions)","metadata":{}},{"id":"e2f3g4h5","cell_type":"code","source":"# Cell 15 — Timing: spark.sql.shuffle.partitions (50, 100, 200)\nfrom pyspark.ml.feature import StandardScaler\nfrom pyspark.ml.classification import RandomForestClassifier\nfrom pyspark.ml import Pipeline\nimport time\n\npartition_values = [50, 100, 200]\nshuffle_timing = {}\n\nfor partitions in partition_values:\n    # Set shuffle partitions\n    spark.conf.set(\"spark.sql.shuffle.partitions\", str(partitions))\n    print(f\"\\n[Shuffle Partitions={partitions}] Training RandomForest ...\")\n\n    scaler_s = StandardScaler(\n        inputCol=\"features\", outputCol=\"scaled_features\",\n        withMean=True, withStd=True\n    )\n    rf_s = RandomForestClassifier(\n        featuresCol=\"scaled_features\", labelCol=\"label\",\n        numTrees=50, maxDepth=6, seed=42\n    )\n    pipeline_s = Pipeline(stages=[scaler_s, rf_s])\n\n    t0 = time.time()\n    model_s = pipeline_s.fit(train_split)\n    preds_s  = model_s.transform(val_split)\n\n    # Force evaluation to complete\n    evaluator_s = MulticlassClassificationEvaluator(\n        labelCol=\"label\", predictionCol=\"prediction\", metricName=\"accuracy\"\n    )\n    acc = evaluator_s.evaluate(preds_s)\n    elapsed = time.time() - t0\n\n    shuffle_timing[partitions] = elapsed\n    print(f\"  ✅  Time={elapsed:.2f}s  |  Accuracy={acc:.4f}\")\n\n# Reset to default\nspark.conf.set(\"spark.sql.shuffle.partitions\", \"50\")\n\nprint(\"\\nShuffle Partitions Timing Summary:\")\nfor p, t in shuffle_timing.items():\n    print(f\"  shuffle.partitions={p}: {t:.2f}s\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T07:02:19.529785Z","iopub.execute_input":"2026-08-19T07:02:19.530283Z","iopub.status.idle":"2026-08-19T07:02:28.710017Z","shell.execute_reply.started":"2026-08-19T07:02:19.530257Z","shell.execute_reply":"2026-08-19T07:02:28.709174Z"}},"outputs":[],"execution_count":null},{"id":"f2g3h4i5","cell_type":"markdown","source":"## Cell 16 — Default Parallelism Timing (vary spark.default.parallelism)","metadata":{}},{"id":"g2h3i4j5","cell_type":"code","source":"# Cell 16 — Timing: spark.default.parallelism (50, 100, 200)\nimport time\n\nparallelism_vals = [50, 100, 200]\ndefault_par_timing = {}\n\nfor par in parallelism_vals:\n    spark.conf.set(\"spark.default.parallelism\", str(par))\n    print(f\"\\n[Default Parallelism={par}] Training RandomForest ...\")\n\n    scaler_p = StandardScaler(\n        inputCol=\"features\", outputCol=\"scaled_features\",\n        withMean=True, withStd=True\n    )\n    rf_p = RandomForestClassifier(\n        featuresCol=\"scaled_features\", labelCol=\"label\",\n        numTrees=50, maxDepth=6, seed=42\n    )\n    pipeline_p = Pipeline(stages=[scaler_p, rf_p])\n\n    t0 = time.time()\n    model_p = pipeline_p.fit(train_split)\n    preds_p  = model_p.transform(val_split)\n    evaluator_p = MulticlassClassificationEvaluator(\n        labelCol=\"label\", predictionCol=\"prediction\", metricName=\"accuracy\"\n    )\n    acc = evaluator_p.evaluate(preds_p)\n    elapsed = time.time() - t0\n\n    default_par_timing[par] = elapsed\n    print(f\"  ✅  Time={elapsed:.2f}s  |  Accuracy={acc:.4f}\")\n\n# Reset to default\nspark.conf.set(\"spark.default.parallelism\", \"100\")\n\nprint(\"\\nDefault Parallelism Timing Summary:\")\nfor p, t in default_par_timing.items():\n    print(f\"  default.parallelism={p}: {t:.2f}s\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T07:03:11.370635Z","iopub.execute_input":"2026-08-19T07:03:11.371034Z","iopub.status.idle":"2026-08-19T07:03:21.057779Z","shell.execute_reply.started":"2026-08-19T07:03:11.371009Z","shell.execute_reply":"2026-08-19T07:03:21.056897Z"}},"outputs":[],"execution_count":null},{"id":"h2i3j4k5","cell_type":"markdown","source":"## Cell 17 — Visualize All Timing Results","metadata":{}},{"id":"i2j3k4l5","cell_type":"code","source":"# Cell 17 — Timing comparison charts\nfig, axes = plt.subplots(1, 3, figsize=(18, 5))\n\n# ── 1. CrossValidator Parallelism ─────────────────────────────────────────────\nax = axes[0]\nax.plot(list(cv_timing.keys()), list(cv_timing.values()),\n        marker='o', color='#4C72B0', linewidth=2.5, markersize=8)\nfor x, y in cv_timing.items():\n    ax.annotate(f\"{y:.1f}s\", (x, y), textcoords=\"offset points\",\n                xytext=(0, 10), ha='center', fontsize=10)\nax.set_xlabel(\"setParallelism Value\", fontsize=12)\nax.set_ylabel(\"Training Time (seconds)\", fontsize=12)\nax.set_title(\"CrossValidator\\nsetParallelism vs Time\", fontsize=13, fontweight='bold')\nax.set_xticks(list(cv_timing.keys()))\nax.grid(linestyle='--', alpha=0.5)\n\n# ── 2. Shuffle Partitions ─────────────────────────────────────────────────────\nax = axes[1]\nax.plot(list(shuffle_timing.keys()), list(shuffle_timing.values()),\n        marker='s', color='#DD8452', linewidth=2.5, markersize=8)\nfor x, y in shuffle_timing.items():\n    ax.annotate(f\"{y:.1f}s\", (x, y), textcoords=\"offset points\",\n                xytext=(0, 10), ha='center', fontsize=10)\nax.set_xlabel(\"spark.sql.shuffle.partitions\", fontsize=12)\nax.set_ylabel(\"Training Time (seconds)\", fontsize=12)\nax.set_title(\"Shuffle Partitions\\nvs Training Time\", fontsize=13, fontweight='bold')\nax.set_xticks(list(shuffle_timing.keys()))\nax.grid(linestyle='--', alpha=0.5)\n\n# ── 3. Default Parallelism ────────────────────────────────────────────────────\nax = axes[2]\nax.plot(list(default_par_timing.keys()), list(default_par_timing.values()),\n        marker='^', color='#55A868', linewidth=2.5, markersize=8)\nfor x, y in default_par_timing.items():\n    ax.annotate(f\"{y:.1f}s\", (x, y), textcoords=\"offset points\",\n                xytext=(0, 10), ha='center', fontsize=10)\nax.set_xlabel(\"spark.default.parallelism\", fontsize=12)\nax.set_ylabel(\"Training Time (seconds)\", fontsize=12)\nax.set_title(\"Default Parallelism\\nvs Training Time\", fontsize=13, fontweight='bold')\nax.set_xticks(list(default_par_timing.keys()))\nax.grid(linestyle='--', alpha=0.5)\n\nplt.suptitle(\"Timing Analysis — Effect of Parallelism & Partition Settings\",\n             fontsize=14, fontweight='bold', y=1.03)\nplt.tight_layout()\nplt.savefig(\"/kaggle/working/timing_analysis.png\", dpi=120, bbox_inches='tight')\nplt.show()\n\n# Summary table\ntiming_summary = pd.DataFrame({\n    \"Setting\"      : [\"setParallelism=1\", \"setParallelism=2\", \"setParallelism=4\",\n                      \"shuffle.partitions=50\", \"shuffle.partitions=100\", \"shuffle.partitions=200\",\n                      \"default.parallelism=50\", \"default.parallelism=100\", \"default.parallelism=200\"],\n    \"Time (s)\"     : list(cv_timing.values()) +\n                     list(shuffle_timing.values()) +\n                     list(default_par_timing.values()),\n    \"Category\"     : [\"CrossValidator\"]*3 + [\"Shuffle Partitions\"]*3 + [\"Default Parallelism\"]*3\n})\nprint(\"\\nFull Timing Summary:\")\nprint(timing_summary.to_string(index=False))\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T07:03:47.606150Z","iopub.execute_input":"2026-08-19T07:03:47.606766Z","iopub.status.idle":"2026-08-19T07:03:48.374788Z","shell.execute_reply.started":"2026-08-19T07:03:47.606740Z","shell.execute_reply":"2026-08-19T07:03:48.374187Z"}},"outputs":[],"execution_count":null},{"id":"j2k3l4m5","cell_type":"markdown","source":"## ✅ Cell 18 — Submission Validation","metadata":{}},{"id":"k2l3m4n5","cell_type":"code","source":"# Cell 18 — Validate submission\nfinal_sub = pd.read_csv(SUBMIT_PATH)\nsample_ids_test = pd.read_csv(SAMPLE_SUB)[\"Id\"].tolist()\n\nmissing_ids = set(sample_ids_test) - set(final_sub[\"Id\"])\ninvalid = [\n    (row[\"Id\"], row[\"Target\"])\n    for _, row in final_sub.iterrows()\n    if not str(row[\"Target\"]).isdigit() or int(row[\"Target\"]) not in range(28)\n]\n\nprint(f\"Total rows    : {len(final_sub)}\")\nprint(f\"Expected rows : {len(sample_ids_test)}\")\nprint(f\"Missing IDs   : {len(missing_ids)}\")\nprint(f\"Invalid labels: {len(invalid)}\")\n\nif not missing_ids and not invalid:\n    print(\"\\n🎉 Submission VALID — ready to submit!\")\nelse:\n    print(\"\\n⚠️ Issues found — review before submitting.\")\n","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:55:09.346767Z","iopub.status.idle":"2026-08-19T06:55:09.347023Z","shell.execute_reply.started":"2026-08-19T06:55:09.346911Z","shell.execute_reply":"2026-08-19T06:55:09.346924Z"}},"outputs":[],"execution_count":null},{"id":"l2m3n4o5","cell_type":"markdown","source":"## 🛑 Cell 19 — Stop Spark Session","metadata":{}},{"id":"m2n3o4p5","cell_type":"code","source":"spark.stop()\nprint(\"Spark session stopped.\")","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2026-08-19T06:55:09.347913Z","iopub.status.idle":"2026-08-19T06:55:09.348161Z","shell.execute_reply.started":"2026-08-19T06:55:09.348050Z","shell.execute_reply":"2026-08-19T06:55:09.348064Z"}},"outputs":[],"execution_count":null}]}