notebooks/analytics.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "222b4132-fc10-4503-a108-592d5e742515",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "from datetime import datetime\n",
    "import numpy as np\n",
    "import pandas as pd\n",
    "import flowcept.analytics as analytics\n",
    "import flowcept.analytics.plot as flow_plot\n",
    "from flowcept import TaskQueryAPI"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c7b11fbf-ec74-46e7-9824-4685a9288c55",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "def ingest_mock_data():\n",
    "    \"\"\"\n",
    "    This function is here just to enable the execution of the functions we are showing in this notebook.\n",
    "    \"\"\"\n",
    "    import json\n",
    "    from uuid import uuid4\n",
    "    from flowcept import Flowcept\n",
    "    test_data_path = '../tests/api/sample_data_with_telemetry_and_rai.json' # This sample data contains a workflow composed of 9 tasks.\n",
    "    with open(test_data_path) as f:\n",
    "        base_data = json.loads(f.read())\n",
    "    \n",
    "    docs = []\n",
    "    wf_id = str(uuid4())\n",
    "    for d in base_data:\n",
    "        new_doc = d.copy()\n",
    "        new_doc.pop(\"_id\")\n",
    "        new_doc[\"task_id\"] = str(uuid4())\n",
    "        new_doc[\"workflow_id\"] = wf_id\n",
    "        new_doc.pop(\"timestamp\", None)\n",
    "        docs.append(new_doc)\n",
    "    \n",
    "    inserted_ids = Flowcept.db._dao().insert_and_update_many_tasks(docs, \"task_id\")\n",
    "    #assert len(inserted_ids) == len(base_data)\n",
    "    return wf_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "176f01c5-5e59-44e3-ad65-409fcfdc2f9b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# Need to run only if this is the first time.\n",
    "wf_id = ingest_mock_data()\n",
    "wf_id"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "96442d46-7ebb-470d-962b-11b65e7aca12",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "#wf_id = '100faab4-ff4c-4f78-92a7-6f20ec1fad83'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e41fe652-d7e8-4e3d-a780-dfec4e5142b0",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "query_api = TaskQueryAPI()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "dad7b3e7-1637-4034-91e0-0f00d4d64941",
   "metadata": {},
   "source": [
    "## Very Simple query returning a DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "2c3cd6d6-fc22-4155-80e0-da7ffc9f8e0e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "_filter = {\n",
    "    \"workflow_id\": wf_id\n",
    "}\n",
    "df = query_api.df_query(_filter, calculate_telemetry_diff=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "d2c04cbe-4b78-49ee-b74d-5e7680a4478f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df.head(3)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "67159316-b97b-4a99-ac22-051ee50a6117",
   "metadata": {},
   "source": [
    "## Cleaning DataFrame"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "8a8c1dd7-9647-4e7a-82e3-f7db7752f824",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "cleaned_df = analytics.clean_dataframe(\n",
    "                        df,\n",
    "                        keep_non_numeric_columns=False,\n",
    "                        keep_only_nans_columns=False,\n",
    "                        keep_task_id=False,\n",
    "                        keep_telemetry_percent_columns=False,\n",
    "                        sum_lists=True,\n",
    "                        aggregate_telemetry=True)\n",
    "cleaned_df.head()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5497b4c8-ba90-4ae4-82d7-0ef821fe2f4f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "sort = [\n",
    "    (\"generated.loss\", TaskQueryAPI.ASC),\n",
    "    (\"generated.responsible_ai_metadata.params\", TaskQueryAPI.ASC),\n",
    "]\n",
    "df = query_api.df_get_top_k_tasks(\n",
    "    filter=_filter,\n",
    "    calculate_telemetry_diff=False,\n",
    "    sort=sort,\n",
    "    k=3,\n",
    ")\n",
    "df.filter(regex='used[.]|generated[.]')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e9df7dfb-72b7-4d77-8447-af73f8314cd4",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Query Returning the Top K tasks using quantile thresholds\n",
    "\n",
    "This query filters values based on quantiles (list only ocurrences with cpu_times < 50% quantile, i.e., median) then sort by cpu, loss, and flops."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c669ac40-60b4-49e0-ae62-a2cda2c5815a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "clauses = [\n",
    "    (\"telemetry_diff.process.cpu_times.user\", \"<\", 0.5),\n",
    "]\n",
    "sort = [\n",
    "    (\"telemetry_diff.process.cpu_times.user\", TaskQueryAPI.ASC),\n",
    "    (\"generated.loss\", TaskQueryAPI.ASC),\n",
    "    (\"generated.responsible_ai_metadata.flops\", TaskQueryAPI.ASC),\n",
    "]\n",
    "df = query_api.df_get_tasks_quantiles(\n",
    "    clauses=clauses,\n",
    "    filter=_filter,\n",
    "    sort=sort,\n",
    "    calculate_telemetry_diff=True,\n",
    "    clean_dataframe=True,\n",
    ")\n",
    "df"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "1be66d76-27f5-48d3-a10f-92f2326eb167",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Correlation Analysis"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "07d8c444-ca2c-4249-8d6c-ecdb7ea786fb",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### Using Pandas' correlation "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "066717e4-2110-4d62-aedd-005c1198cefa",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df.corr()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "00792ce5-7c72-443c-91f9-61c926444fc8",
   "metadata": {
    "tags": []
   },
   "source": [
    "#### Using FlowCept's functions for correlations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e03dab0b-1a03-46a7-bbb2-4d16339abfe1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = query_api.df_query(_filter, calculate_telemetry_diff=True)\n",
    "df = analytics.clean_dataframe(df, aggregate_telemetry=True, sum_lists=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "9a322160-7b79-430f-af55-387a8c1c2969",
   "metadata": {
    "tags": []
   },
   "source": [
    "##### All correlations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "89923e60-b251-45aa-8723-d42c548f8ea1",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "analytics.analyze_correlations(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "ba306bb3-b4ed-4ec8-a8e2-75e04ee9c681",
   "metadata": {
    "tags": []
   },
   "source": [
    "##### Only correlations >= 0.9 (absolute) and using a different method"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "fe702218-c80c-4e78-a642-8a91b5571b1d",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "analytics.analyze_correlations(df, method='spearman', threshold=0.9)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f68a5db8-fcbe-4762-83fe-255a19a3ccc8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "analytics.analyze_correlations_between(df, col_pattern1=\"generated.\", col_pattern2=\"used.\", threshold=0.5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "43601683-a091-412a-bbc6-e66f78546fc9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "analytics.analyze_correlations_used_vs_generated(df, threshold=0.8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "01ea1a46-7fe7-4334-b546-b23943af98e4",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "analytics.analyze_correlations_used_vs_telemetry_diff(df, threshold=0.8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "5b3c1356-6209-4d92-b87f-d84f00b20041",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "analytics.analyze_correlations_generated_vs_telemetry_diff(df, threshold=0.8)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bae58142-3f70-4df8-a57a-4f75eef0cca8",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "analytics.describe_col(df, col='generated.loss')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "f9a5bdd4-b5ed-441d-9b96-bdd57fe89bd6",
   "metadata": {},
   "outputs": [],
   "source": [
    "analytics.describe_cols(df, cols=['generated.loss','generated.responsible_ai_metadata.params'], col_labels=['Loss', '#Params'])"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "e79ddb7e-d5c4-4315-8a08-cf2787f12b89",
   "metadata": {
    "tags": []
   },
   "source": [
    "## Plots"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "c369915d-b12d-4bf7-b0f4-02e5b5be8a9b",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "_filter = {\n",
    "    \"workflow_id\": wf_id\n",
    "}\n",
    "df = query_api.df_query(_filter, calculate_telemetry_diff=True, clean_dataframe=True, sum_lists=True, aggregate_telemetry=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "e765a93d-a005-4d77-9d42-4acde84bb72a",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "flow_plot.heatmap(df)"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "07bbeb1b-c2ff-4d8c-8ef8-f6fc475c967d",
   "metadata": {},
   "source": [
    "## Plotting relevant 'candidates' and comparing it with the `query_api.df_get_tasks_quantiles` function. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "4613ace3-ab5a-4553-9629-ac94daa30c0f",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df.to_csv('sample_data.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6bb073fa-6e17-403e-8c9f-3884b86119f5",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "x_col = 'generated.loss'\n",
    "y_col = 'telemetry_diff.cpu.times_avg.user'\n",
    "color_col = 'generated.responsible_ai_metadata.params'\n",
    "flow_plot.scatter2d_with_colors(df,\n",
    "                                x_col='generated.loss',\n",
    "                                y_col='telemetry_diff.cpu.times_avg.user',\n",
    "                                color_col='generated.responsible_ai_metadata.params',\n",
    "                                x_label='Loss',\n",
    "                                y_label='User CPU',                                \n",
    "                                color_label='#Params',\n",
    "                                xaxis_title='Loss',\n",
    "                                yaxis_title='User CPU',\n",
    "                                plot_horizon_line=True,\n",
    "                                horizon_quantile=0.5,\n",
    "                                plot_pareto=False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "cf639f68-00e7-4f1f-924e-e22f08c61dd9",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "clauses = [\n",
    "    (y_col, \"<=\", 0.5),\n",
    "]\n",
    "sort = [\n",
    "    (y_col, TaskQueryAPI.ASC),\n",
    "    (x_col, TaskQueryAPI.ASC),\n",
    "    (color_col, TaskQueryAPI.ASC),\n",
    "]\n",
    "df = query_api.df_get_tasks_quantiles(\n",
    "    clauses=clauses,\n",
    "    filter=_filter,\n",
    "    sort=sort,\n",
    "    calculate_telemetry_diff=True,\n",
    ")\n",
    "df[['task_id', x_col, y_col, color_col]]"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "b3b52f74-10fa-45bb-bf4d-dad73824d2db",
   "metadata": {
    "tags": []
   },
   "source": [
    "### Show everything we captured about that 'good' candidate, highlighted in the pareto front blue dot in the plot above."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6ff83981-3a4c-4d26-a3ca-5a9c3649917c",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "df.query(f\"task_id == '{df.head(1)['task_id'].values[0]}'\") "
   ]
  },
  {
   "cell_type": "markdown",
   "id": "f3d3b02c-5c73-484b-a1c1-1961c812bc24",
   "metadata": {},
   "source": [
    "### Find Interesting Tasks with data that are sensitve according to correlations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "b2ea6476-2eb3-4990-bf85-4858901c4422",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "result = query_api.find_interesting_tasks_based_on_correlations_generated_and_telemetry_data(filter=_filter)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "62597c13-f075-45e2-a0bb-9f224bdfa20e",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "result.items()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "130eb28b-6c37-437e-8397-3f3471bc93ac",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "# task_id, res = next(iter(result.items()))\n",
    "# res"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "2db5fa87-3fd0-4b2d-85b5-35a670e0756a",
   "metadata": {},
   "source": [
    "### Finding Tasks with Outlier Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "6f827807-2c08-4112-b9e2-789617e8b2c3",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "tasks_with_outliers = query_api.df_find_outliers(\n",
    "    outlier_threshold=5,\n",
    "    calculate_telemetry_diff=True,\n",
    "    filter=_filter,\n",
    "    clean_dataframe=True,\n",
    "    keep_task_id=True\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "bb1d501b-5322-4c54-a72a-46c8e82f51eb",
   "metadata": {
    "tags": []
   },
   "outputs": [],
   "source": [
    "selected_columns = set(tasks_with_outliers['outlier_columns'].explode())\n",
    "selected_columns.add(\"task_id\")\n",
    "selected_columns.add(\"outlier_columns\")\n",
    "result_df = tasks_with_outliers.loc[:, tasks_with_outliers.columns.isin(selected_columns)]\n",
    "result_df"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.15"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}