notebooks/generator.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Nuclio - Generator function"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Environment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: ignore\n",
    "import nuclio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "%nuclio: setting kind to 'nuclio'\n"
     ]
    }
   ],
   "source": [
    "%nuclio config kind = \"nuclio\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Configurations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Setups\n",
    "> Please make sure all the packages in the following `nuclio cmd` cell are installed, if not, please remove the `-c` flag and run the cell"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%nuclio cmd -c\n",
    " \n",
    "# Utils\n",
    "pip install pyarrow\n",
    "pip install pandas\n",
    "pip install pytimeparse\n",
    "\n",
    "# Igz DB\n",
    "pip install v3io_frames --upgrade\n",
    "\n",
    "# Function\n",
    "pip install -i https://test.pypi.org/simple/ v3io-generator\n",
    "pip install faker"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Show location to copy to `%nuclio env` configuration"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "%nuclio: setting 'SAVE_DEPLOYMENT' environment variable\n",
      "%nuclio: setting 'DEPLOYMENT_TABLE' environment variable\n",
      "%nuclio: setting 'METRICS_CONFIGURATION_FILEPATH' environment variable\n",
      "%nuclio: setting 'SAVE_TO' environment variable\n",
      "%nuclio: setting '# SAVE_TO' environment variable\n",
      "%nuclio: setting 'SECS_TO_GENERATE' environment variable\n",
      "%nuclio: setting 'SAVE_TO_TSDB' environment variable\n"
     ]
    },
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "%nuclio: cannot find \"=\" in line\n",
      "%nuclio: cannot find \"=\" in line\n",
      "%nuclio: cannot find \"=\" in line\n",
      "%nuclio: cannot find \"=\" in line\n"
     ]
    }
   ],
   "source": [
    "%%nuclio env\n",
    "\n",
    "# Deployment\n",
    "SAVE_DEPLOYMENT=1\n",
    "DEPLOYMENT_TABLE=devices\n",
    "\n",
    "# Metrics\n",
    "METRICS_CONFIGURATION_FILEPATH = /User/demo-network-operations/src/metric_configurations.yaml\n",
    "\n",
    "# Parquet\n",
    "SAVE_TO=/User/demo-network-operations/streaming/metrics\n",
    "# SAVE_TO=metrics\n",
    "\n",
    "SECS_TO_GENERATE=3600\n",
    "\n",
    "# Save as\n",
    "SAVE_TO_TSDB=0"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: start-code"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "import time\n",
    "import yaml\n",
    "import pandas as pd\n",
    "import itertools\n",
    "import datetime\n",
    "\n",
    "# DB Connection\n",
    "import v3io_frames as v3f\n",
    "\n",
    "# Data generator\n",
    "from v3io_generator import metrics_generator, deployment_generator"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Helper functions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _create_deployment():\n",
    "    print('creating deployment')\n",
    "    # Create meta-data factory\n",
    "    dep_gen = deployment_generator.deployment_generator()\n",
    "    faker=dep_gen.get_faker()\n",
    "\n",
    "    # Design meta-data\n",
    "    dep_gen.add_level(name='company',number=2,level_type=faker.company)\n",
    "    dep_gen.add_level('data_center',number=2,level_type=faker.street_name)\n",
    "    dep_gen.add_level('device',number=2,level_type=faker.msisdn)\n",
    "\n",
    "    # Create meta-data\n",
    "    deployment_df = dep_gen.generate_deployment()\n",
    "    return deployment_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _is_deployment_exist(path):\n",
    "    # Checking shared path for the devices table\n",
    "    return os.path.exists(f'/v3io/bigdata/{path}')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _get_deployment_from_kv(client, path):\n",
    "    print(f'Retrieving deployment from {path}')\n",
    "    context.logger.debug(f'Retrieving deployment from {path}')\n",
    "    # Read the devices table from our KV store\n",
    "    deployment_df = client.read(backend='kv', table=path)\n",
    "    \n",
    "    # Reset index to column\n",
    "    deployment_df.index.name = 'device'\n",
    "    deployment_df = deployment_df.reset_index()\n",
    "    return deployment_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "def _save_deployment_to_kv(path, df, client=v3f.Client('framesd:8081')):\n",
    "    # Save deployment to our KV store\n",
    "    client.write(backend='kv', table='netops_devices',dfs=df, index_cols=['device'])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "def get_or_create_deployment(path, save_to_cloud=False, client=None):\n",
    "    if client and _is_deployment_exist(path):\n",
    "        # Get deployment from KV\n",
    "        deployment_df = _get_deployment_from_kv(client, path)\n",
    "    else:\n",
    "        # Create deployment\n",
    "        deployment_df = _create_deployment()\n",
    "        \n",
    "        if client and save_to_cloud:\n",
    "            _save_deployment_to_kv(path, deployment_df, client)\n",
    "\n",
    "    return deployment_df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "def set_indexes(df):\n",
    "    df = df.set_index(['timestamp', 'company', 'data_center', 'device'])\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_metrics_to_tsdb(context, metrics: pd.DataFrame):\n",
    "    print('Saving metrics to TSDB')\n",
    "    \n",
    "    context.v3f.write('tsdb', context.metrics_table, metrics)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [],
   "source": [
    "def save_metrics_to_parquet(context, metrics):\n",
    "    print('Saving metrics to Parquet')\n",
    "    df = pd.concat(itertools.chain(metrics))\n",
    "    \n",
    "    # Need to fix timestamps from ns to ms if we write to parquet\n",
    "    df = df.reset_index()\n",
    "    df['timestamp'] = df.loc[:, 'timestamp'].astype('datetime64[ms]')\n",
    "    \n",
    "    # Fix indexes\n",
    "    df = set_indexes(df)\n",
    "    \n",
    "    # Save parquet\n",
    "    first_timestamp = df.index[0][0].strftime('%Y%m%dT%H%M%S')\n",
    "    last_timestamp = df.index[-1][0].strftime('%Y%m%dT%H%M%S')\n",
    "    filename = first_timestamp + '-' + last_timestamp + '.parquet'\n",
    "    print(filename)\n",
    "    filepath = os.path.join(context.metrics_table, filename)\n",
    "    print(filepath)\n",
    "    with open(filepath, 'wb+') as f:\n",
    "        df.to_parquet(f)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "def is_deployment_initialized(context):\n",
    "    return hasattr(context, 'metric_generator')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Init context"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "def init_context(context):\n",
    "    \n",
    "    # Get saving configuration\n",
    "    save_to_tsdb = (int(os.getenv('SAVE_TO_TSDB', 1)) == 1)\n",
    "    \n",
    "    # Set metrics table\n",
    "    metrics_table = os.getenv('SAVE_TO', 'netops_metrics')\n",
    "    setattr(context, 'metrics_table', metrics_table) \n",
    "\n",
    "    # TSDB Based demo\n",
    "    if save_to_tsdb:\n",
    "        context.logger.debug('Saving to TSDB')\n",
    "        # Create our DB client\n",
    "        client = v3f.Client(address='framesd:8081', container='bigdata')\n",
    "        \n",
    "        # Create TSDB table if needed\n",
    "        client.create('tsdb', metrics_table, rate='1/s', if_exists=1)\n",
    "        \n",
    "        # Set saving function\n",
    "        setattr(context, 'write', save_metrics_to_tsdb)\n",
    "    \n",
    "    # Parquet based demo\n",
    "    else:\n",
    "        context.logger.debug('Saving to Parquet')\n",
    "        # Set empty client for verification purposes\n",
    "        client = None\n",
    "          \n",
    "        # Create saving directory\n",
    "        filepath = os.path.join(metrics_table)\n",
    "        if not os.path.exists(filepath):\n",
    "            os.makedirs(filepath)\n",
    "        \n",
    "        # Set saving function\n",
    "        setattr(context, 'write', save_metrics_to_parquet)\n",
    "    \n",
    "          \n",
    "    # Set batch endtime\n",
    "    secs_to_generate = os.getenv('SECS_TO_GENERATE', 10)\n",
    "    setattr(context, 'secs_to_generate', secs_to_generate)\n",
    "    \n",
    "     \n",
    "    \n",
    "    # Generate or create deployment\n",
    "    deployment_df = get_or_create_deployment(os.environ['DEPLOYMENT_TABLE'], os.environ['SAVE_DEPLOYMENT'], client)\n",
    "    # Convert to log_dataset\n",
    "    \n",
    "    deployment_df['cpu_utilization'] = 70\n",
    "    deployment_df['latency'] = 0\n",
    "    deployment_df['packet_loss'] = 0\n",
    "    deployment_df['throughput'] = 290\n",
    "    deployment_df.head()\n",
    "    \n",
    "    # Get metrics configuration\n",
    "    # Move to get-object from store:///\n",
    "    with open(os.getenv('METRICS_CONFIGURATION_FILEPATH', '/configurations/metrics_configuration.yaml'), 'r') as f:\n",
    "        metrics_configuration = yaml.load(f)\n",
    "        \n",
    "    # Create metrics generator\n",
    "    initial_timestamp = int(os.getenv('initial_timestamp', (datetime.datetime.now()-datetime.timedelta(days=1)).timestamp()))\n",
    "    met_gen = metrics_generator.Generator_df(metrics_configuration, \n",
    "                                             user_hierarchy=deployment_df, \n",
    "                                             initial_timestamp=initial_timestamp)\n",
    "    setattr(context, 'metric_generator', met_gen)\n",
    "    \n",
    "    # Set client\n",
    "    setattr(context, 'v3f', client)\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Handler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "def handler(context, event):\n",
    "       \n",
    "    # Create metrics generator based on YAML configuration and deployment\n",
    "    metrics = context.metric_generator.generate_range(start_time=datetime.datetime.now(),\n",
    "                                     end_time=datetime.datetime.now()+datetime.timedelta(seconds=int(context.secs_to_generate)),\n",
    "                                     as_df=True,\n",
    "                                     as_iterator=True)\n",
    "    \n",
    "    # Save Generated metrics\n",
    "    context.write(context, metrics)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: end-code"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Generate data configuration file"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Overwriting ../src/metric_configurations.yaml\n"
     ]
    }
   ],
   "source": [
    "%%writefile ../src/metric_configurations.yaml\n",
    "errors: {length_in_ticks: 50, rate_in_ticks: 150}\n",
    "timestamps: {interval: 5s, stochastic_interval: true}\n",
    "metrics:\n",
    "  cpu_utilization:\n",
    "    accuracy: 2\n",
    "    distribution: normal\n",
    "    distribution_params: {mu: 70, noise: 0, sigma: 10}\n",
    "    is_threshold_below: true\n",
    "    past_based_value: false\n",
    "    produce_max: false\n",
    "    produce_min: false\n",
    "    validation:\n",
    "      distribution: {max: 1, min: -1, validate: false}\n",
    "      metric: {max: 100, min: 0, validate: true}\n",
    "  latency:\n",
    "    accuracy: 2\n",
    "    distribution: normal\n",
    "    distribution_params: {mu: 0, noise: 0, sigma: 5}\n",
    "    is_threshold_below: true\n",
    "    past_based_value: false\n",
    "    produce_max: false\n",
    "    produce_min: false\n",
    "    validation:\n",
    "      distribution: {max: 1, min: -1, validate: false}\n",
    "      metric: {max: 100, min: 0, validate: true}\n",
    "  packet_loss:\n",
    "    accuracy: 0\n",
    "    distribution: normal\n",
    "    distribution_params: {mu: 0, noise: 0, sigma: 2}\n",
    "    is_threshold_below: true\n",
    "    past_based_value: false\n",
    "    produce_max: false\n",
    "    produce_min: false\n",
    "    validation:\n",
    "      distribution: {max: 1, min: -1, validate: false}\n",
    "      metric: {max: 50, min: 0, validate: true}\n",
    "  throughput:\n",
    "    accuracy: 2\n",
    "    distribution: normal\n",
    "    distribution_params: {mu: 250, noise: 0, sigma: 20}\n",
    "    is_threshold_below: false\n",
    "    past_based_value: false\n",
    "    produce_max: false\n",
    "    produce_min: false\n",
    "    validation:\n",
    "      distribution: {max: 1, min: -1, validate: false}\n",
    "      metric: {max: 300, min: 0, validate: true}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Generate dataset locally"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# nuclio: ignore\n",
    "init_context(context)\n",
    "event = nuclio.Event(body='')\n",
    "output = handler(context, event)\n",
    "output"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Deploy to cluster\n",
    "(For streaming demo)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from mlrun import code_to_function, mount_v3io\n",
    "\n",
    "fn = code_to_function(name='nuclio-generator',\n",
    "                      kind='nuclio', with_doc=False)\n",
    "fn.spec.base_spec['spec']['build']['baseImage'] = 'mlrun/ml-models'\n",
    "fn.apply(mount_v3io())\n",
    "fn.add_trigger('cron', nuclio.triggers.CronTrigger(interval='1m'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fn.save()\n",
    "fn.export('../src/generator.yaml')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "fn.deploy()"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "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.6.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}