{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# 1. Treeni ilmamudel ja salvesta katsed Databricksis\n", "\n", "**Eesmärk:** treenid kaks närvivõrku, võrdled neid ja valid ühe kasutamiseks.\n", "Närvivõrk on mudel, mis õpib näidete põhjal sisendi ja vastuse vahelist seost. Siin ennustab ta järgmist päeva.\n", "\n", "Enne: järgi `index.html` samme 1–3. Vali **Serverless** ehk Databricksi hallatav arvutusressurss.\n", "Lisa **Environment → Dependencies** alla: `mlflow>=3.3,<4`, `scikit-learn>=1.5,<2`, `pandas>=2.2,<3`, `numpy>=1.26,<3`, `requests>=2.31,<3`, `cloudpickle>=3,<4`. Vajuta **Apply**.\n", "Dependency ehk sõltuvus on lisapakett: valmis kood, mida meie programm vajab.\n", "\n", "**Käivita lahtrid ükshaaval ülevalt alla.** Koodilahter on kast, milles on arvutile antavad käsud. Käivitamiseks vajuta lahtri kolmnurka või Shift+Enter.\n", "Ära kirjuta siia paroole ega juurdepääsutõendeid." ], "id": "cell-000" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. Tööriistad\n", "`import` võtab valmis tööriista kasutusele. `pandas` töötleb tabeleid, `numpy` arvutab arvudega ja `sklearn` sisaldab masinõppe tööriistu.\n", "`mlflow` salvestab katsed. `MlflowClient` on tööriist registri juhtimiseks.\n", "`tracking_uri` ütleb, kuhu katsed salvestada. `registry_uri` valib mudeliregistri. Databricksis pole vaja aadressi `localhost:5000`.\n", "`autolog(disable=True)` lülitab automaatse logimise välja, sest kirjutame kõik vajalikud salvestuskäsud ise." ], "id": "cell-001" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "import json, re, hashlib\n", "from pathlib import Path\n", "from datetime import datetime, timedelta\n", "from zoneinfo import ZoneInfo\n", "import numpy as np\n", "import pandas as pd\n", "import sklearn, mlflow\n", "from mlflow import MlflowClient\n", "from mlflow.models import infer_signature\n", "from sklearn.neural_network import MLPRegressor\n", "from sklearn.preprocessing import StandardScaler\n", "from sklearn.pipeline import Pipeline\n", "from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score\n", "\n", "mlflow.set_tracking_uri(\"databricks\")\n", "mlflow.set_registry_uri(\"databricks-uc\")\n", "mlflow.autolog(disable=True)\n", "print(\"MLflow:\", mlflow.__version__, \"| scikit-learn:\", sklearn.__version__)" ], "id": "cell-002" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. Sinu failide asukoht\n", "Muuda järgmises lahtris `VOLUME` väärtus enda failikausta teeks. **Path** tähendab failitee kirjeldust.\n", "Näites on `workspace` kataloog, `ilm_anna` skeem ehk projekti jaotis ja `andmed` volume ehk failikaust.\n", "Kood loeb kataloogi ja skeemi nimest välja, kuhu mudel salvestada. Nii salvestatakse sinu mudelid valitud jaotisse." ], "id": "cell-003" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "# MUUDA ainult seda teed: Catalog → sinu volume → Copy path.\n", "VOLUME = Path(\"/Volumes/workspace/ilm_anna/andmed\")\n", "if not VOLUME.is_dir():\n", " raise ValueError(\"Seda failikausta ei leitud. Kopeeri oma volume'i tee VOLUME väärtuseks.\")\n", "osad = VOLUME.parts\n", "if len(osad) != 5 or osad[1] != \"Volumes\":\n", " raise ValueError(\"Kasuta volume'i juurkausta: /Volumes/kataloog/skeem/volume\")\n", "CATALOG, SCHEMA = osad[2], osad[3]\n", "MODEL_NAME = f\"{CATALOG}.{SCHEMA}.stockholmi_ilm\"\n", "ENDPOINT = f\"stockholmi-ilm-{SCHEMA.replace('_','-')}\"\n", "print(\"Mudeli nimi:\", MODEL_NAME)\n", "print(\"Teenuse soovituslik nimi:\", ENDPOINT)" ], "id": "cell-004" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. Loe andmed\n", "**CSV** on tekstifail, kus andmed on ridade ja veergudena. **DataFrame** on sama tabel Pythoni mälus.\n", "`read_csv` loeb faili; `head` näitab esimesi ridu. `DATE` on kuupäev. `STOCKHOLM_temp_mean` on Stockholmi päeva keskmine temperatuur Celsiuse kraadides.\n", "Kontroll: tabelis on palju päevi ja kuupäevad näevad välja nagu `20000101`." ], "id": "cell-005" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "df = pd.read_csv(VOLUME / \"weather_prediction_dataset.csv\")\n", "print(\"Päevi:\", len(df), \"| veerge:\", len(df.columns))\n", "display(df.head())" ], "id": "cell-006" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. Millest mudel õpib?\n", "**Tunnus** on üks sisendarv, näiteks Oslo eilne temperatuur. **Sihtmuutuja** on vastus, mida tahame ennustada.\n", "`shift(-1)` toob järgmise päeva vastuse tänase rea juurde. `shift(i)` toob varasema päeva tunnuse tänase rea juurde.\n", "**Lag** tähendab ajalist nihet: `day-0` on lähtepäev, `day-1` eelmine päev. Kasutame nelja päeva.\n", "**Korrelatsioon** kirjeldab arvude koos muutumist; see ei tõesta põhjuslikku seost. Valime tunnuseid ainult treeningandmete abil.\n", "\n", "Andmed jagunevad ajaliselt kolmeks: varaseimad 70% õppimiseks (**train**), järgmised 15% kahe mudeli võrdlemiseks (**validation**) ja viimased 15% lõplikuks kontrolliks (**test**).\n", "Protsendid on ligikaudsed, sest esimestel päevadel pole nelja päeva ajalugu ja viimasel päeval pole homset vastust.\n", "**Andmeleke** tekiks, kui kasutaksime õppimisel tulevase kontrollosa vastuseid. Siin hoiame need eraldi.\n", "\n", "Järgmine lahter defineerib funktsiooni ehk nimega tööjuhise. Lahter pärast seda kasutab seda juhist.\n", "Valime ainult temperatuuriveerud: neid oskame hiljem Open-Meteost küsida. Tunnuste arv arvutatakse, see ei pea olema täpselt 40." ], "id": "cell-007" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "def prepare_data(df, lags=4, threshold=0.85):\n", " \"\"\"Create features using only the training interval for feature selection.\"\"\"\n", " target = 'STOCKHOLM_temp_mean'\n", " if not {'DATE', target}.issubset(df.columns):\n", " raise ValueError('CSV peab sisaldama veerge DATE ja STOCKHOLM_temp_mean.')\n", " df = df.copy()\n", " df['DATE'] = pd.to_datetime(df['DATE'].astype(str), format='%Y%m%d')\n", " df = df.sort_values('DATE').set_index('DATE')\n", " if df.index.has_duplicates or not df.index.to_series().diff().dropna().eq(pd.Timedelta(days=1)).all():\n", " raise ValueError('Kuupäevad peavad olema järjest: iga päev üks rida, ilma vahedeta.')\n", " if len(df) < 100:\n", " raise ValueError('Selles harjutuses kasuta vähemalt 100 päeva andmeid.')\n", " # Valime ainult temperatuurid, mida oskame ka Open-Meteost küsida.\n", " columns = [c for c in df if re.fullmatch(r'.+_temp_(mean|max|min)', c) and c != target]\n", " raw = df[columns].apply(pd.to_numeric, errors='raise').astype(float)\n", " y = pd.to_numeric(df[target], errors='raise').shift(-1).rename('homne_temperatuur')\n", " label_day = pd.Series(df.index + pd.Timedelta(days=1), index=df.index)\n", " validation_start = df.index[int(len(df) * 0.70)]\n", " test_start = df.index[int(len(df) * 0.85)]\n", " training_mask = label_day < validation_start\n", " correlations = raw.loc[training_mask].corrwith(y.loc[training_mask]).abs()\n", " selected = correlations[correlations > 0.3].index.tolist()\n", " if not selected:\n", " raise ValueError('Ükski temperatuuriveerg ei läbinud esimest tunnuste valikut.')\n", " all_features = pd.concat([\n", " raw[selected].shift(i).add_suffix(f'_day-{i}') for i in range(lags)\n", " ], axis=1)\n", " correlations = all_features.loc[training_mask].corrwith(y.loc[training_mask]).abs()\n", " features = correlations[correlations > threshold].sort_values(ascending=False).index.tolist()\n", " if not features:\n", " raise ValueError('Ükski tunnus ei läbinud korrelatsioonipiiri. Kontrolli andmeid või vähenda piiri.')\n", " X = all_features[features]\n", " valid = X.notna().all(axis=1) & y.notna() & np.isfinite(X).all(axis=1) & np.isfinite(y)\n", " masks = {\n", " 'train': valid & (label_day < validation_start),\n", " 'validation': valid & (label_day >= validation_start) & (label_day < test_start),\n", " 'test': valid & (label_day >= test_start),\n", " }\n", " parts = {}\n", " for name, mask in masks.items():\n", " if not mask.any():\n", " raise ValueError(f'Andmeosas {name} ei ole kasutatavaid ridu.')\n", " parts[name] = {\n", " 'X': X.loc[mask].astype(float), 'y': y.loc[mask].astype(float),\n", " 'baseline': df.loc[mask, target].astype(float),\n", " 'target_dates': label_day.loc[mask],\n", " }\n", " return parts, features" ], "id": "cell-008" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "LAGS = 4\n", "KORR_LAVI = 0.85\n", "parts, tunnused = prepare_data(df, lags=LAGS, threshold=KORR_LAVI)\n", "print(\"Valitud tunnuste arv:\", len(tunnused))\n", "print(\"Esimesed tunnused:\", tunnused[:5])\n", "for nimi, osa in parts.items():\n", " print(nimi, len(osa[\"X\"]), \"rida; ennustatavad päevad:\",\n", " osa[\"target_dates\"].min().date(), \"kuni\", osa[\"target_dates\"].max().date())" ], "id": "cell-009" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. Katsete kaust\n", "**Experiment** on katsete kogum. **Run** on üks katse: üks treenimine koos seadete ja tulemustega.\n", "`current_user()` küsib Databricksilt sinu kasutajanime. `spark.sql` käivitab SQL-keeles päringu; SQL on tabelitelt küsimuste küsimise keel.\n", "Siin kasutame seda vaid oma kasutajanime leidmiseks, mitte mudeli treenimiseks. **Spark** on Databricksi andmetöötlusvahend.\n", "Eksperimendi tee algab sinu isiklikust Workspace-kaustast." ], "id": "cell-010" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "kasutaja = spark.sql(\"SELECT current_user() AS nimi\").first()[\"nimi\"]\n", "EXPERIMENT = f\"/Users/{kasutaja}/stockholmi-ilm\"\n", "mlflow.set_experiment(EXPERIMENT)\n", "print(\"Katsete asukoht:\", EXPERIMENT)" ], "id": "cell-011" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 6. Kaks treenimist\n", "**MLPRegressor** on närvivõrk, mille vastus on arv. **Regressioon** tähendabki arvu ennustamist.\n", "`hidden_layer_sizes=(50, 30)` tähendab kaht peidetud kihti: esimeses 50, teises 30 arvutusüksust ehk neuronit.\n", "`max_iter` seab õppimise korduste ülempiiri. `random_state=42` fikseerib juhusliku alguse, et katseid oleks lihtsam võrrelda.\n", "**StandardScaler** teisendab tunnused võrreldavale skaalale. **Pipeline** seob skaleerija ja mudeli üheks tervikuks, et sama teisendus toimuks ka ennustamisel.\n", "\n", "**Parameeter** on sinu valitud seade. **Mõõdik** on tulemuse hinne. **RMSE** mõõdab prognoosivea suurust ja karistab suuri vigu tugevamalt; väiksem on parem. **MAE** on absoluutvigade keskmine, samuti kraadides. **R²** võrdleb tulemust konstantse keskmise ennustamisega; 1 on parim ja tulemus võib olla ka negatiivne.\n", "**Võrdlusmudel** ütleb lihtsalt „homme on sama temperatuur kui täna“. Vaatame ka selle RMSE-d.\n", "\n", "**Artefakt** on katsega salvestatud fail. `tunnused.json` jätab meelde veerud ja linnad; JSON on programmidele mugav nime–väärtuse andmevorming.\n", "**Signature** kirjeldab mudeli sisendi veerge ja andmetüüpe ning väljundit. **Input example** on näidisrida.\n", "`log_model` salvestab mudeli; `register_model` teeb sellest registris nummerdatud versiooni. **Cloudpickle** on siin mudeli salvestusvorming; laeme ainult enda loodud mudeli.\n", "\n", "Oota, kuni mõlemad katsed lõpetavad. **ConvergenceWarning** tähendab, et õppimine jõudis korduste piirini; mudel võib olemas olla, aga hoiatus tuleks tulemust hinnates kirja panna." ], "id": "cell-012" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "def moodikud(tegelik, prognoos):\n", " return {\"rmse\": float(np.sqrt(mean_squared_error(tegelik, prognoos))),\n", " \"mae\": float(mean_absolute_error(tegelik, prognoos)),\n", " \"r2\": float(r2_score(tegelik, prognoos))}\n", "\n", "koord = pd.read_csv(VOLUME / \"linnade_koordinaadid.csv\")\n", "vajalikud_linnad = {re.fullmatch(r\"(.+)_(temp_mean|temp_max|temp_min)_day-\\d+\", t)[1] for t in tunnused}\n", "puuduvad = vajalikud_linnad - set(koord[\"linn\"])\n", "if puuduvad:\n", " raise ValueError(f\"Koordinaadifailist puuduvad linnad: {puuduvad}\")\n", "meta = {\"features\": tunnused, \"lags\": LAGS,\n", " \"coordinates\": koord.to_dict(orient=\"records\"), \"timezone\": \"Europe/Stockholm\"}\n", "train, val, test = parts[\"train\"], parts[\"validation\"], parts[\"test\"]\n", "tulemused = []\n", "mudelid = {}\n", "for kihid in [(50, 30), (100, 50)]:\n", " with mlflow.start_run(run_name=f\"mlp-{kihid[0]}-{kihid[1]}\") as run:\n", " mudel = Pipeline([\n", " (\"scaler\", StandardScaler()),\n", " (\"mlp\", MLPRegressor(hidden_layer_sizes=kihid, max_iter=1000, random_state=42)),\n", " ])\n", " mudel.fit(train[\"X\"], train[\"y\"])\n", " hinnang = moodikud(val[\"y\"], mudel.predict(val[\"X\"]))\n", " mlflow.log_params({\"hidden_layer_sizes\": str(kihid), \"lags\": LAGS,\n", " \"korr_lavi\": KORR_LAVI, \"tunnuste_arv\": len(tunnused),\n", " \"max_iter\": 1000, \"random_state\": 42})\n", " mlflow.log_metrics({f\"validation_{k}\":v for k,v in hinnang.items()})\n", " mlflow.log_metric(\"validation_baseline_rmse\", moodikud(val[\"y\"],val[\"baseline\"])[\"rmse\"])\n", " mlflow.log_dict(meta, \"tunnused.json\")\n", " mlflow.log_dict({\"data_sha256\": hashlib.sha256((VOLUME/\"weather_prediction_dataset.csv\").read_bytes()).hexdigest(),\n", " \"sklearn\": sklearn.__version__, \"mlflow\": mlflow.__version__,\n", " \"split\": \"chronological 70/15/15 by target date\"}, \"katse_info.json\")\n", " info = mlflow.sklearn.log_model(\n", " sk_model=mudel, name=\"mudel\", serialization_format=\"cloudpickle\",\n", " signature=infer_signature(train[\"X\"].head(5), mudel.predict(train[\"X\"].head(5))),\n", " input_example=train[\"X\"].head(1),\n", " )\n", " version = mlflow.register_model(info.model_uri, MODEL_NAME)\n", " r = {\"run_id\":run.info.run_id, \"kihid\":str(kihid), \"version\":str(version.version), **hinnang}\n", " tulemused.append(r)\n", " mudelid[str(version.version)] = mudel\n", "display(pd.DataFrame(tulemused))" ], "id": "cell-013" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 7. Vali mudel võrdlusosa järgi\n", "**Alias** on viit ühele versioonile. `champion` tähendab siin „kasutamiseks valitud mudel“.\n", "Valime väiksema validation-RMSE-ga mudeli. Seejärel mõõdame selle viga seni puutumata testandmetel.\n", "Testitulemust ei kasutata siin teise mudeli valimiseks. Viimane treenitud mudel ei ole automaatselt parim.\n", "`load_model` loeb salvestatud mudeli tagasi. See kontrollib salvestamist, kuid ei tõesta veel, et veebiteenus töötab." ], "id": "cell-014" }, { "cell_type": "code", "metadata": {}, "execution_count": null, "outputs": [], "source": [ "parim = min(tulemused, key=lambda r: r[\"rmse\"])\n", "client = MlflowClient()\n", "client.set_registered_model_alias(MODEL_NAME, \"champion\", parim[\"version\"])\n", "laetud = mlflow.pyfunc.load_model(f\"models:/{MODEL_NAME}/{parim['version']}\")\n", "test_hinnang = moodikud(test[\"y\"], laetud.predict(test[\"X\"]))\n", "for nimi,vaartus in test_hinnang.items():\n", " client.log_metric(parim[\"run_id\"], f\"test_{nimi}\", vaartus)\n", "client.log_metric(parim[\"run_id\"], \"test_baseline_rmse\", moodikud(test[\"y\"],test[\"baseline\"])[\"rmse\"])\n", "\n", "# Salvestame päringu, millega saab teenust ilma interneti ilmaandmeteta katsetada.\n", "naidis = {\"dataframe_split\": {\"columns\":tunnused, \"data\":test[\"X\"].head(1).values.tolist()}}\n", "(VOLUME/\"naidisparing.json\").write_text(json.dumps(naidis,ensure_ascii=False,indent=2),encoding=\"utf-8\")\n", "(VOLUME/\"projekt.json\").write_text(json.dumps({\"model_name\":MODEL_NAME,\"endpoint\":ENDPOINT},indent=2),encoding=\"utf-8\")\n", "print(\"Valitud versioon:\", parim[\"version\"])\n", "print(\"Lõpliku kontrolli tulemused:\", test_hinnang)\n", "print(\"Võrdlusmudeli RMSE:\", moodikud(test[\"y\"],test[\"baseline\"])[\"rmse\"])\n", "print(\"Teenuse loomisel vali mudel\", MODEL_NAME, \"ja versioon\", parim[\"version\"])\n", "print(\"Näidispäringu vastus selles notebookis:\", laetud.predict(test[\"X\"].head(1)))\n", "print(json.dumps(naidis,ensure_ascii=False))" ], "id": "cell-015" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Järgmine samm\n", "Mine tagasi `index.html` sammude 5–7 juurde: vaata katseid, loo **Serving endpoint** ehk mudeli veebiteenus ja saada `naidisparing.json` teenusele.\n", "Seejärel ava `02_tanane_prognoos.ipynb`.\n", "Kui muudad `champion` silti hiljem, tuleb teenuses eraldi mudeliversioon uuendada. Sildi muutmine ei uuenda juba töötavat teenust." ], "id": "cell-016" } ], "metadata": { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3" }, "language_info": { "name": "python" } }, "nbformat": 4, "nbformat_minor": 5 }